diff --git a/.github/scripts/uk_replay_drain.sh b/.github/scripts/uk_replay_drain.sh new file mode 100755 index 0000000..18ec51a --- /dev/null +++ b/.github/scripts/uk_replay_drain.sh @@ -0,0 +1,65 @@ +#!/usr/bin/env bash +# Drain the UK replay qualification queue (see .github/workflows/uk-replay.yml). +# +# For each pending request, oldest first: mark it running, run it end to end +# (pipeline/uk_replay_qualify.py run-request), commit the receipt and results +# to the state branch, then write the terminal STATUS.json naming the +# receipt's digest and that commit. Stops when the queue is empty or the time +# budget is spent; the next scheduled run continues. +set -euo pipefail + +STATE=state +CONTROL=(uv run --with rfc8785==0.1.4 --with jsonschema --with huggingface_hub python) +BUDGET_SECONDS=$((300 * 60)) +started=$(date +%s) + +now() { date -u +%Y-%m-%dT%H:%M:%SZ; } + +push_state() { + git -C "$STATE" add -A + if git -C "$STATE" diff --cached --quiet; then + return 0 + fi + git -C "$STATE" commit -qm "$1" + for attempt in 1 2 3 4 5; do + if git -C "$STATE" push -q origin "HEAD:$STATE_BRANCH"; then + return 0 + fi + git -C "$STATE" pull -q --rebase origin "$STATE_BRANCH" + done + echo "could not push the state branch after five attempts" >&2 + return 1 +} + +while :; do + git -C "$STATE" pull -q --rebase origin "$STATE_BRANCH" + request_id=$("${CONTROL[@]}" -m pipeline.uk_replay_queue pending --state "$STATE" --now "$(now)" | head -n 1) + if [ -z "$request_id" ]; then + echo "queue empty" + break + fi + if [ $(( $(date +%s) - started )) -gt "$BUDGET_SECONDS" ]; then + echo "time budget spent; the next scheduled run continues" + break + fi + echo "::group::request $request_id" + "${CONTROL[@]}" -m pipeline.uk_replay_queue set-status --state "$STATE" \ + --request-id "$request_id" --status running --run-url "$RUN_URL" --now "$(now)" + push_state "Run $request_id" + + work="$RUNNER_TEMP/requests/$request_id" + outcome=$("${CONTROL[@]}" -m pipeline.uk_replay_qualify run-request \ + --request "$STATE/requests/$request_id/REQUEST.json" \ + --state-dir "$STATE" --work "$work" --runner "$REPLAY_RUNNER" | tail -n 1) + status=${outcome%% *} + receipt_sha256=${outcome##* } + push_state "Receipt for $request_id: $status" + receipt_commit=$(git -C "$STATE" rev-parse HEAD) + + "${CONTROL[@]}" -m pipeline.uk_replay_queue set-status --state "$STATE" \ + --request-id "$request_id" --status "$status" --run-url "$RUN_URL" --now "$(now)" \ + --receipt-sha256 "$receipt_sha256" --receipt-commit "$receipt_commit" + push_state "Status for $request_id: $status" + rm -rf "$work" + echo "::endgroup::" +done diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fd5132e..2c35025 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,6 +44,41 @@ jobs: - run: uv run --with pytest --with hypothesis --with pypdf --with pyyaml --with numpy --with psutil --with rfc8785==0.1.4 --with jsonschema pytest tests/ -q env: PYTHONPATH: . + # REPLAY NAMESPACES: every committed event replay (the legacy + # results/uk/events and each results/uk/replays/) is re-staged + # and re-compared from its committed artifacts, every committed release + # diff is re-rendered, and every sealed namespace must match its + # receipt byte for byte. Engine-free; the no-drift gate below fails on + # any regenerated byte that differs. + - name: Replay namespaces regenerate and seals hold + env: + PYTHONPATH: . + run: | + set -euo pipefail + run() { uv run --with rfc8785==0.1.4 --with jsonschema python "$@"; } + for root in results/uk/events results/uk/replays/*/; do + [ -d "$root" ] || continue + case "$root" in */diffs/) continue ;; esac + found=0 + for manifest in "$root"/*/RUN_MANIFEST.json; do + [ -f "$manifest" ] || continue + found=1 + dir=$(dirname "$manifest"); event=$(basename "$dir") + run pipeline/stage_uk_event.py --event "$event" --artifact-dir "$dir" > /dev/null + run pipeline/compare_uk_event.py --event "$event" --staged "$dir/STAGED.jsonl" --output-dir "$dir" > /dev/null + done + if [ "$found" = 1 ]; then + run pipeline/compare_uk_event.py --summary --events-root "$root" > /dev/null + echo "regenerated $root" + fi + done + for diff in results/uk/replays/diffs/*/; do + [ -f "$diff/DIFF.json" ] || continue + base=$(run -c "import json,sys; print(json.load(open(sys.argv[1]))['base']['root'])" "$diff/DIFF.json") + head=$(run -c "import json,sys; print(json.load(open(sys.argv[1]))['head']['root'])" "$diff/DIFF.json") + run -m pipeline.diff_uk_event_runs --base "$base" --head "$head" --output-dir "$diff" --check + done + run -m pipeline.uk_replay_qualify verify-sealed - run: git diff --exit-code - name: Upload built database uses: actions/upload-artifact@v4 @@ -52,6 +87,80 @@ jobs: path: data/scorecard.db retention-days: 30 + replay-engine: + # The event replay against each certified variant's own engine: the + # exact environment from the variant's requirements freeze, the + # construction guards, and the committed premise and diagnostic files + # regenerated byte for byte. With the HUGGING_FACE_TOKEN secret it also + # replays one real measure-year on the variant's certified dataset and + # compares the artifact with the committed bytes (replay determinism); + # without the secret (forks) that step announces it did not run. + name: replay engine · ${{ matrix.variant }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + variant: + - pe5.0.2_uk2.89.2_populace-uk-2023-dd68c73-4aa4b14-20260619T023711Z + - pe6.2.6_uk2.125.1_policyengine-uk-data-1.58.0 + env: + PYTHONPATH: . + VARIANT: ${{ matrix.variant }} + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + - name: Install the variant's exact environment + run: | + set -euo pipefail + field() { python3 -c "import json,sys; d=json.load(open('data/uk/replay_variants.json')); v=[v for v in d['variants'] if v['variant_id']==sys.argv[1]][0]; print(eval(sys.argv[2], {'v': v}))" "$VARIANT" "$1"; } + echo "REQUIREMENTS=$(field "v['requirements']['path']")" >> "$GITHUB_ENV" + echo "ENGINE=$(field "v['identity']['packages']['policyengine-uk']['version']")" >> "$GITHUB_ENV" + echo "RESULT_ROOT=$(field "v['result_root']")" >> "$GITHUB_ENV" + echo "LEGACY=$(field "v['layout'] == 'legacy_events'")" >> "$GITHUB_ENV" + - run: | + uv venv --python 3.12 .venv-variant + uv pip install --python .venv-variant/bin/python -r "$REQUIREMENTS" + uv pip install --python .venv-variant/bin/python rfc8785==0.1.4 jsonschema + - name: The installed environment is exactly the variant's + run: .venv-variant/bin/python -c "from pipeline import uk_replay_variants as rv; rv.verify_installed(rv.get_variant('$VARIANT')); print('installed environment matches $VARIANT')" + - name: Construction guards on the variant's engine + run: | + set -euo pipefail + for event in autumn_budget_2024 autumn_statement_2023 spring_budget_2024 spring_statement_2025 spring_budget_2023; do + .venv-variant/bin/python -m pipeline.compute_uk_event --event "$event" --variant "$VARIANT" --guards > /dev/null + echo "guards hold: $event" + done + - name: Construction premises regenerate + run: .venv-variant/bin/python -m pipeline.audit_uk_event_premises --all --expect-engine "$ENGINE" --output "$RESULT_ROOT/CONSTRUCTION_PREMISES.json" --check + - name: Re-verified diagnostics regenerate + if: env.LEGACY == 'False' + run: | + set -euo pipefail + .venv-variant/bin/python pipeline/diagnose_uk_event_models.py --reverify results/uk/events/MODEL_DIAGNOSTICS.json --expect-engine "$ENGINE" --output "$RUNNER_TEMP/MODEL_DIAGNOSTICS.json" + cmp "$RUNNER_TEMP/MODEL_DIAGNOSTICS.json" "$RESULT_ROOT/MODEL_DIAGNOSTICS.json" + - name: One real measure-year replays byte for byte + if: env.LEGACY == 'False' + env: + HUGGING_FACE_TOKEN: ${{ secrets.HUGGING_FACE_TOKEN }} + run: | + set -euo pipefail + if [ -z "${HUGGING_FACE_TOKEN:-}" ]; then + echo "::notice::HUGGING_FACE_TOKEN not available (fork PR or unset secret): the real replay determinism check did NOT run" + exit 0 + fi + uv run --with huggingface_hub python - <<'PY' + import json, os + from huggingface_hub import hf_hub_download + d = json.load(open("data/uk/replay_variants.json")) + v = [v for v in d["variants"] if v["variant_id"] == os.environ["VARIANT"]][0]["identity"]["dataset"] + print(hf_hub_download(repo_id=v["repo_id"], repo_type=v["repo_type"], filename=v["path"], revision=v["revision"], token=os.environ["HUGGING_FACE_TOKEN"])) + PY + key=autumn_budget_2024__private_school_vat_20pct + .venv-variant/bin/python -m pipeline.compute_uk_event --event autumn_budget_2024 --variant "$VARIANT" \ + --measures "$key" --years 2026 --workers 1 --output-dir "$RUNNER_TEMP/determinism" + cmp "$RUNNER_TEMP/determinism/${key}_2026.json" "$RESULT_ROOT/autumn_budget_2024/${key}_2026.json" + echo "replayed $key 2026 byte-identical to the committed artifact" + app: name: typecheck · build runs-on: ubuntu-latest diff --git a/.github/workflows/uk-replay.yml b/.github/workflows/uk-replay.yml new file mode 100644 index 0000000..10b77d9 --- /dev/null +++ b/.github/workflows/uk-replay.yml @@ -0,0 +1,127 @@ +name: UK replay qualification + +# WHY THIS EXISTS +# +# Release qualification in policyengine.py (W4) asks this repository to rerun +# the OBR fiscal-event replay corpus on a candidate model/data pair before it +# is promoted, and to return an artifact-bound receipt. The interface is +# docs/uk_replay/QUALIFICATION_INTERFACE.md; the run is +# pipeline/uk_replay_qualify.py. +# +# GitHub keeps at most one pending run per concurrency group and cancels the +# older pending ones, so a request is first written to the persistent queue +# on the `replay-state` branch (enqueue job, no concurrency group). The drain +# job then runs queued requests one at a time. A scheduled run drains anything +# a cancelled run left behind. A request whose run died mid-way is retried +# after the stale bound in pipeline/uk_replay_queue.py. +# +# Simulations: one drain job at a time, at most two workers inside it, so +# never more than two simulations at once from this workflow. +# +# Secrets: HUGGING_FACE_TOKEN reads the certified datasets. Without it a +# request is rejected with data_access_failed rather than skipped. + +on: + repository_dispatch: + types: [uk-replay-qualification-request] + workflow_dispatch: + inputs: + request: + description: "A qualification request (JSON). Empty: drain the queue only." + required: false + default: "" + schedule: + - cron: "41 */3 * * *" + +permissions: + contents: write + +env: + STATE_BRANCH: replay-state + +jobs: + enqueue: + name: record the request + if: github.event_name == 'repository_dispatch' || (github.event_name == 'workflow_dispatch' && github.event.inputs.request != '') + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + - name: Check out the state branch + run: | + set -euo pipefail + if git ls-remote --exit-code --heads origin "$STATE_BRANCH" >/dev/null; then + git fetch -q origin "$STATE_BRANCH" + git worktree add -q state "origin/$STATE_BRANCH" + git -C state switch -q -C "$STATE_BRANCH" + else + git worktree add -q --orphan -b "$STATE_BRANCH" state + fi + git -C state config user.name "github-actions[bot]" + git -C state config user.email "41898282+github-actions[bot]@users.noreply.github.com" + - name: Write the request + env: + DISPATCH_REQUEST: ${{ toJSON(github.event.client_payload.request) }} + MANUAL_REQUEST: ${{ github.event.inputs.request }} + run: | + set -euo pipefail + if [ "${{ github.event_name }}" = "repository_dispatch" ]; then + printf '%s' "$DISPATCH_REQUEST" > "$RUNNER_TEMP/request.json" + else + printf '%s' "$MANUAL_REQUEST" > "$RUNNER_TEMP/request.json" + fi + - name: Enqueue and push + env: + PYTHONPATH: . + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + run: | + set -euo pipefail + now=$(date -u +%Y-%m-%dT%H:%M:%SZ) + uv run --with rfc8785==0.1.4 --with jsonschema python -m pipeline.uk_replay_queue \ + enqueue --state state --request "$RUNNER_TEMP/request.json" --run-url "$RUN_URL" --now "$now" \ + | tee "$RUNNER_TEMP/enqueued.txt" + git -C state add -A + if git -C state diff --cached --quiet; then + echo "already recorded: $(cat "$RUNNER_TEMP/enqueued.txt")" + exit 0 + fi + git -C state commit -qm "Queue $(cut -d' ' -f1 "$RUNNER_TEMP/enqueued.txt")" + for attempt in 1 2 3 4 5; do + git -C state push -q origin "HEAD:$STATE_BRANCH" && exit 0 + git -C state pull -q --rebase origin "$STATE_BRANCH" + done + exit 1 + + drain: + name: run queued requests + needs: [enqueue] + if: always() && (needs.enqueue.result == 'success' || needs.enqueue.result == 'skipped') + runs-on: ubuntu-latest + timeout-minutes: 340 + concurrency: + group: uk-replay-drain + cancel-in-progress: false + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + - name: Check out the state branch + run: | + set -euo pipefail + if ! git ls-remote --exit-code --heads origin "$STATE_BRANCH" >/dev/null; then + echo "no state branch yet; nothing queued" + echo "EMPTY=1" >> "$GITHUB_ENV" + exit 0 + fi + git fetch -q origin "$STATE_BRANCH" + git worktree add -q state "origin/$STATE_BRANCH" + git -C state switch -q -C "$STATE_BRANCH" + git -C state config user.name "github-actions[bot]" + git -C state config user.email "41898282+github-actions[bot]@users.noreply.github.com" + - name: Drain the queue + if: env.EMPTY != '1' + env: + PYTHONPATH: . + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + HUGGING_FACE_TOKEN: ${{ secrets.HUGGING_FACE_TOKEN }} + REPLAY_RUNNER: github-actions ${{ runner.os }} ${{ runner.arch }} + run: bash .github/scripts/uk_replay_drain.sh diff --git a/data/uk/replay_variants.json b/data/uk/replay_variants.json index ab0a90a..60743ba 100644 --- a/data/uk/replay_variants.json +++ b/data/uk/replay_variants.json @@ -53,11 +53,11 @@ "packaged_bundle_manifest_sha256": "255dbb893f455d078a6bcece87399f0a22478f56bc4c6e526b758c4fd5b4db24", "packages": { "policyengine": { - "record_sha256": "5d5de97a40dbc525bc41a6347fc793ba772bd78f9e0630f2f1d382bc92575e68", + "record_sha256": "8d5eec548a5ce206a8fe59ef018542c701e26c0f927351f92339a7a57591b990", "version": "5.0.2" }, "policyengine-core": { - "record_sha256": "dd6570fe988159d89e45df66ccc43355f70abddb8a3a58d246ef38ce711e5687", + "record_sha256": "103128a1c2a3397268b65e33b83edddfa8e406a4eb64fb6f4ff2310db421c770", "version": "3.27.1" }, "policyengine-uk": { @@ -83,9 +83,10 @@ "model_package": "policyengine-uk", "model_version": "2.89.2", "policyengine_version": "5.0.2" - } + }, + "requirements_sha256": "450e64b173b66333027ac36e056e19928d23df1cbe2690e462bab09b31ddff9d" }, - "identity_sha256": "5c642304a4fff8f30074899b2bb530197920703add1b92304a4c3c3a48c5d176", + "identity_sha256": "3f1a33f31ac70a8f0beac2373b6408c8be3366128ec0d0e3c1b716de3880003c", "layout": "legacy_events", "note": "The bundle the track 1 replays were computed against. Its 1.3GB populace dataset needs Modal's 32GiB containers for full events.", "requirements": { @@ -145,12 +146,12 @@ "packaged_bundle_manifest_sha256": "8633ad814c9c85e53c7c513f69c4c7b1500a03cd2f6ef133d6e2ea8346b2db54", "packages": { "policyengine": { - "record_sha256": "072606d8f554123484caff5a1e0f12c3203b96d03f5990ebc4d80d6491441fd0", + "record_sha256": "f5990ac205b959621c56585c16ed963a5d1b07b1403adaa7242e0131fbf61f40", "version": "6.2.6", "wheel_sha256": "9ac20a2599cc1f4b50c128c2245042b105c40592e07fc25d39c99a529c9f410b" }, "policyengine-core": { - "record_sha256": "7a4e8a82c3467681971f5160428296b06e65357b8a70801f2cd631ea0943394a", + "record_sha256": "53a92f2e67982a16faeba61b8b3d145d3ab9a50d298e92c512d61166ffe5d5d4", "version": "3.33.0", "wheel_sha256": "2e32d02a8910ab683a2b7600987904b351d058fc2bd5910a8ea6641438bcd789" }, @@ -178,9 +179,10 @@ "model_package": "policyengine-uk", "model_version": "2.125.1", "policyengine_version": "6.2.6" - } + }, + "requirements_sha256": "5c333faef091e6780b2c178ac7584249ac7db0159841921727b913505577bcdb" }, - "identity_sha256": "f241374dea3ca00aeba8849437481da34e6c6c9ddb0bb9b75d3e72bbbe22fd69", + "identity_sha256": "3f62a208a9bf522cccb391fe65c1411124abaf4488ab760c522b0568f5b417aa", "layout": "replay_namespace", "note": "The newest certified pair on 11 October 2026; data built with policyengine-uk 2.122.2 and certified for 2.125.1.", "requirements": { diff --git a/docs/uk_replay/QUALIFICATION_INTERFACE.md b/docs/uk_replay/QUALIFICATION_INTERFACE.md index d616748..15b712a 100644 --- a/docs/uk_replay/QUALIFICATION_INTERFACE.md +++ b/docs/uk_replay/QUALIFICATION_INTERFACE.md @@ -102,8 +102,10 @@ Two flows use the same receiver: * `request_id` is not a field: it is the SHA-256 of the request's JCS bytes. Both sides test against the golden fixture. -* `requested_by` accepts additional provenance fields (for example - `run_attempt`); every other object is closed. +* `requested_by` names the requesting repository and commit; + `workflow_run_id` is optional (a scorecard catch-up run started outside + Actions has none), and additional provenance fields such as `run_attempt` + are accepted. Every other object is closed. * The wrapper is identified by `version` plus either a PyPI wheel (`{"pypi": {"wheel_sha256": …}}`) or, for an unreleased candidate whose regenerated manifest is not on PyPI yet, a public git commit. Scorecard diff --git a/docs/uk_replay/RECIPE.md b/docs/uk_replay/RECIPE.md index a274129..675e268 100644 --- a/docs/uk_replay/RECIPE.md +++ b/docs/uk_replay/RECIPE.md @@ -422,6 +422,13 @@ PYTHONPATH=. .venv-candidate/bin/python -m pipeline.compute_uk_event --event aut PYTHONPATH=. .venv-candidate/bin/python pipeline/stage_uk_event.py --event autumn_budget_2024 --artifact-dir results/uk/replays/pe6.2.6_uk2.125.1_policyengine-uk-data-1.58.0/autumn_budget_2024 ``` +On the shared Mac, check the host reservation before a local run: +`python3 ~/reviews/fleet-ops/tools/reservation.py allowed w6-replay` exits 1 +while a reserved run (the 29 October–2 November triple-lock rebuild) holds +heavy local work back. Qualification runs from policyengine.py go through +the receiver workflow on GitHub runners instead; see +[RELEASE_REPLAY.md](RELEASE_REPLAY.md). + The dataset must already be local: the runner finds the certified bytes by SHA-256 in the Hugging Face cache, or takes `--dataset-file `, and copies them into `.venv-replay-checks/runtime//data/`. Results diff --git a/docs/uk_replay/RELEASE_REPLAY.md b/docs/uk_replay/RELEASE_REPLAY.md index 769cdcb..7695de4 100644 --- a/docs/uk_replay/RELEASE_REPLAY.md +++ b/docs/uk_replay/RELEASE_REPLAY.md @@ -81,14 +81,15 @@ A variant record: | Field | Meaning | |---|---| -| `variant_id` | `pe_uk_`; must equal the slug recomputed from `identity`. | +| `variant_id` | `pe_uk_`, recomputed from `identity`; when another identity already holds that slug (same versions, other wheels or numerical environment) the id adds `--`. | | `identity` | Everything that determines execution. Its canonical-JSON SHA-256 is `identity_sha256`. | -| `identity.packages` | For `policyengine`, `policyengine-uk`, `policyengine-core`: exact `version`, and `record_sha256`: the SHA-256 of the distribution's own `RECORD` entries, sorted, excluding installer bookkeeping (`INSTALLER`, `REQUESTED`, `RECORD`, `direct_url.json`) and bytecode. Every listed file is checked on disk against its recorded hash when the identity is read, so the digest binds the wheel's installed contents, and pip and uv installs of one wheel agree. Optional `wheel_sha256` when the wheel itself was hashed. | +| `identity.packages` | For `policyengine`, `policyengine-uk`, `policyengine-core`: exact `version`, and `record_sha256`: the SHA-256 of the distribution's own `RECORD` entries, sorted, excluding installer bookkeeping (`INSTALLER`, `REQUESTED`, `RECORD`, `direct_url.json`, `uv_cache.json`), bytecode, and the console-script wrappers the installer generates outside site-packages (their shebang names the environment's interpreter). Every listed file is checked on disk against its recorded hash when the identity is read, so the digest binds the wheel's installed contents, and pip and uv installs of one wheel agree. Optional `wheel_sha256` when the wheel itself was hashed. | | `identity.packaged_bundle_manifest_sha256` | SHA-256 of `policyengine/data/bundle/manifest.json` as installed. | | `identity.release_bundle` | The exact `pe.uk.uk_latest.release_bundle` dict expected at runtime under **packaged (offline) certification**. | | `identity.dataset` | `repo_id`, `repo_type`, `revision`, `path`, `sha256`, `size_bytes`, `uri`. | | `identity.materialization` | `hf_cache_mirror` (policyengine 5.x: the existing HF-cache plus `data/uk/policyengine-local` mirror) or `policyengine_data_dir` (6.x: a verified copy at `/data/`). | | `identity.python` | Interpreter minor version, e.g. `3.12`. | +| `identity.requirements_sha256` | The SHA-256 of the requirements freeze below: the numerical environment (numpy, pandas, h5py and the rest) is part of the identity, because it can move numbers. | | `requirements` | `{path, sha256}`: the exact freeze installed for this variant. Legacy: `docs/uk_replay/requirements.txt`. Others: `docs/uk_replay/requirements/.txt`. | | `data_years` | Calendar years the population supports: from the dataset's first `time_period` through the engine's extension end, as observed in `simulation.dataset.years`. | | `data_years_evidence` | The observed `time_period`, `dataset.years` and the engine file/line that sets the extension end. | @@ -115,7 +116,9 @@ failure: and `compatible_model_packages` agree with the record. 2. **Installed environment.** Installed versions of the three packages equal the record; every installed file matches its `RECORD` hash and the - normalised `RECORD` hashes to `record_sha256`; the + normalised `RECORD` hashes to `record_sha256`; every distribution pinned in + the requirements freeze is installed at its pinned version (extra + distributions, such as the receipt tools, are allowed); the installed packaged manifest hashes to `packaged_bundle_manifest_sha256`; the interpreter minor version matches. 3. **Certification path.** Before importing the engine the runner removes @@ -154,17 +157,21 @@ FY2023–24 counterparts; that loss is shown, never hidden. * Legacy results stay where they are, in `results/uk/events//`. * Every other variant writes to `results/uk/replays//`: - `VARIANT.json` (the validated record plus the runtime receipt), + `/` (the same files as a legacy event directory), `SUMMARY.md`, `SUPPORT.json`, `CONSTRUCTION_PREMISES.json`, `MODEL_DIAGNOSTICS.json`, - `/` (the same files as a legacy event directory), `SUMMARY.md`, and - finally `REPLAY_RECEIPT.json`. -* `results/uk/replays/INDEX.json` maps each variant to its root (legacy: - `results/uk/events`), status and receipt. -* A namespace with `REPLAY_RECEIPT.json` is **sealed**: the receipt binds the - SHA-256 of every file under it, the runner refuses to write there, and CI - fails if any sealed byte changes. A rerun of the same variant goes to a new - namespace suffix (`/rerun-`) and its determinism against the - sealed run is itself a check. + `REQUEST.json` (the qualification request the run answers), and finally + `NAMESPACE_MANIFEST.json` and `REPLAY_RECEIPT.json`. +* `data/uk/replay_variants.json` is the index: each record's `result_root` + names its namespace (legacy: `results/uk/events`) and its `status`. A + candidate's record and results live on the `replay-state` branch under + `requests//results/` until it is certified. +* A namespace with `REPLAY_RECEIPT.json` is **sealed**: + `NAMESPACE_MANIFEST.json` lists the SHA-256 of every file under it (and of + its release diff), the receipt binds that manifest's digest, the runner + refuses to write there, and CI fails if any sealed byte changes. A rerun of + a sealed variant writes to an explicit separate output directory, and its + bytes against the sealed run are a determinism check (CI replays one real + measure-year this way). * Release differences live in `results/uk/replays/diffs/__/`. ## Version-scoped diagnoses @@ -258,7 +265,73 @@ issues; this pipeline files none. ## Qualification requests and receipts See [QUALIFICATION_INTERFACE.md](QUALIFICATION_INTERFACE.md) for the request -W4 sends before promotion, the receipt returned, and the persistent queue. +W4 sends before promotion and the receipt returned. + +### The receiver + +`.github/workflows/uk-replay.yml` receives `repository_dispatch` events of +type `uk-replay-qualification-request` (and manual requests). Its `enqueue` +job writes the request to the persistent queue on the `replay-state` branch +before anything else (`pipeline/uk_replay_queue.py`), so GitHub cancelling +an older pending run never loses one. Its `drain` job, one at a time +(concurrency group `uk-replay-drain`), runs `.github/scripts/uk_replay_drain.sh`. +That script takes the oldest pending request, marks it running, and runs +`pipeline/uk_replay_qualify.py run-request`: + +1. Validate the request (`schema_invalid`). +2. Download the requested dataset with `HUGGING_FACE_TOKEN` into the Hugging + Face cache and check its size and SHA-256 (`data_access_failed`, + `artifact_hash_mismatch`). +3. Build the candidate environment: download the model and core wheels + without dependencies, check their SHA-256 against the request, then + install them, the wrapper (its PyPI wheel or its public commit) and the + receipt tools. Re-qualifying a certified pair installs that pair's exact + requirements freeze; a new candidate resolves within the newest certified + freeze, falling back to a free resolution only if its own pins conflict. +4. Probe inside that environment: the installed identity, the release + bundle under packaged certification (`release_bundle_mismatch`), the + certification naming this model and this dataset (`incompatible_pair`), + and the simulation's data years (`no_supported_years`). If the + environment reproduces a committed variant exactly, the run uses that + record; otherwise it builds a candidate record whose requirements are the + environment's normalised freeze. +5. Run the corpus: every event's compute, staging and comparison, the + summary, the construction premises, the re-verified diagnostics. +6. Diff against `compare_to` (default: the newest other certified variant + whose results are committed). +7. Seal the namespace (`NAMESPACE_MANIFEST.json`, then + `REPLAY_RECEIPT.json`) and copy it, the diff and the requirements to + `replay-state:requests//results/`. + +The receipt is committed first; the terminal `STATUS.json` follows in a +later commit naming the receipt's digest and that commit. A step that fails +before the candidate's identity is known yields a `failed` receipt with no +execution block; an unexpected exception also yields a `failed` receipt, so a +request never loops. A request still `running` seven hours after its last +update (a cancelled or crashed job) is retried. A scheduled run every three +hours drains whatever is left. + +At most one drain job runs, with at most two simulation workers, so this +workflow never runs more than two simulations at once. These runs use GitHub +runners; manual replays on the shared Mac also respect the host reservation +(`python3 ~/reviews/fleet-ops/tools/reservation.py allowed w6-replay`), which +holds heavy local work back from 29 October to 2 November. + +### Replay in CI + +`ci.yml` runs two kinds of replay check on every pull request and push: + +* Engine-free, in the main job: every committed namespace (legacy and each + variant) is re-staged and re-compared from its artifacts, every committed + release diff is re-rendered with `--check`, and every sealed namespace must + match its receipt byte for byte; the no-drift gate fails on any change. +* `replay engine`, once per certified variant: installs the variant's exact + freeze, checks the installed environment against the record, runs every + event's construction guards, regenerates the committed construction + premises (and, for non-legacy variants, the re-verified diagnostics) byte + for byte, and, with the `HUGGING_FACE_TOKEN` secret, replays one real + measure-year on the certified dataset and compares the artifact with the + committed bytes. Without the secret that step says it did not run. ## Invariants diff --git a/docs/uk_replay/schemas/qualification_request.v1.json b/docs/uk_replay/schemas/qualification_request.v1.json index 735b66c..16921af 100644 --- a/docs/uk_replay/schemas/qualification_request.v1.json +++ b/docs/uk_replay/schemas/qualification_request.v1.json @@ -105,7 +105,7 @@ "requested_by": { "type": "object", "additionalProperties": true, - "required": ["repository", "workflow_run_id", "commit"], + "required": ["repository", "commit"], "properties": { "repository": {"type": "string", "pattern": "^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"}, "workflow_run_id": {"type": "integer", "minimum": 1}, diff --git a/docs/uk_replay/schemas/replay_receipt.v1.json b/docs/uk_replay/schemas/replay_receipt.v1.json index c5d8d9a..f127f28 100644 --- a/docs/uk_replay/schemas/replay_receipt.v1.json +++ b/docs/uk_replay/schemas/replay_receipt.v1.json @@ -429,10 +429,7 @@ "if": { "properties": { "status": { - "enum": [ - "complete", - "failed" - ] + "const": "complete" } } }, @@ -475,6 +472,29 @@ } } } + }, + { + "if": { + "properties": { + "status": { + "const": "failed" + } + } + }, + "then": { + "properties": { + "rejection": { + "type": "null" + }, + "integrity": { + "properties": { + "status": { + "const": "failed" + } + } + } + } + } } ], "$defs": { diff --git a/pipeline/compute_uk_event.py b/pipeline/compute_uk_event.py index aab3fca..e8e4b03 100644 --- a/pipeline/compute_uk_event.py +++ b/pipeline/compute_uk_event.py @@ -599,6 +599,12 @@ def main(argv: list[str] | None = None) -> int: "--variant", help="certified replay variant (data/uk/replay_variants.json); default: the file's default", ) + ap.add_argument( + "--variants-file", + type=Path, + default=variants.VARIANTS_PATH, + help="variant records to select from (a qualification run adds its candidate)", + ) ap.add_argument( "--dataset-file", type=Path, @@ -619,7 +625,7 @@ def main(argv: list[str] | None = None) -> int: help="spawn at most two year workers; each holds one live managed simulation", ) args = ap.parse_args(argv) - variant = variants.get_variant(args.variant) + variant = variants.get_variant(args.variant, path=args.variants_file.resolve()) legacy = variants.is_legacy(variant) variant_fields = variant_identity_fields(variant) default_registry, default_output = event_paths(args.event) diff --git a/pipeline/diff_uk_event_runs.py b/pipeline/diff_uk_event_runs.py new file mode 100644 index 0000000..03b86bd --- /dev/null +++ b/pipeline/diff_uk_event_runs.py @@ -0,0 +1,1013 @@ +"""Describe how the event replay moved between two certified releases. + +Compares two complete replay namespaces (the legacy ``results/uk/events`` or +a variant's ``results/uk/replays/``) after re-verifying each +side's staging, comparison and artifact receipts. Every flag is an +investigation prompt, not a verdict: nothing here passes or fails a model. +The PE/OBR ratio bins are reported unchanged, as transitions. + + PYTHONPATH=. python -m pipeline.diff_uk_event_runs \\ + --base results/uk/events \\ + --head results/uk/replays/pe6.2.6_uk2.125.1_policyengine-uk-data-1.58.0 + +Flags (docs/uk_replay/RELEASE_REPLAY.md, "Release differences"): + +* numerical: both sides numeric and the move exceeds + max(£1m, 1% of the larger magnitude); +* structural: a sign reversal or zero crossing, a changed artifact schema or + construction orientation; +* coverage: a counterpart, artifact or source row on one side only, in a + year both sides support; +* mapping: changed head variables, channels or definitions; +* support: a year supported on one side only (never also coverage); +* inertness: an inert in-force construction on one side only; +* construction: changed executed reform dictionaries or premise checks. + A changed current-law value of a touched parameter is context, not a flag. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import re +import sys +from collections import defaultdict +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from pipeline import compare_uk_event as compare # noqa: E402 +from pipeline.compare_uk_obr_costings import atomic_write_bytes # noqa: E402 + +NUMERICAL_FLOOR_GBP = 1_000_000.0 +NUMERICAL_RELATIVE = 0.01 +# Accounting roundoff on national aggregates, not a materiality bound. +ZERO_GBP = 0.5 +FLAGS = ( + "numerical", + "structural", + "coverage", + "mapping", + "support", + "inertness", + "construction", +) +PACKET_FLAGS = frozenset(FLAGS) - {"support"} +COMPUTABLE_CLASSES = ("expressible", "partial") +LEVELS = ("source_row", "measure_year", "head") +DIFF_ROOT = ROOT / "results" / "uk" / "replays" / "diffs" +VARIANTS_PATH = ROOT / "data" / "uk" / "replay_variants.json" +URL = re.compile(r"https?://[^\s\"'<>)\]]+") +PACKET_PENDING = "to be completed by the investigating lane" + + +class DiffError(ValueError): + """A side cannot be compared: its receipts do not verify.""" + + +def canonical_bytes(value: Any) -> bytes: + return ( + json.dumps(value, indent=1, sort_keys=True, allow_nan=False) + "\n" + ).encode() + + +def sha256_bytes(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def numerical_threshold(base: float, head: float) -> float: + return max(NUMERICAL_FLOOR_GBP, NUMERICAL_RELATIVE * max(abs(base), abs(head))) + + +def numerically_moved(base: float, head: float) -> bool: + return abs(head - base) > numerical_threshold(base, head) + + +def is_zero(value: float) -> bool: + return abs(value) <= ZERO_GBP + + +def sign_details(base: float, head: float) -> list[dict]: + if not (math.isfinite(base) and math.isfinite(head)): + return [{"flag": "structural", "code": "non_finite"}] + if is_zero(base) != is_zero(head): + return [{"flag": "structural", "code": "zero_crossing"}] + if not is_zero(base) and (base > 0) != (head > 0): + return [{"flag": "structural", "code": "sign_reversal"}] + return [] + + +def numeric_details(base: float, head: float) -> list[dict]: + details = sign_details(base, head) + if numerically_moved(base, head): + details.append( + { + "flag": "numerical", + "code": "move_above_threshold", + "threshold_gbp": numerical_threshold(base, head), + } + ) + return details + + +# --- loading one side ----------------------------------------------------------- + + +def _json(path: Path) -> Any: + return json.loads(path.read_bytes()) + + +def side_identity(manifest: dict) -> dict: + variant_id = manifest.get("variant_id") or ( + f"pe{manifest['policyengine_version']}_uk{manifest['engine_version']}" + f"_{manifest['data_bundle']}" + ) + return { + "variant_id": variant_id, + "policyengine_version": manifest["policyengine_version"], + "policyengine_uk_version": manifest["engine_version"], + "policyengine_core_version": manifest["policyengine_core_version"], + "data_bundle": manifest["data_bundle"], + "certified_dataset_sha256": manifest["certified_dataset_sha256"], + } + + +def resolve_root(value: str, *, artifact_root: Path = ROOT) -> Path: + """A namespace path, or a variant id from data/uk/replay_variants.json.""" + path = Path(value) + if path.is_dir(): + return path.resolve() + variants = artifact_root / "data" / "uk" / "replay_variants.json" + if variants.exists(): + for record in _json(variants)["variants"]: + if record["variant_id"] == value: + return (artifact_root / record["result_root"]).resolve() + raise DiffError(f"{value!r} is neither a replay namespace nor a variant id") + + +def load_side(root: Path, *, artifact_root: Path = ROOT) -> dict: + """Every event of a namespace, re-verified through its receipts.""" + root = root.resolve() + events = {} + identities = set() + for event_dir in sorted(p for p in root.iterdir() if p.is_dir()): + manifest_path = event_dir / "RUN_MANIFEST.json" + if not manifest_path.exists(): + continue + event = event_dir.name + manifest = _json(manifest_path) + if manifest.get("event") != event: + raise DiffError(f"{manifest_path}: event differs from its directory") + try: + rows = compare.load_verified_comparison( + event_dir / "COMPARISON.json", artifact_root=artifact_root + ) + except (compare.EventComparisonError, FileNotFoundError, ValueError) as exc: + raise DiffError(f"{event_dir}: comparison does not verify: {exc}") from exc + artifacts = {} + for relative, digest in sorted(manifest["artifacts"].items()): + payload = (artifact_root / relative).read_bytes() + if sha256_bytes(payload) != digest: + raise DiffError(f"{relative}: artifact bytes differ from RUN_MANIFEST") + artifact = json.loads(payload) + artifacts[(artifact["measure_key"], artifact["year"])] = { + "path": relative, + "sha256": digest, + "artifact": artifact, + } + staging_path = event_dir / "STAGING_MANIFEST.json" + staging = _json(staging_path) + provenance_path = event_dir / "COMPARISON_PROVENANCE.json" + provenance = _json(provenance_path) + registry = _json(artifact_root / provenance["registry_path"]) + registry_years = sorted(registry["calendar_years"]) + supported = manifest.get("supported_years", registry_years) + identity = side_identity(manifest) + identities.add(json.dumps(identity, sort_keys=True)) + events[event] = { + "manifest": manifest, + "rows": {row["source_row_id"]: row for row in rows}, + "artifacts": artifacts, + "registry": registry, + "registry_sha256": provenance["registry_sha256"], + "registry_years": registry_years, + "supported_years": sorted(supported), + "inert_pairs": { + (row["measure_key"], row["year"]) + for row in staging.get("inert_measure_years", []) + }, + "inputs": { + "run_manifest_sha256": sha256_bytes(manifest_path.read_bytes()), + "staging_manifest_sha256": sha256_bytes(staging_path.read_bytes()), + "comparison_provenance_sha256": sha256_bytes( + provenance_path.read_bytes() + ), + }, + } + if not events: + raise DiffError(f"{root}: no replayed events") + if len(identities) != 1: + raise DiffError(f"{root}: events were produced by different executions") + premises_path = root / "CONSTRUCTION_PREMISES.json" + premises = _json(premises_path) if premises_path.exists() else None + return { + "root": root, + "identity": json.loads(identities.pop()), + "events": events, + "premises": premises, + "premises_sha256": sha256_bytes(premises_path.read_bytes()) + if premises is not None + else None, + } + + +# --- comparing ------------------------------------------------------------------ + + +def _record(event, level, key, measure_key, year, base, head, details, context): + flags = sorted({detail["flag"] for detail in details}, key=FLAGS.index) + delta = None if base is None or head is None else head - base + return { + "event": event, + "level": level, + "key": key, + "measure_key": measure_key, + "year": year, + "base_gbp": base, + "head_gbp": head, + "delta_gbp": delta, + "threshold_gbp": None if delta is None else numerical_threshold(base, head), + "flags": flags, + "details": sorted(details, key=lambda d: (FLAGS.index(d["flag"]), d["code"])), + "context": context, + } + + +def _support(event: str, year: int, base: dict | None, head: dict | None) -> list: + on_base = base is not None and year in base["supported_years"] + on_head = head is not None and year in head["supported_years"] + if base is not None and head is not None and on_base != on_head: + return [ + { + "flag": "support", + "code": "supported_on_base_only" + if on_base + else "supported_on_head_only", + } + ] + return [] + + +def compare_source_rows(event: str, base: dict | None, head: dict | None) -> list: + out = [] + ids = set(base["rows"] if base else ()) | set(head["rows"] if head else ()) + for source_row_id in sorted(ids): + b = base["rows"].get(source_row_id) if base else None + h = head["rows"].get(source_row_id) if head else None + row = b or h + year = row["year"] + # Only a row that could carry a counterpart is affected by support; + # outside-scope and not-expressible rows have none in any year. + computable = any( + r is not None and r.get("classification") in COMPUTABLE_CLASSES + for r in (b, h) + ) + details = _support(event, year, base, head) if computable else [] + bv = b["pe_value_gbp"] if b else None + hv = h["pe_value_gbp"] if h else None + supported_both = all( + side is not None and year in side["supported_years"] + for side in (base, head) + ) + if not details and (supported_both or not computable): + if b is None or h is None: + details.append( + { + "flag": "coverage", + "code": "source_row_on_base_only" + if h is None + else "source_row_on_head_only", + } + ) + else: + inert = { + side: r.get("status") == "inert_construction" + for side, r in (("base", b), ("head", h)) + } + if inert["base"] != inert["head"]: + details.append( + { + "flag": "inertness", + "code": "inert_on_base_only" + if inert["base"] + else "inert_on_head_only", + } + ) + elif (bv is None) != (hv is None): + details.append( + { + "flag": "coverage", + "code": "counterpart_on_base_only" + if hv is None + else "counterpart_on_head_only", + } + ) + if bv is not None and hv is not None: + details += numeric_details(bv, hv) + if b["head_variables"] != h["head_variables"]: + details.append( + { + "flag": "mapping", + "code": "head_variables_changed", + "base": b["head_variables"], + "head": h["head_variables"], + } + ) + out.append( + _record( + event, + "source_row", + source_row_id, + row["measure_key"], + year, + bv, + hv, + details, + { + "fy": row["fy"], + "tax_head": row.get("tax_head"), + "obr_value_gbp": row["obr_value_gbp"], + "status": { + "base": b.get("status") if b else None, + "head": h.get("status") if h else None, + }, + "ratio_bin": { + "base": b.get("ratio_bin") if b else None, + "head": h.get("ratio_bin") if h else None, + }, + }, + ) + ) + return out + + +def _premise_details(event, measure_key, year, base_premises, head_premises): + """Construction flags from premise records; current-law moves are context.""" + + def measure(premises): + return ( + ((premises or {}).get("events", {}).get(event, {})) + .get("measures", {}) + .get(measure_key) + ) + + b, h = measure(base_premises), measure(head_premises) + details, context = [], {} + if b is None or h is None: + return details, context + for name in sorted(set(b.get("checks", {})) | set(h.get("checks", {}))): + before = (b.get("checks", {}).get(name) or {}).get("holds") + after = (h.get("checks", {}).get(name) or {}).get("holds") + if before != after: + details.append( + { + "flag": "construction", + "code": "premise_check_changed", + "check": name, + "base": before, + "head": after, + } + ) + by = (b.get("years", {}) or {}).get(str(year), {}).get("paths", {}) + hy = (h.get("years", {}) or {}).get(str(year), {}).get("paths", {}) + moved = {} + for path in sorted(set(by) | set(hy)): + bp, hp = by.get(path, {}), hy.get(path, {}) + for world in ("baseline_world", "reform_world"): + # A world that is current law on both sides moved with current + # law (context below); only an executed override is construction. + overrides = any( + side.get(world) != side.get("current_law") for side in (bp, hp) + ) + if overrides and bp.get(world) != hp.get(world): + details.append( + { + "flag": "construction", + "code": f"{world}_value_changed", + "path": path, + "base": bp.get(world), + "head": hp.get(world), + } + ) + if bp.get("current_law") != hp.get("current_law"): + moved[path] = {"base": bp.get("current_law"), "head": hp.get("current_law")} + if moved: + context["current_law_changed"] = moved + return details, context + + +def compare_artifacts( + event: str, base: dict | None, head: dict | None, premises +) -> list: + out = [] + pairs = set(base["artifacts"] if base else ()) | set( + head["artifacts"] if head else () + ) + for measure_key, year in sorted(pairs): + b = ( + (base["artifacts"].get((measure_key, year)) or {}).get("artifact") + if base + else None + ) + h = ( + (head["artifacts"].get((measure_key, year)) or {}).get("artifact") + if head + else None + ) + details = _support(event, year, base, head) + context = {} + bv = b["measure_total_gbp"] if b else None + hv = h["measure_total_gbp"] if h else None + head_records = [] + if not details: + if b is None or h is None: + details.append( + { + "flag": "coverage", + "code": "artifact_on_base_only" + if h is None + else "artifact_on_head_only", + } + ) + else: + details += numeric_details(bv, hv) + if b.get("schema_version") != h.get("schema_version"): + details.append( + {"flag": "structural", "code": "artifact_schema_changed"} + ) + if b["construction"] != h["construction"]: + details.append( + { + "flag": "structural", + "code": "construction_orientation_changed", + "base": b["construction"], + "head": h["construction"], + } + ) + for field in ("head_channels", "head_definitions"): + if b.get(field) != h.get(field): + details.append({"flag": "mapping", "code": f"{field}_changed"}) + for world in ("baseline_world", "reform_world"): + if b[world].get("reform_dict") != h[world].get("reform_dict"): + details.append( + { + "flag": "construction", + "code": f"{world}_reform_dict_changed", + } + ) + inert = ( + (measure_key, year) in base["inert_pairs"], + (measure_key, year) in head["inert_pairs"], + ) + if inert[0] != inert[1]: + details.append( + { + "flag": "inertness", + "code": "inert_on_base_only" + if inert[0] + else "inert_on_head_only", + } + ) + extra, context = _premise_details( + event, measure_key, year, premises[0], premises[1] + ) + details += extra + for variable in sorted(set(b["head_effects"]) | set(h["head_effects"])): + hb = b["head_effects"].get(variable) + hh = h["head_effects"].get(variable) + variable_details = ( + numeric_details(hb, hh) + if hb is not None and hh is not None + else [{"flag": "mapping", "code": "head_on_one_side_only"}] + ) + head_records.append( + _record( + event, + "head", + variable, + measure_key, + year, + hb, + hh, + variable_details, + { + "channel": { + "base": b["head_channels"].get(variable), + "head": h["head_channels"].get(variable), + } + }, + ) + ) + out.append( + _record( + event, + "measure_year", + "total", + measure_key, + year, + bv, + hv, + details, + context, + ) + ) + out += head_records + return out + + +def bin_transitions(event_records: list[dict]) -> dict: + table = defaultdict(lambda: defaultdict(int)) + for record in event_records: + if record["level"] != "source_row": + continue + bins = record["context"]["ratio_bin"] + if bins["base"] is not None and bins["head"] is not None: + table[bins["base"]][bins["head"]] += 1 + return {base: dict(sorted(heads.items())) for base, heads in sorted(table.items())} + + +def flag_counts(records: list[dict]) -> dict: + return {flag: sum(flag in r["flags"] for r in records) for flag in FLAGS} + + +def diff_sides(base: dict, head: dict) -> dict: + records = [] + events = {} + premises = (base["premises"], head["premises"]) + for event in sorted(set(base["events"]) | set(head["events"])): + b = base["events"].get(event) + h = head["events"].get(event) + event_records = compare_source_rows(event, b, h) + compare_artifacts( + event, b, h, premises + ) + records += event_records + events[event] = { + "present": {"base": b is not None, "head": h is not None}, + "registry_changed": bool( + b and h and b["registry_sha256"] != h["registry_sha256"] + ), + "supported_years": { + "base": b["supported_years"] if b else None, + "head": h["supported_years"] if h else None, + }, + "flag_counts": flag_counts(event_records), + "bin_transitions": bin_transitions(event_records), + } + records.sort( + key=lambda r: ( + r["event"], + r["measure_key"], + r["year"], + LEVELS.index(r["level"]), + r["key"], + ) + ) + identity_changes = { + field: {"base": base["identity"][field], "head": head["identity"][field]} + for field in sorted(base["identity"]) + if base["identity"][field] != head["identity"][field] + } + + def side_header(side): + return { + "identity": side["identity"], + "root": str(side["root"].relative_to(ROOT)) + if side["root"].is_relative_to(ROOT) + else str(side["root"]), + "inputs": { + event: data["inputs"] for event, data in sorted(side["events"].items()) + }, + "premises_sha256": side["premises_sha256"], + } + + return { + "schema_version": 1, + "description": ( + "Movements of the event replay between two certified releases. " + "Flags are investigation prompts, not verdicts; ratio bins are " + "descriptive." + ), + "base": side_header(base), + "head": side_header(head), + "identity_changes": identity_changes, + "thresholds": { + "numerical_floor_gbp": NUMERICAL_FLOOR_GBP, + "numerical_relative": NUMERICAL_RELATIVE, + "zero_gbp": ZERO_GBP, + }, + "flag_counts": flag_counts(records), + "events": events, + "rows": records, + } + + +# --- packets ---------------------------------------------------------------------- + + +def _cited_sources(measure: dict | None) -> list[str]: + if measure is None: + return [] + return sorted(set(URL.findall(json.dumps(measure, ensure_ascii=False)))) + + +def _reproduce(side: dict, event: str, measure_key: str, year: int) -> str: + variant = side["identity"]["variant_id"] + return ( + f"PYTHONPATH=. python pipeline/compute_uk_event.py --event {event} " + f"--variant {variant} --measures {measure_key} --years {year} --workers 1 " + f"--output-dir .venv-replay-checks/reproductions/{variant}/{event}/{measure_key}_{year}" + ) + + +def build_packets(diff: dict, base: dict, head: dict) -> dict[str, dict]: + groups = defaultdict(list) + for record in diff["rows"]: + if PACKET_FLAGS & set(record["flags"]): + groups[(record["event"], record["measure_key"], record["year"])].append( + record + ) + packets = {} + for (event, measure_key, year), records in sorted(groups.items()): + sides = {"base": base["events"].get(event), "head": head["events"].get(event)} + registry = (sides["head"] or sides["base"])["registry"] + measure = next( + (m for m in registry["measures"] if m["measure_key"] == measure_key), None + ) + all_records = [ + r + for r in diff["rows"] + if (r["event"], r["measure_key"], r["year"]) == (event, measure_key, year) + ] + source_cells = [] + for source in (measure or {}).get("source_rows", []): + if int(source["fy"][:4]) != year: + continue + row = next( + ( + r + for r in all_records + if r["level"] == "source_row" + and r["key"] == source["source_row_id"] + ), + None, + ) + source_cells.append( + { + "source_row_id": source["source_row_id"], + "source_row_number": source.get("source_row_number"), + "source_table": source.get("source_table"), + "source_column": source.get("source_column"), + "fy": source["fy"], + "tax_head": source.get("tax_head"), + "obr_value_gbp": source["value_gbp"], + "base_gbp": row["base_gbp"] if row else None, + "head_gbp": row["head_gbp"] if row else None, + "flags": row["flags"] if row else [], + } + ) + + def artifact_ref(side): + entry = (side or {}).get("artifacts", {}).get((measure_key, year)) + if entry is None: + return None + artifact = entry["artifact"] + return { + "path": entry["path"], + "sha256": entry["sha256"], + "construction": artifact["construction"], + "baseline_reform_dict": artifact["baseline_world"].get("reform_dict"), + "reform_reform_dict": artifact["reform_world"].get("reform_dict"), + } + + def premises_for(side_premises): + measure_premises = ( + ((side_premises or {}).get("events", {}).get(event, {})) + .get("measures", {}) + .get(measure_key) + ) + if measure_premises is None: + return None + return { + "checks": measure_premises.get("checks"), + "year": measure_premises.get("years", {}).get(str(year)), + } + + name = f"{event}__{measure_key.removeprefix(event + '__')}__{year}" + packets[name] = { + "packet_schema": 1, + "event": event, + "measure_key": measure_key, + "title": (measure or {}).get("title"), + "year": year, + "fy": f"{year}-{str(year + 1)[-2:]}", + "flags": sorted({f for r in records for f in r["flags"]}, key=FLAGS.index), + "identities": {"base": base["identity"], "head": head["identity"]}, + "registry_sha256": { + side: (data or {}).get("registry_sha256") + for side, data in sides.items() + }, + "artifacts": {side: artifact_ref(data) for side, data in sides.items()}, + "records": all_records, + "source_cells": source_cells, + "premises": { + "base": premises_for(base["premises"]), + "head": premises_for(head["premises"]), + }, + "reproduce": { + side: _reproduce(data, event, measure_key, year) + for side, data in (("base", base), ("head", head)) + }, + "cited_sources": _cited_sources(measure), + "minimal_case": {"status": PACKET_PENDING}, + "statutory_evidence": {"status": PACKET_PENDING}, + "filing": ( + "This pipeline files no issues. The coordinating session reviews " + "the packet and decides whether to file upstream." + ), + } + return packets + + +# --- rendering ---------------------------------------------------------------------- + + +def _gbp(value: float | None) -> str: + return "—" if value is None else f"{value / 1e6:,.1f}" + + +def render_packet_markdown(name: str, packet: dict) -> str: + lines = [ + f"# Investigation packet: {packet['event']} / {packet['measure_key']} / CY{packet['year']}", + "", + packet["filing"], + "", + f"* Title: {packet['title']}", + f"* Flags: {', '.join(packet['flags'])}", + f"* Base: `{packet['identities']['base']['variant_id']}`", + f"* Head: `{packet['identities']['head']['variant_id']}`", + "", + "## Movements (£m, positive = gain to the Exchequer)", + "", + "| Level | Key | Base | Head | Change | Flags | Details |", + "|---|---|---:|---:|---:|---|---|", + ] + for record in packet["records"]: + detail = "; ".join( + d["code"] + + (f" ({d['check']})" if "check" in d else "") + + (f" {d['path']}" if "path" in d else "") + for d in record["details"] + ) + lines.append( + f"| {record['level']} | {compare._md(record['key'])} | {_gbp(record['base_gbp'])} | " + f"{_gbp(record['head_gbp'])} | {_gbp(record['delta_gbp'])} | " + f"{', '.join(record['flags']) or '—'} | {compare._md(detail) or '—'} |" + ) + lines += [ + "", + "## OBR source cells", + "", + "| Source row | Table / column | FY | Head | OBR £m | Base £m | Head £m |", + "|---|---|---|---|---:|---:|---:|", + ] + for cell in packet["source_cells"]: + lines.append( + f"| {cell['source_row_id']} (row {cell['source_row_number']}) | " + f"{compare._md(cell['source_table'])} / {compare._md(cell['source_column'])} | " + f"{cell['fy']} | {compare._md(cell['tax_head'])} | {_gbp(cell['obr_value_gbp'])} | " + f"{_gbp(cell['base_gbp'])} | {_gbp(cell['head_gbp'])} |" + ) + lines += ["", "## Executed worlds", ""] + for side in ("base", "head"): + ref = packet["artifacts"][side] + lines.append( + f"* {side}: " + + ( + "no artifact" + if ref is None + else f"`{ref['path']}` (sha256 `{ref['sha256']}`), {ref['construction']}; " + f"baseline dict `{json.dumps(ref['baseline_reform_dict'], sort_keys=True)}`; " + f"reform dict `{json.dumps(ref['reform_reform_dict'], sort_keys=True)}`" + ) + ) + lines += ["", "## Reproduce", ""] + for side, command in packet["reproduce"].items(): + lines += [f"{side}:", "", "```bash", command, "```", ""] + lines += ["## Sources the registry cites", ""] + lines += [f"* {url}" for url in packet["cited_sources"]] or ["* none recorded"] + lines += [ + "", + "## Minimal case", + "", + PACKET_PENDING.capitalize() + ".", + "", + "## Statutory evidence", + "", + PACKET_PENDING.capitalize() + ".", + "", + ] + return "\n".join(lines) + + +def render_markdown(diff: dict, packet_names: list[str], max_rows: int) -> str: + base, head = diff["base"]["identity"], diff["head"]["identity"] + lines = [ + f"# Release diff: `{base['variant_id']}` → `{head['variant_id']}`", + "", + diff["description"], + "", + ( + "A numerical flag means the move exceeds max(£1m, 1% of the larger " + "magnitude). Structural, coverage, mapping, support, inertness and " + "construction flags are raised regardless of size. A year supported on " + "one side only is a support flag, not coverage." + ), + "", + "## Executions", + "", + "| Field | Base | Head |", + "|---|---|---|", + ] + for field in sorted(base): + lines.append(f"| {field} | `{base[field]}` | `{head[field]}` |") + lines += [ + "", + "## Flags by event", + "", + "| Event | Supported years (base → head) | " + " | ".join(FLAGS) + " |", + "|---|---|" + "---:|" * len(FLAGS), + ] + for event, data in diff["events"].items(): + years = data["supported_years"] + + def span(values): + return "absent" if not values else f"{values[0]}–{values[-1]}" + + lines.append( + f"| {event} | {span(years['base'])} → {span(years['head'])} | " + + " | ".join(str(data["flag_counts"][flag]) for flag in FLAGS) + + " |" + ) + lines.append( + "| **all** | | " + + " | ".join(str(diff["flag_counts"][flag]) for flag in FLAGS) + + " |" + ) + flagged = [r for r in diff["rows"] if r["flags"] and r["flags"] != ["support"]] + flagged.sort( + key=lambda r: ( + -(abs(r["delta_gbp"]) if r["delta_gbp"] is not None else math.inf), + r["event"], + r["measure_key"], + r["year"], + LEVELS.index(r["level"]), + r["key"], + ) + ) + shown = flagged[:max_rows] + lines += [ + "", + "## Flagged movements", + "", + f"Showing {len(shown)} of {len(flagged)} rows with a flag other than support, " + "rows without a numerical change first, then by absolute change. £m, " + "positive = gain to the Exchequer.", + "", + "| Event | Measure | CY | Level | Key | Base | Head | Change | Flags |", + "|---|---|---:|---|---|---:|---:|---:|---|", + ] + for r in shown: + lines.append( + f"| {r['event']} | {r['measure_key'].removeprefix(r['event'] + '__')} | {r['year']} | " + f"{r['level']} | {compare._md(r['key'])} | {_gbp(r['base_gbp'])} | {_gbp(r['head_gbp'])} | " + f"{_gbp(r['delta_gbp'])} | {', '.join(r['flags'])} |" + ) + support_only = [r for r in diff["rows"] if r["flags"] == ["support"]] + by_year = defaultdict(int) + for r in support_only: + by_year[(r["event"], r["year"], r["level"])] += 1 + lines += [ + "", + "## Years supported on one side only", + "", + "These rows have no packet; the population of one release cannot supply the year.", + "", + "| Event | CY | Level | Rows |", + "|---|---:|---|---:|", + ] + lines += [ + f"| {event} | {year} | {level} | {count} |" + for (event, year, level), count in sorted(by_year.items()) + ] or ["| none | | | |"] + lines += ["", "## PE/OBR ratio-bin transitions (source rows on both sides)", ""] + for event, data in diff["events"].items(): + lines += [ + f"### {event}", + "", + "| Base bin | Head bin | Rows |", + "|---|---|---:|", + ] + lines += [ + f"| {b} | {h} | {n} |" + for b, heads in data["bin_transitions"].items() + for h, n in heads.items() + ] or ["| — | — | 0 |"] + lines.append("") + lines += ["## Investigation packets", ""] + lines += [f"* [{name}](packets/{name}.md)" for name in packet_names] or ["* none"] + return "\n".join(lines) + "\n" + + +def render(base: dict, head: dict, *, max_rows: int = 200) -> dict[str, bytes]: + diff = diff_sides(base, head) + packets = build_packets(diff, base, head) + diff["packets"] = sorted(packets) + files = { + "DIFF.json": canonical_bytes(diff), + "DIFF.md": render_markdown(diff, sorted(packets), max_rows).encode(), + } + for name, packet in packets.items(): + files[f"packets/{name}.json"] = canonical_bytes(packet) + files[f"packets/{name}.md"] = render_packet_markdown(name, packet).encode() + return files + + +def write(files: dict[str, bytes], output_dir: Path) -> None: + stale = ( + {p for p in (output_dir / "packets").glob("*") if p.is_file()} + if (output_dir / "packets").is_dir() + else set() + ) + for relative, payload in sorted(files.items()): + path = output_dir / relative + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_bytes(path, payload) + stale.discard(path) + for path in stale: + path.unlink() + + +def check(files: dict[str, bytes], output_dir: Path) -> list[str]: + """Paths whose bytes on disk differ from a fresh render (CI's no-drift gate).""" + drift = [] + for relative, payload in sorted(files.items()): + path = output_dir / relative + if not path.is_file() or path.read_bytes() != payload: + drift.append(relative) + packets = output_dir / "packets" + if packets.is_dir(): + for path in sorted(packets.iterdir()): + if f"packets/{path.name}" not in files: + drift.append(f"packets/{path.name}") + return drift + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--base", required=True, help="namespace path or variant id") + parser.add_argument("--head", required=True, help="namespace path or variant id") + parser.add_argument("--output-dir", type=Path) + parser.add_argument("--max-rows", type=int, default=200) + parser.add_argument( + "--check", + action="store_true", + help="compare a fresh render with the files on disk", + ) + args = parser.parse_args(argv) + base = load_side(resolve_root(args.base)) + head = load_side(resolve_root(args.head)) + output_dir = args.output_dir or ( + DIFF_ROOT + / f"{base['identity']['variant_id']}__{head['identity']['variant_id']}" + ) + files = render(base, head, max_rows=args.max_rows) + if args.check: + drift = check(files, output_dir) + if drift: + print("release diff differs from a fresh render:", *drift, sep="\n ") + return 1 + print(f"{output_dir}: {len(files)} files match a fresh render") + return 0 + write(files, output_dir) + counts = json.loads(files["DIFF.json"])["flag_counts"] + print(f"wrote {len(files)} files to {output_dir}; flags: {counts}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pipeline/uk_replay_qualify.py b/pipeline/uk_replay_qualify.py new file mode 100644 index 0000000..7326515 --- /dev/null +++ b/pipeline/uk_replay_qualify.py @@ -0,0 +1,1169 @@ +"""Qualify a candidate model/data pair by replaying the event corpus on it. + +A qualification request (docs/uk_replay/QUALIFICATION_INTERFACE.md) names a +candidate exactly. This module turns it into a candidate replay variant, or a +rejection with a code; runs the supported corpus on it; diffs the run against +a base variant; and seals the namespace with an artifact-bound receipt. + +Steps, each a subcommand so the receiver workflow can run them in the right +environment: + +* ``plan``: validate the request and print the install commands for its + exact wheels (control environment; needs rfc8785 and jsonschema). +* ``probe``: inside the candidate environment, after installation, verify + the installed identity against the request, materialise the dataset and + read the release bundle, certification and data years. +* ``corpus``: inside the candidate environment, run compute, staging and + comparison for every event, then premises, diagnostics and the diff. +* ``receipt``: build, validate and seal the receipt; write + NAMESPACE_MANIFEST.json and REPLAY_RECEIPT.json. + +Nothing here decides whether a candidate is good: the receipt reports +integrity and descriptive diff flags; the promotion policy is release +qualification's. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import platform +import subprocess +import sys +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from pipeline import uk_replay_variants as variants # noqa: E402 + +EVENTS = ( + "autumn_budget_2024", + "autumn_statement_2023", + "spring_budget_2024", + "spring_statement_2025", + "spring_budget_2023", +) +CORPUS_VERSION = "uk-fiscal-event-replay/1" +REJECTION_CODES = ( + "schema_invalid", + "artifact_hash_mismatch", + "release_bundle_mismatch", + "incompatible_pair", + "data_access_failed", + "no_supported_years", +) +MANIFEST_NAME = "NAMESPACE_MANIFEST.json" +POLICYENGINE_GIT = "https://github.com/PolicyEngine/policyengine.py" +# Run in the candidate environment beside the engine, never resolved freely. +RUNTIME_TOOLS = ("pytest", "hypothesis", "rfc8785==0.1.4", "jsonschema") + + +class Rejected(Exception): + """The request cannot be qualified; the receipt carries the code.""" + + def __init__(self, code: str, detail: str): + if code not in REJECTION_CODES: + raise ValueError(f"unknown rejection code {code}") + super().__init__(f"{code}: {detail}") + self.code = code + self.detail = detail + + +def canonical_bytes(value: Any) -> bytes: + return ( + json.dumps(value, indent=1, sort_keys=True, allow_nan=False) + "\n" + ).encode() + + +def sha256_bytes(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +# --- the request ------------------------------------------------------------------ + + +def request_id(request: dict) -> str: + """Validate against the v1 schema; schema failures are rejections.""" + from pipeline import uk_replay_interface as interface + + try: + return interface.validate_request(request) + except Exception as exc: # noqa: BLE001 - jsonschema's errors, digest errors + raise Rejected("schema_invalid", str(exc).splitlines()[0]) from exc + + +def download_plan(request: dict, venv: Path, wheels: Path) -> list[list[str]]: + """Create the candidate environment and download the pinned wheels, + without dependencies, for ``verify_wheels`` to check.""" + candidate = request["candidate"] + python = str(venv / "bin" / "python") + pinned = [ + f"policyengine-uk=={candidate['policyengine_uk']['version']}", + f"policyengine-core=={candidate['policyengine_core']['version']}", + ] + if "pypi" in candidate["policyengine"]["source"]: + pinned.append(f"policyengine=={candidate['policyengine']['version']}") + return [ + ["uv", "venv", "--python", "3.12", str(venv)], + ["uv", "pip", "install", "--python", python, "pip"], + [ + python, + "-m", + "pip", + "download", + "--no-deps", + "--only-binary", + ":all:", + "-d", + str(wheels), + *pinned, + ], + ] + + +def requested_versions(request: dict) -> dict[str, str]: + candidate = request["candidate"] + return { + "policyengine": candidate["policyengine"]["version"], + "policyengine-uk": candidate["policyengine_uk"]["version"], + "policyengine-core": candidate["policyengine_core"]["version"], + } + + +def matching_certified(request: dict, document: dict) -> dict | None: + """A committed certified variant of exactly this pair, if any: same + versions, packaged manifest and dataset, and the same wheel digests where + both record them.""" + candidate = request["candidate"] + wheels = { + "policyengine-uk": candidate["policyengine_uk"]["wheel_sha256"], + "policyengine-core": candidate["policyengine_core"]["wheel_sha256"], + } + if "pypi" in candidate["policyengine"]["source"]: + wheels["policyengine"] = candidate["policyengine"]["source"]["pypi"][ + "wheel_sha256" + ] + for record in document["variants"]: + identity = record["identity"] + versions = {n: p["version"] for n, p in identity["packages"].items()} + if ( + record["status"] == "certified" + and versions == requested_versions(request) + and identity["packaged_bundle_manifest_sha256"] + == candidate["policyengine"]["packaged_bundle_manifest_sha256"] + and identity["dataset"]["sha256"] == candidate["data"]["sha256"] + and all( + identity["packages"][name].get("wheel_sha256", digest) == digest + for name, digest in wheels.items() + ) + ): + return record + return None + + +def latest_certified(document: dict) -> dict: + """The newest certified variant: the file is append-only, newest last.""" + return [r for r in document["variants"] if r["status"] == "certified"][-1] + + +def environment_lines(record: dict, *, root: Path = ROOT) -> list[str]: + """A variant's exact pins without the three engine packages.""" + pins = variants.requirements_pins( + (root / record["requirements"]["path"]).read_text() + ) + return [ + f"{name}=={version}" + for name, version in sorted(pins.items()) + if name not in variants.PACKAGES + ] + + +def install_command( + request: dict, + venv: Path, + wheels: Path, + *, + environment: Path | None = None, + exact: bool = False, +) -> list[str]: + """Install the verified local wheels, the wrapper from its public commit + when it is not a PyPI wheel, and the receipt tools. Re-qualifying a + certified pair installs that pair's exact numerical environment (``-r``); + a new candidate resolves within the newest certified one (``-c``). The + installed RECORD digests then bind the installed contents.""" + source = request["candidate"]["policyengine"]["source"] + command = [ + "uv", + "pip", + "install", + "--python", + str(venv / "bin" / "python"), + *sorted(str(path) for path in wheels.glob("*.whl")), + ] + if "git" in source: + command.append( + f"policyengine @ git+{POLICYENGINE_GIT}@{source['git']['commit']}" + ) + if environment is not None: + command += ["-r" if exact else "-c", str(environment)] + return command + list(RUNTIME_TOOLS) + + +def prepare( + request: dict, venv: Path, wheels: Path, *, root: Path = ROOT +) -> dict[str, str]: + """Build the candidate environment from exactly the requested wheels.""" + wheels.mkdir(parents=True, exist_ok=True) + for command in download_plan(request, venv, wheels): + run(command) + digests = verify_wheels(request, wheels) + document = variants.load_document(root / "data" / "uk" / "replay_variants.json") + match = matching_certified(request, document) + environment = wheels.parent / f"{wheels.name}-environment.txt" + environment.write_text( + "\n".join(environment_lines(match or latest_certified(document), root=root)) + + "\n" + ) + try: + run( + install_command( + request, venv, wheels, environment=environment, exact=bool(match) + ) + ) + except subprocess.CalledProcessError: + if match is not None: + raise + # The candidate's own requirements conflict with the newest certified + # environment: resolve freely; the recorded freeze shows what moved. + run(install_command(request, venv, wheels)) + return digests + + +def verify_wheels(request: dict, wheels: Path) -> dict[str, str]: + """Each downloaded wheel's SHA-256 against the request.""" + candidate = request["candidate"] + expected = { + "policyengine_uk": candidate["policyengine_uk"], + "policyengine_core": candidate["policyengine_core"], + } + if "pypi" in candidate["policyengine"]["source"]: + expected["policyengine"] = { + "version": candidate["policyengine"]["version"], + "wheel_sha256": candidate["policyengine"]["source"]["pypi"]["wheel_sha256"], + } + found = {} + for path in sorted(wheels.glob("*.whl")): + name, version = path.name.split("-")[:2] + found[name.lower()] = (version, variants.sha256_file(path)) + out = {} + for name, want in expected.items(): + version, digest = found.get(name, (None, None)) + if version != want["version"] or digest != want["wheel_sha256"]: + raise Rejected( + "artifact_hash_mismatch", + f"{name} wheel {version} sha256 {digest} is not the requested " + f"{want['version']} {want['wheel_sha256']}", + ) + out[name] = digest + return out + + +# --- probing the installed candidate --------------------------------------------------- + + +def probe( + request: dict, runtime_root: Path, *, dataset_file: Path | None = None +) -> dict: + """Run inside the candidate environment. Engine-dependent.""" + import importlib.metadata + + candidate = request["candidate"] + installed = variants.installed_identity() + want = { + "policyengine": candidate["policyengine"]["version"], + "policyengine-uk": candidate["policyengine_uk"]["version"], + "policyengine-core": candidate["policyengine_core"]["version"], + } + have = {name: row["version"] for name, row in installed["packages"].items()} + if have != want: + raise Rejected("artifact_hash_mismatch", f"installed {have}, requested {want}") + if ( + installed["packaged_bundle_manifest_sha256"] + != candidate["policyengine"]["packaged_bundle_manifest_sha256"] + ): + raise Rejected( + "artifact_hash_mismatch", + "installed packaged bundle manifest differs from the request", + ) + variants.isolate_certification_environment() + data = candidate["data"] + dataset_identity = { + key: data[key] + for key in ("repo_id", "repo_type", "revision", "path", "sha256", "size_bytes") + } + record_stub = { + "variant_id": "probe", + "identity": { + "dataset": dataset_identity, + "materialization": "policyengine_data_dir", + }, + } + try: + dataset = variants.materialize_dataset( + record_stub, runtime_root, source=dataset_file + ) + except variants.VariantError as exc: + raise Rejected("data_access_failed", str(exc)) from exc + os.chdir(runtime_root) + import policyengine as pe + + bundle = dict(pe.uk.uk_latest.release_bundle) + if bundle != candidate["expected_release_bundle"]: + differences = sorted( + key + for key in set(bundle) | set(candidate["expected_release_bundle"]) + if bundle.get(key) != candidate["expected_release_bundle"].get(key) + ) + raise Rejected("release_bundle_mismatch", f"fields differ: {differences}") + certification = pe.uk.uk_latest.data_certification + certified_for = getattr(certification, "certified_for_model_version", None) + if certified_for != want["policyengine-uk"]: + raise Rejected( + "incompatible_pair", + f"the packaged certification names policyengine-uk {certified_for}", + ) + if ( + bundle.get("certified_data_artifact_sha256") != data["sha256"] + or bundle.get("certified_data_build_id") != data["build_id"] + or bundle.get("default_dataset_uri") != variants.hf_uri(dataset_identity) + ): + raise Rejected( + "incompatible_pair", "the dataset is not the bundle's certified artifact" + ) + simulation = pe.uk.managed_microsimulation() + years = sorted(int(year) for year in simulation.dataset.years) + del simulation + return { + "installed": installed, + "release_bundle": bundle, + "certification": { + "source": ( + f"policyengine.py {want['policyengine']} packaged bundle manifest " + f"(qualification request; {json.dumps(candidate['policyengine']['source'], sort_keys=True)})" + ), + "compatibility_basis": bundle.get("compatibility_basis"), + "certified_for_model_version": certified_for, + "data_build_id": bundle.get("certified_data_build_id"), + }, + "dataset": dataset_identity, + "dataset_path": str(dataset), + "data_years": years, + "python": platform.python_version(), + "platform": platform.platform(), + "freeze": subprocess.run( + ["uv", "pip", "freeze", "--python", sys.executable], + check=True, + capture_output=True, + text=True, + ).stdout, + "versions": {name: importlib.metadata.version(name) for name in want}, + } + + +def candidate_record( + request: dict, probed: dict, *, root: Path = ROOT, wheel_sha256: dict | None = None +) -> dict: + """The variant this request runs as. + + A committed variant whose identity the installed environment reproduces + exactly (packages, manifest, bundle, dataset and every pinned + distribution) is returned unchanged. Otherwise a candidate record is + built, with its normalised freeze written as its requirements.""" + installed = probed["installed"] + document = variants.load_document(root / "data" / "uk" / "replay_variants.json") + + def engine(packages: dict) -> dict: + return { + name: {"version": row["version"], "record_sha256": row["record_sha256"]} + for name, row in packages.items() + } + + for record in document["variants"]: + identity = record["identity"] + pins = variants.requirements_pins( + (root / record["requirements"]["path"]).read_text() + ) + if ( + engine(identity["packages"]) == engine(installed["packages"]) + and identity["release_bundle"] == probed["release_bundle"] + and identity["dataset"] == probed["dataset"] + and not variants.identity_mismatches(record, installed) + and not variants.environment_mismatches(pins, installed) + ): + return record + packages = {} + for name in variants.PACKAGES: + packages[name] = dict(installed["packages"][name]) + key = name.replace("-", "_") + if wheel_sha256 and key in wheel_sha256: + packages[name]["wheel_sha256"] = wheel_sha256[key] + freeze = variants.normalized_freeze( + probed["freeze"], {n: p["version"] for n, p in installed["packages"].items()} + ) + identity = { + "packages": packages, + "packaged_bundle_manifest_sha256": installed["packaged_bundle_manifest_sha256"], + "release_bundle": probed["release_bundle"], + "dataset": probed["dataset"], + "materialization": "policyengine_data_dir", + "python": installed["python"], + "requirements_sha256": sha256_bytes(freeze.encode()), + } + taken = { + variants.variant_slug(r["identity"]): r["identity_sha256"] + for r in document["variants"] + } + variant_id = variants.variant_id_for(identity, taken) + requirements = root / "docs" / "uk_replay" / "requirements" / f"{variant_id}.txt" + if requirements.exists() and requirements.read_text() != freeze: + raise RuntimeError(f"{requirements} already holds another environment") + requirements.parent.mkdir(parents=True, exist_ok=True) + requirements.write_text(freeze) + years = probed["data_years"] + record = { + "variant_id": variant_id, + "identity_sha256": variants.identity_sha256(identity), + "identity": identity, + "status": "candidate" if request["kind"] == "candidate" else "certified", + "layout": "replay_namespace", + "result_root": f"{variants.REPLAY_RESULT_ROOT}/{variant_id}", + "requirements": { + "path": str(requirements.relative_to(root)), + "sha256": variants.sha256_file(requirements), + }, + "data_years": list(range(years[0], years[-1] + 1)), + "data_years_evidence": { + "simulation_dataset_years": years, + "source": "Measured by pipeline/uk_replay_qualify.py probe in the candidate environment.", + }, + "certification": probed["certification"], + "executor": ["local"], + "qualification_request_sha256": request_id(request), + } + variants.validate_record(record, root=root) + registry_years = set() + for event in EVENTS: + registry = json.loads( + (root / f"data/uk/events/{event}_measures.json").read_bytes() + ) + registry_years |= set(registry["calendar_years"]) + if not variants.supported_years(record, registry_years)[0]: + raise Rejected("no_supported_years", f"data years {years} meet no event year") + return record + + +# --- the corpus ------------------------------------------------------------------------ + + +def run(command: list[str], *, cwd: Path = ROOT) -> None: + print("+", " ".join(command), flush=True) + subprocess.run( + command, check=True, cwd=cwd, env={**os.environ, "PYTHONPATH": str(ROOT)} + ) + + +def run_corpus( + record: dict, + variants_file: Path, + *, + workers: int = 2, + dataset_file: Path | None = None, + events: tuple[str, ...] = EVENTS, +) -> list[dict]: + """Compute, stage and compare every event; returns integrity failures.""" + python = sys.executable + namespace = variants.namespace_root(record) + failures = [] + for event in events: + output = namespace / event + try: + command = [ + python, + "-m", + "pipeline.compute_uk_event", + "--event", + event, + "--variant", + record["variant_id"], + "--variants-file", + str(variants_file), + "--workers", + str(workers), + "--output-dir", + str(output), + ] + if dataset_file is not None: + command += ["--dataset-file", str(dataset_file)] + if (output / "RUN_MANIFEST.json").exists() or any( + output.glob("RUN_PROGRESS_*.json") + ): + command.append("--resume") + run(command) + run( + [ + python, + "pipeline/stage_uk_event.py", + "--event", + event, + "--artifact-dir", + str(output), + ] + ) + run( + [ + python, + "pipeline/compare_uk_event.py", + "--event", + event, + "--staged", + str(output / "STAGED.jsonl"), + "--output-dir", + str(output), + ] + ) + except subprocess.CalledProcessError as exc: + failures.append( + {"code": "event_run_failed", "detail": str(exc), "event": event} + ) + run( + [ + python, + "pipeline/compare_uk_event.py", + "--summary", + "--events-root", + str(namespace), + ] + ) + engine = record["identity"]["packages"]["policyengine-uk"]["version"] + run( + [ + python, + "-m", + "pipeline.audit_uk_event_premises", + "--all", + "--expect-engine", + engine, + "--output", + str(namespace / "CONSTRUCTION_PREMISES.json"), + ] + ) + run( + [ + python, + "pipeline/diagnose_uk_event_models.py", + "--reverify", + str(ROOT / "results/uk/events/MODEL_DIAGNOSTICS.json"), + "--expect-engine", + engine, + "--output", + str(namespace / "MODEL_DIAGNOSTICS.json"), + ] + ) + support = {} + for event in events: + manifest_path = namespace / event / "RUN_MANIFEST.json" + if manifest_path.exists(): + manifest = json.loads(manifest_path.read_bytes()) + support[event] = { + "supported_years": manifest["supported_years"], + "unsupported_years": manifest["unsupported_years"], + } + (namespace / "SUPPORT.json").write_bytes(canonical_bytes(support)) + return failures + + +# --- receipts and sealing ------------------------------------------------------------ + + +def namespace_files(namespace: Path) -> dict[str, str]: + return { + str(path.relative_to(namespace)): variants.sha256_file(path) + for path in sorted(namespace.rglob("*")) + if path.is_file() and path.name not in (MANIFEST_NAME, variants.RECEIPT_NAME) + } + + +def event_results(record: dict, events=EVENTS) -> tuple[dict, list[dict]]: + namespace = variants.namespace_root(record) + out, failures = {}, [] + for event in events: + directory = namespace / event + try: + manifest_bytes = (directory / "RUN_MANIFEST.json").read_bytes() + staging_bytes = (directory / "STAGING_MANIFEST.json").read_bytes() + provenance_bytes = (directory / "COMPARISON_PROVENANCE.json").read_bytes() + except FileNotFoundError as exc: + failures.append( + {"code": "event_outputs_missing", "detail": str(exc), "event": event} + ) + continue + manifest, staging = json.loads(manifest_bytes), json.loads(staging_bytes) + out[event] = { + "registry_sha256": manifest["registry_sha256"], + "supported_years": manifest["supported_years"], + "unsupported_years": manifest["unsupported_years"], + "run_manifest_sha256": sha256_bytes(manifest_bytes), + "staging_manifest_sha256": sha256_bytes(staging_bytes), + "comparison_provenance_sha256": sha256_bytes(provenance_bytes), + "artifacts": len(manifest["artifacts"]), + "full_supported_grid_complete": bool( + staging.get("full_supported_grid_complete") + ), + "inert_measure_years": staging.get("inert_measure_years", []), + } + return out, failures + + +def diff_summary(diff_dir: Path) -> dict: + payload = (diff_dir / "DIFF.json").read_bytes() + diff = json.loads(payload) + return { + "base_variant_id": diff["base"]["identity"]["variant_id"], + "diff_sha256": sha256_bytes(payload), + "flag_counts": diff["flag_counts"], + "flagged_rows": [ + { + "event": row["event"], + "measure_key": row["measure_key"], + "year": row["year"], + "level": row["level"], + "key": row["key"], + "flags": row["flags"], + "base_gbp": row["base_gbp"], + "head_gbp": row["head_gbp"], + } + for row in diff["rows"] + if row["flags"] + ], + } + + +def build_receipt( + request: dict, + *, + record: dict | None, + rejection: Rejected | None = None, + failures: list[dict] | None = None, + diff_dir: Path | None = None, + results_path: str, + ref: str = "replay-state", + runner: str = "local", +) -> dict: + from pipeline import uk_replay_interface as interface + + try: + rid = interface.request_id(request) + except Exception: # noqa: BLE001 - an unparseable request still gets a receipt + rid = interface.jcs_sha256(request) + failures = list(failures or []) + receipt = { + "schema": interface.RECEIPT_SCHEMA_ID, + "request_id": rid, + "status": "rejected" if rejection else "complete", + "rejection": {"code": rejection.code, "detail": rejection.detail} + if rejection + else None, + "variant_id": record["variant_id"] if record else None, + "variant_identity_sha256": record["identity_sha256"] if record else None, + "execution": None, + "corpus": None, + "integrity": {"status": "not_run", "failures": []}, + "diff": None, + "results": { + "repository": "PolicyEngine/policyengine-scorecard", + "ref": ref, + "path": results_path, + }, + } + if record is None and rejection is None: + # A step failed before the candidate's identity was established. + receipt["status"] = "failed" + receipt["integrity"] = {"status": "failed", "failures": failures} + if record is not None and rejection is None: + identity = record["identity"] + receipt["execution"] = { + "packages": identity["packages"], + "packaged_bundle_manifest_sha256": identity[ + "packaged_bundle_manifest_sha256" + ], + "release_bundle": identity["release_bundle"], + "dataset_sha256": identity["dataset"]["sha256"], + "requirements_sha256": record["requirements"]["sha256"], + "python": platform.python_version(), + "platform": platform.platform(), + "runner": runner, + } + events, missing = event_results(record) + failures += missing + receipt["corpus"] = {"corpus_version": CORPUS_VERSION, "events": events} + complete = events and all( + e["full_supported_grid_complete"] for e in events.values() + ) + receipt["integrity"] = { + "status": "failed" + if failures + else "complete" + if complete + else "incomplete", + "failures": failures, + } + if failures: + receipt["status"] = "failed" + if diff_dir is not None and (diff_dir / "DIFF.json").exists(): + receipt["diff"] = diff_summary(diff_dir) + return receipt + + +def seal(namespace: Path, receipt: dict, *, diff_dir: Path | None = None) -> dict: + """Bind every namespace file (and the diff) and write the sealed receipt.""" + from pipeline import uk_replay_interface as interface + + manifest = { + "schema_version": 1, + "files": namespace_files(namespace), + "diff_files": { + str(path.relative_to(diff_dir)): variants.sha256_file(path) + for path in sorted(diff_dir.rglob("*")) + if path.is_file() + } + if diff_dir is not None and diff_dir.is_dir() + else {}, + } + payload = canonical_bytes(manifest) + (namespace / MANIFEST_NAME).write_bytes(payload) + receipt = dict(receipt) + receipt["results"] = { + **receipt["results"], + "manifest_sha256": sha256_bytes(payload), + } + sealed = interface.seal_receipt(receipt) + interface.validate(sealed, interface.RECEIPT_SCHEMA_ID) + (namespace / variants.RECEIPT_NAME).write_bytes(canonical_bytes(sealed)) + return sealed + + +def verify_sealed(namespace: Path, *, diff_dir: Path | None = None) -> list[str]: + """Problems with a sealed namespace; empty when every byte matches.""" + from pipeline import uk_replay_interface as interface + + problems = [] + receipt_path = namespace / variants.RECEIPT_NAME + manifest_path = namespace / MANIFEST_NAME + if not receipt_path.is_file() or not manifest_path.is_file(): + return [f"{namespace}: not sealed"] + receipt = json.loads(receipt_path.read_bytes()) + try: + interface.validate(receipt, interface.RECEIPT_SCHEMA_ID) + except Exception as exc: # noqa: BLE001 + problems.append(f"{receipt_path}: {exc}") + manifest_bytes = manifest_path.read_bytes() + if receipt.get("results", {}).get("manifest_sha256") != sha256_bytes( + manifest_bytes + ): + problems.append(f"{manifest_path}: digest differs from the receipt") + manifest = json.loads(manifest_bytes) + actual = namespace_files(namespace) + if actual != manifest["files"]: + changed = sorted( + set(actual.items()) ^ set(manifest["files"].items()), + key=lambda item: item[0], + ) + problems.append( + f"{namespace}: files differ from the seal: {[c[0] for c in changed][:10]}" + ) + if diff_dir is not None and manifest.get("diff_files"): + actual_diff = { + str(path.relative_to(diff_dir)): variants.sha256_file(path) + for path in sorted(diff_dir.rglob("*")) + if path.is_file() + } + if actual_diff != manifest["diff_files"]: + problems.append(f"{diff_dir}: files differ from the seal") + return problems + + +# --- one request, end to end --------------------------------------------------------- + +PROBE_REJECTED = 3 + + +def fetch_dataset(request: dict) -> Path: + """Download the requested artifact into the Hugging Face cache with the + receiver's token; probe later finds it there by digest.""" + from huggingface_hub import hf_hub_download + + data = request["candidate"]["data"] + token = os.environ.get("HUGGING_FACE_TOKEN") or os.environ.get("HF_TOKEN") + try: + path = Path( + hf_hub_download( + repo_id=data["repo_id"], + repo_type=data["repo_type"], + filename=data["path"], + revision=data["revision"], + token=token or None, + ) + ) + except Exception as exc: # noqa: BLE001 - authentication, absence, network + raise Rejected( + "data_access_failed", f"{type(exc).__name__}: {exc}".splitlines()[0] + ) from exc + if ( + path.stat().st_size != data["size_bytes"] + or variants.sha256_file(path) != data["sha256"] + ): + raise Rejected( + "artifact_hash_mismatch", f"{data['path']} bytes differ from the request" + ) + return path + + +def base_namespace(request: dict, record: dict, document: dict) -> Path | None: + """The namespace to diff against: compare_to, or the newest certified + variant other than this one whose results are present.""" + wanted = request.get("compare_to", "latest_certified") + candidates = ( + [r for r in document["variants"] if r["variant_id"] == wanted] + if wanted != "latest_certified" + else [ + r + for r in reversed(document["variants"]) + if r["status"] == "certified" and r["variant_id"] != record["variant_id"] + ] + ) + for candidate in candidates: + root = variants.namespace_root(candidate) + if any(root.glob("*/COMPARISON.json")): + return root + return None + + +def run_request(request: dict, state_dir: Path, work: Path, *, runner: str) -> dict: + """Qualify one request and write its receipt and results to the state + checkout. Every outcome, including a rejection, produces a receipt.""" + import shutil + + from pipeline import uk_replay_interface as interface + + try: + rid = request_id(request) + except Rejected: + rid = interface.jcs_sha256(request) + destination = state_dir / "requests" / rid + results_path = f"requests/{rid}/results/" + work.mkdir(parents=True, exist_ok=True) + venv, wheels = work / "venv", work / "wheels" + variants_file = work / "variants.json" + record = rejection = diff_dir = None + failures: list[dict] = [] + try: + request_id(request) + fetch_dataset(request) + digests = prepare(request, venv, wheels) + (work / "wheels.json").write_bytes(canonical_bytes(digests)) + python = str(venv / "bin" / "python") + (work / "request.json").write_bytes(canonical_bytes(request)) + probe_run = subprocess.run( + [ + python, + "-m", + "pipeline.uk_replay_qualify", + "probe", + "--request", + str(work / "request.json"), + "--runtime-root", + str(work / "runtime"), + "--wheel-digests", + str(work / "wheels.json"), + "--variants-file", + str(variants_file), + "--rejection-file", + str(work / "rejection.json"), + ], + cwd=ROOT, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + ) + if probe_run.returncode == PROBE_REJECTED: + refused = json.loads((work / "rejection.json").read_bytes()) + raise Rejected(refused["code"], refused["detail"]) + if probe_run.returncode != 0: + raise subprocess.CalledProcessError( + probe_run.returncode, "probe", probe_run.stdout, probe_run.stderr + ) + variant_id = probe_run.stdout.strip().splitlines()[-1] + record = variants.get_variant(variant_id, path=variants_file) + run( + [ + python, + "-m", + "pipeline.uk_replay_qualify", + "corpus", + "--variant", + variant_id, + "--variants-file", + str(variants_file), + "--failures", + str(work / "failures.json"), + ] + ) + failures = json.loads((work / "failures.json").read_bytes()) + document = variants.load_document(variants_file) + base = base_namespace(request, record, document) + if base is not None: + base_id = ( + json.loads(next(base.glob("*/RUN_MANIFEST.json")).read_bytes()).get( + "variant_id" + ) + or variants.get_variant(path=variants_file)["variant_id"] + ) + diff_dir = ( + ROOT + / variants.REPLAY_RESULT_ROOT + / "diffs" + / f"{base_id}__{variant_id}" + ) + run( + [ + python, + "-m", + "pipeline.diff_uk_event_runs", + "--base", + str(base), + "--head", + str(variants.namespace_root(record)), + "--output-dir", + str(diff_dir), + ] + ) + except Rejected as exc: + rejection = exc + except subprocess.CalledProcessError as exc: + failures.append( + { + "code": "qualification_step_failed", + "detail": f"{exc.cmd}: exit {exc.returncode}", + } + ) + except Exception as exc: # noqa: BLE001 - a crash must still produce a receipt + failures.append( + { + "code": "unexpected_error", + "detail": f"{type(exc).__name__}: {exc}".splitlines()[0][:500], + } + ) + receipt = build_receipt( + request, + record=None if rejection else record, + rejection=rejection, + failures=failures, + diff_dir=diff_dir, + results_path=results_path, + runner=runner, + ) + if record is not None and rejection is None: + namespace = variants.namespace_root(record) + sealed = seal(namespace, receipt, diff_dir=diff_dir) + target = destination / "results" + shutil.copytree(namespace, target / record["result_root"], dirs_exist_ok=True) + if diff_dir is not None and diff_dir.is_dir(): + shutil.copytree( + diff_dir, + target / diff_dir.relative_to(ROOT), + dirs_exist_ok=True, + ) + requirements = ROOT / record["requirements"]["path"] + (target / record["requirements"]["path"]).parent.mkdir( + parents=True, exist_ok=True + ) + shutil.copy2(requirements, target / record["requirements"]["path"]) + (target / "VARIANT.json").write_bytes(canonical_bytes(record)) + else: + sealed = interface.seal_receipt(receipt) + interface.validate(sealed, interface.RECEIPT_SCHEMA_ID) + destination.mkdir(parents=True, exist_ok=True) + (destination / variants.RECEIPT_NAME).write_bytes(canonical_bytes(sealed)) + return sealed + + +# --- the command line ------------------------------------------------------------------ + + +def _load(path: Path) -> dict: + return json.loads(path.read_text(encoding="utf-8")) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + commands = parser.add_subparsers(dest="command", required=True) + validate = commands.add_parser("validate") + validate.add_argument("--request", type=Path, required=True) + prep = commands.add_parser("prepare") + prep.add_argument("--request", type=Path, required=True) + prep.add_argument("--venv", type=Path, required=True) + prep.add_argument("--wheels", type=Path, required=True) + prep.add_argument("--output", type=Path, required=True) + probe_parser = commands.add_parser("probe") + probe_parser.add_argument("--request", type=Path, required=True) + probe_parser.add_argument("--runtime-root", type=Path, required=True) + probe_parser.add_argument("--dataset-file", type=Path) + probe_parser.add_argument("--wheel-digests", type=Path) + probe_parser.add_argument("--variants-file", type=Path, required=True) + probe_parser.add_argument("--rejection-file", type=Path) + request_parser = commands.add_parser("run-request") + request_parser.add_argument("--request", type=Path, required=True) + request_parser.add_argument("--state-dir", type=Path, required=True) + request_parser.add_argument("--work", type=Path, required=True) + request_parser.add_argument( + "--runner", default=os.environ.get("REPLAY_RUNNER", "local") + ) + corpus = commands.add_parser("corpus") + corpus.add_argument("--variant", required=True) + corpus.add_argument("--variants-file", type=Path, required=True) + corpus.add_argument("--workers", type=int, choices=(1, 2), default=2) + corpus.add_argument("--dataset-file", type=Path) + corpus.add_argument("--failures", type=Path, required=True) + receipt = commands.add_parser("receipt") + receipt.add_argument("--request", type=Path, required=True) + receipt.add_argument("--variant") + receipt.add_argument("--variants-file", type=Path) + receipt.add_argument("--rejection") + receipt.add_argument("--failures", type=Path) + receipt.add_argument("--diff-dir", type=Path) + receipt.add_argument("--results-path", required=True) + receipt.add_argument("--ref", default="replay-state") + receipt.add_argument("--runner", default=os.environ.get("REPLAY_RUNNER", "local")) + receipt.add_argument("--output", type=Path) + check = commands.add_parser("verify-sealed") + check.add_argument("namespaces", nargs="*", type=Path) + args = parser.parse_args(argv) + + if args.command == "validate": + print(request_id(_load(args.request))) + return 0 + if args.command == "prepare": + digests = prepare( + _load(args.request), args.venv.resolve(), args.wheels.resolve() + ) + args.output.write_bytes(canonical_bytes(digests)) + return 0 + if args.command == "probe": + request = _load(args.request) + digests = _load(args.wheel_digests) if args.wheel_digests else None + try: + probed = probe( + request, args.runtime_root.resolve(), dataset_file=args.dataset_file + ) + os.chdir(ROOT) + record = candidate_record(request, probed, wheel_sha256=digests) + except Rejected as exc: + if args.rejection_file is None: + raise + args.rejection_file.write_bytes( + canonical_bytes({"code": exc.code, "detail": exc.detail}) + ) + return PROBE_REJECTED + document = variants.load_document() + document["variants"] = [ + v for v in document["variants"] if v["variant_id"] != record["variant_id"] + ] + [record] + args.variants_file.write_bytes(canonical_bytes(document)) + print(record["variant_id"]) + return 0 + if args.command == "run-request": + receipt = run_request( + _load(args.request), args.state_dir, args.work.resolve(), runner=args.runner + ) + print(f"{receipt['status']} {receipt['receipt_sha256']}") + return 0 + if args.command == "corpus": + record = variants.get_variant(args.variant, path=args.variants_file.resolve()) + failures = run_corpus( + record, + args.variants_file.resolve(), + workers=args.workers, + dataset_file=args.dataset_file, + ) + args.failures.write_bytes(canonical_bytes(failures)) + return 0 + if args.command == "receipt": + request = _load(args.request) + record = ( + variants.get_variant(args.variant, path=args.variants_file.resolve()) + if args.variant + else None + ) + rejection = None + if args.rejection: + code, _, detail = args.rejection.partition(":") + rejection = Rejected(code.strip(), detail.strip() or code.strip()) + failures = ( + _load(args.failures) if args.failures and args.failures.exists() else [] + ) + built = build_receipt( + request, + record=record, + rejection=rejection, + failures=failures, + diff_dir=args.diff_dir, + results_path=args.results_path, + ref=args.ref, + runner=args.runner, + ) + if record is not None and rejection is None: + sealed = seal( + variants.namespace_root(record), built, diff_dir=args.diff_dir + ) + else: + from pipeline import uk_replay_interface as interface + + sealed = interface.seal_receipt(built) + interface.validate(sealed, interface.RECEIPT_SCHEMA_ID) + if args.output: + args.output.write_bytes(canonical_bytes(sealed)) + print(sealed["receipt_sha256"]) + return 0 + if args.command == "verify-sealed": + namespaces = args.namespaces or sorted( + path.parent + for path in (ROOT / variants.REPLAY_RESULT_ROOT).glob( + f"*/{variants.RECEIPT_NAME}" + ) + ) + problems = [] + for namespace in namespaces: + receipt = json.loads((namespace / variants.RECEIPT_NAME).read_bytes()) + diff = None + if receipt.get("diff"): + diff = ( + ROOT + / variants.REPLAY_RESULT_ROOT + / "diffs" + / f"{receipt['diff']['base_variant_id']}__{receipt['variant_id']}" + ) + problems += verify_sealed(namespace, diff_dir=diff) + for problem in problems: + print(problem) + print(f"{len(namespaces)} sealed namespace(s) checked") + return 1 if problems else 0 + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pipeline/uk_replay_queue.py b/pipeline/uk_replay_queue.py new file mode 100644 index 0000000..ea8f466 --- /dev/null +++ b/pipeline/uk_replay_queue.py @@ -0,0 +1,179 @@ +"""The persistent qualification queue on the ``replay-state`` branch. + +GitHub keeps at most one pending run per concurrency group and cancels older +pending runs, so a request must be written down before anything else can +lose it. Each request lives at ``requests//`` in a checkout of +the state branch: ``REQUEST.json`` (the request as received), ``STATUS.json`` +(schema policyengine.uk-replay-status/v1) and, once terminal, +``REPLAY_RECEIPT.json`` and ``results/``. Enqueueing is idempotent: a +request already present keeps its status. The workflow commits and pushes +after each call; this module only edits files. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timedelta +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from pipeline import uk_replay_interface as interface # noqa: E402 + +TERMINAL = ("complete", "rejected", "failed") +STATUSES = ("queued", "running", *TERMINAL) +# A running request whose last update is older than this has lost its run +# (a cancelled or crashed job never writes a terminal status). +STALE_RUNNING = timedelta(hours=7) + + +def canonical_bytes(value) -> bytes: + return ( + json.dumps(value, indent=1, sort_keys=True, ensure_ascii=False) + "\n" + ).encode() + + +def request_dir(state: Path, request_id: str) -> Path: + if len(request_id) != 64 or any(c not in "0123456789abcdef" for c in request_id): + raise ValueError("request_id must be 64 lowercase hex characters") + return state / "requests" / request_id + + +def read_status(state: Path, request_id: str) -> dict | None: + path = request_dir(state, request_id) / "STATUS.json" + return json.loads(path.read_bytes()) if path.exists() else None + + +def write_status(state: Path, status: dict) -> None: + interface.validate(status, interface.STATUS_SCHEMA_ID) + path = request_dir(state, status["request_id"]) / "STATUS.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(canonical_bytes(status)) + + +def enqueue(state: Path, request: dict, *, run_url: str, now: str) -> tuple[str, str]: + """Record a request; returns (request_id, its status). Idempotent. + + A request that fails the schema is still recorded, by the digest of its + canonical bytes, so the receipt that rejects it has an address.""" + try: + request_id = interface.validate_request(request) + except Exception: # noqa: BLE001 - recorded, rejected by the drain step + request_id = interface.jcs_sha256(request) + existing = read_status(state, request_id) + if existing is not None: + return request_id, existing["status"] + directory = request_dir(state, request_id) + directory.mkdir(parents=True, exist_ok=True) + (directory / "REQUEST.json").write_bytes(canonical_bytes(request)) + write_status( + state, + { + "schema": interface.STATUS_SCHEMA_ID, + "request_id": request_id, + "status": "queued", + "history": [{"status": "queued", "at": now, "run_url": run_url}], + }, + ) + return request_id, "queued" + + +def set_status( + state: Path, + request_id: str, + status: str, + *, + run_url: str, + now: str, + receipt_sha256: str | None = None, + receipt_commit: str | None = None, +) -> dict: + current = read_status(state, request_id) + if current is None: + raise ValueError(f"{request_id} is not queued") + if current["status"] in TERMINAL: + raise ValueError( + f"{request_id} is already {current['status']}; terminal states are final" + ) + if status not in STATUSES: + raise ValueError(f"unknown status {status}") + updated = { + **current, + "status": status, + "history": [ + *current["history"], + {"status": status, "at": now, "run_url": run_url}, + ], + } + if status in TERMINAL: + if not (receipt_sha256 and receipt_commit): + raise ValueError("a terminal status names its receipt digest and commit") + updated.update(receipt_sha256=receipt_sha256, receipt_commit=receipt_commit) + write_status(state, updated) + return updated + + +def _at(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def pending(state: Path, *, now: str) -> list[str]: + """Requests to run, oldest first: queued, or running past the stale bound.""" + out = [] + for path in sorted((state / "requests").glob("*/STATUS.json")): + status = json.loads(path.read_bytes()) + last = status["history"][-1]["at"] + if status["status"] == "queued" or ( + status["status"] == "running" and _at(now) - _at(last) > STALE_RUNNING + ): + out.append((status["history"][0]["at"], status["request_id"])) + return [request_id for _, request_id in sorted(out)] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + commands = parser.add_subparsers(dest="command", required=True) + add = commands.add_parser("enqueue") + add.add_argument("--state", type=Path, required=True) + add.add_argument("--request", type=Path, required=True) + add.add_argument("--run-url", required=True) + add.add_argument("--now", required=True) + nxt = commands.add_parser("pending") + nxt.add_argument("--state", type=Path, required=True) + nxt.add_argument("--now", required=True) + mark = commands.add_parser("set-status") + mark.add_argument("--state", type=Path, required=True) + mark.add_argument("--request-id", required=True) + mark.add_argument("--status", choices=STATUSES, required=True) + mark.add_argument("--run-url", required=True) + mark.add_argument("--now", required=True) + mark.add_argument("--receipt-sha256") + mark.add_argument("--receipt-commit") + args = parser.parse_args(argv) + if args.command == "enqueue": + request = json.loads(args.request.read_text(encoding="utf-8")) + request_id, status = enqueue( + args.state, request, run_url=args.run_url, now=args.now + ) + print(f"{request_id} {status}") + elif args.command == "pending": + print("\n".join(pending(args.state, now=args.now))) + else: + set_status( + args.state, + args.request_id, + args.status, + run_url=args.run_url, + now=args.now, + receipt_sha256=args.receipt_sha256, + receipt_commit=args.receipt_commit, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pipeline/uk_replay_variants.py b/pipeline/uk_replay_variants.py index a2f1ee4..2513961 100644 --- a/pipeline/uk_replay_variants.py +++ b/pipeline/uk_replay_variants.py @@ -50,8 +50,16 @@ VERSION = re.compile(r"^[0-9]+(?:\.[0-9]+)*[a-z0-9.+-]*$") PIN = re.compile(r"([A-Za-z0-9][A-Za-z0-9_.-]*)==([A-Za-z0-9_.+!-]+)") # Installer bookkeeping differs between pip and uv for the same wheel; only -# the distribution's own files identify what was installed. -INSTALLER_METADATA = ("INSTALLER", "REQUESTED", "RECORD", "direct_url.json") +# the distribution's own files identify what was installed. Paths outside +# site-packages ("../../../bin/...") are console-script wrappers the +# installer generates, with the environment's interpreter in their shebang. +INSTALLER_METADATA = ( + "INSTALLER", + "REQUESTED", + "RECORD", + "direct_url.json", + "uv_cache.json", +) class VariantError(ValueError): @@ -91,6 +99,44 @@ def variant_slug(identity: dict) -> str: ) +def variant_id_for(identity: dict, taken: dict[str, str] | None = None) -> str: + """The version slug, suffixed with the identity digest when another + identity already holds that slug (same versions, other wheels or + numerical environment).""" + slug = variant_slug(identity) + digest = identity_sha256(identity) + holder = (taken or {}).get(slug) + return slug if holder in (None, digest) else f"{slug}--{digest[:10]}" + + +def normalized_freeze(text: str, versions: dict[str, str]) -> str: + """A freeze as exact pins. Engine packages installed from a verified local + wheel or a commit appear as ``name @ ``; they become ``name==version`` + (the wheel and installed-file digests bind them) with the source kept as a + comment. Any other non-pinned line refuses.""" + lines = [] + for line in text.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + name, at, source = stripped.partition(" @ ") + key = re.sub(r"[-_.]+", "-", name).lower() + if at and key in versions: + lines.append( + f"# {key} installed from {source.split('/')[-1] if source.startswith('file:') else source}" + ) + lines.append(f"{key}=={versions[key]}") + elif at: + raise VariantError(f"freeze line is not an exact pin: {stripped}") + else: + lines.append(stripped) + pins = requirements_pins("\n".join(lines)) + missing = sorted(set(versions) - set(pins)) + if missing: + raise VariantError(f"freeze lacks {missing}") + return "\n".join(lines) + "\n" + + def hf_uri(dataset: dict) -> str: return f"hf://{dataset['repo_id']}/{dataset['path']}@{dataset['revision']}" @@ -189,7 +235,11 @@ def need(condition: bool, message: str) -> None: uk = packages["policyengine-uk"]["version"] wrapper = packages["policyengine"]["version"] need( - record.get("variant_id") == variant_slug(identity), + record.get("variant_id") + in ( + variant_slug(identity), + f"{variant_slug(identity)}--{identity_sha256(identity)[:10]}", + ), "variant_id differs from its identity", ) need( @@ -280,6 +330,10 @@ def need(condition: bool, message: str) -> None: f"result_root must be {expected_root}", ) requirements = record.get("requirements") or {} + need( + identity.get("requirements_sha256") == requirements.get("sha256"), + "the identity binds another requirements file", + ) requirements_path = root / str(requirements.get("path", "")) if not requirements.get("path") or not requirements_path.is_file(): problems.append("requirements file is missing") @@ -342,6 +396,7 @@ def _record_rows(distribution) -> list[tuple[str, str, str]]: and name in INSTALLER_METADATA or "__pycache__/" in path or path.endswith(".pyc") + or path.startswith("../") ): continue rows.append((path, digest, size)) @@ -394,9 +449,25 @@ def installed_identity(*, verify_files: bool = True) -> dict: "packages": packages, "packaged_bundle_manifest_sha256": sha256_file(manifest), "python": f"{sys.version_info.major}.{sys.version_info.minor}", + "distributions": { + re.sub(r"[-_.]+", "-", d.metadata["Name"]).lower(): d.version + for d in importlib.metadata.distributions() + }, } +def environment_mismatches(pins: dict[str, str], installed: dict) -> list[str]: + """Every pinned distribution installed at its pinned version. Extra + distributions (the receipt tools) are allowed; a missing or different + numerical dependency is not.""" + have = installed.get("distributions", {}) + return [ + f"{name}: pinned {version}, installed {have.get(name)}" + for name, version in sorted(pins.items()) + if have.get(name) != version + ] + + def identity_mismatches(record: dict, installed: dict) -> list[str]: identity = record["identity"] out = [] @@ -424,9 +495,13 @@ def identity_mismatches(record: dict, installed: dict) -> list[str]: return out -def verify_installed(record: dict, *, installed: dict | None = None) -> dict: +def verify_installed( + record: dict, *, installed: dict | None = None, root: Path = ROOT +) -> dict: installed = installed if installed is not None else installed_identity() mismatches = identity_mismatches(record, installed) + pins = requirements_pins((root / record["requirements"]["path"]).read_text()) + mismatches += environment_mismatches(pins, installed) if mismatches: raise VariantError( f"{record['variant_id']}: installed environment is not the certified one: " diff --git a/tests/test_uk_event_diff.py b/tests/test_uk_event_diff.py new file mode 100644 index 0000000..07ec8c6 --- /dev/null +++ b/tests/test_uk_event_diff.py @@ -0,0 +1,614 @@ +"""The release diff: flags describe movements; nothing passes or fails. + +Synthetic namespaces go through the real artifact, staging and comparison +writers, so the diff reads them through the same verified loaders it uses on +real replays. The committed legacy namespace is diffed against itself. +""" + +import copy +import hashlib +import json +import random +from pathlib import Path + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from pipeline import compare_uk_event as compare +from pipeline import compute_uk_ab2025 as ab +from pipeline import compute_uk_event as compute +from pipeline import diff_uk_event_runs as diff +from pipeline import stage_uk_event as stage + +EVENT = "test_event" +KEY = f"{EVENT}__tax_and_benefit" + + +def measure(modifier=0.2, heads=None): + rows = [] + for year in (2024, 2025): + fy = f"{year}-{str(year + 1)[-2:]}" + rows += [ + { + "source_row_id": f"tax_{year}", + "source_row_number": year, + "fy": fy, + "metric": "revenue_change", + "tax_head": "Income tax", + "value_gbp": 100, + "source_table": "T", + "source_column": "C", + }, + { + "source_row_id": f"benefit_{year}", + "source_row_number": year + 1, + "fy": fy, + "metric": "exchequer_impact", + "tax_head": "Welfare inside cap", + "value_gbp": -30, + "source_table": "T", + "source_column": "D", + }, + { + "source_row_id": f"corporation_{year}", + "source_row_number": year + 2, + "fy": fy, + "metric": "revenue_change", + "tax_head": "Corporation tax", + "value_gbp": 4, + "classification": "out_of_household_scope", + "source_table": "T", + "source_column": "E", + }, + ] + heads = heads or [ + {"obr_head": "Income tax", "pe_variables": ["income_tax"], "channel": "tax"}, + { + "obr_head": "Welfare inside cap", + "pe_variables": ["universal_credit"], + "channel": "spending", + }, + ] + return { + "measure_key": KEY, + "title": "An announcement, see https://www.gov.uk/example", + "computability": "partial", + "classification": "partial", + "construction": "reversal_on_certified_world", + "commences_fy": "2024-25", + "pe_baseline_modifier": {"gov.tax.rate": modifier}, + "head_variables": sorted({v for h in heads for v in h["pe_variables"]}), + "heads": heads, + "source_rows": rows, + } + + +def simulation(values): + return { + "aggregates_gbp": {k: float(v) for k, v in values.items()}, + "variable_metadata": {k: {"entity": "person"} for k in values}, + "policyengine_bundle": { + "model_version": "x", + "certified_data_build_id": "build", + }, + "dataset_sha256_before": "a" * 64, + "dataset_sha256_after": "a" * 64, + "performance": {}, + } + + +def make_namespace( + tmp_path, + name, + *, + effects=None, + supported=(2024, 2025), + variant=None, + modifier=0.2, + heads=None, + skip=(), + identical=(), +): + """A complete namespace for one event; effects maps year -> {variable: gain}.""" + effects = effects or { + 2024: {"income_tax": 100e6, "universal_credit": -30e6}, + 2025: {"income_tax": 110e6, "universal_credit": -33e6}, + } + m = measure(modifier, heads) + registry = { + "event_slug": EVENT, + "event": EVENT, + "calendar_years": [2024, 2025], + "measures": [m], + } + registry_path = ( + tmp_path / "data" / "uk" / "events" / f"{EVENT}_{name}_measures.json" + ) + registry_path.parent.mkdir(parents=True, exist_ok=True) + registry_path.write_bytes(compute.canonical_bytes(registry)) + registry_sha = hashlib.sha256(registry_path.read_bytes()).hexdigest() + # The comparison loader binds to the canonical registry path. + canonical = tmp_path / "data" / "uk" / "events" / f"{EVENT}_measures.json" + canonical.write_bytes(registry_path.read_bytes()) + root = tmp_path / "ns" / name + event_dir = root / EVENT + event_dir.mkdir(parents=True) + channels = compute.variable_channels(m) + world = ab.worlds_for(m, {KEY: m}, years=[2024, 2025]) + variant_fields = ( + {"variant_id": variant, "variant_identity_sha256": "e" * 64} if variant else {} + ) + artifacts = {} + for year in supported: + if year in skip: + continue + gains = effects[year] + certified = {v: 1000e6 for v in channels} + reversed_world = { + v: certified[v] + (gains[v] if channels[v] == "tax" else -gains[v]) + for v in channels + } + if year in identical: + reversed_world = dict(certified) + artifact = compute.build_artifact( + measure=m, + world=world, + year=year, + baseline=simulation(reversed_world), + reform=simulation(certified), + certified=simulation(certified), + registry_sha256=registry_sha, + event=EVENT, + variant_fields=variant_fields, + ) + path = event_dir / f"{KEY}_{year}.json" + path.write_bytes(compute.canonical_bytes(artifact)) + artifacts[str(path.relative_to(tmp_path))] = hashlib.sha256( + path.read_bytes() + ).hexdigest() + manifest = { + "event": EVENT, + "registry_sha256": registry_sha, + "certified_dataset_sha256": "a" * 64, + "artifacts": artifacts, + "policyengine_version": "6.2.6" if variant else "5.0.2", + "engine_version": "2.125.1" if variant else "2.89.2", + "policyengine_core_version": "3.33.0" if variant else "3.27.1", + "data_bundle": "build", + } + if variant: + manifest.update( + variant_fields, + data_years=list(range(min(supported), 2031)), + supported_years=list(supported), + unsupported_years=[ + {"year": y, "reason": "before_data_year"} + for y in (2024, 2025) + if y not in supported + ], + full_supported_grid=[{"measure_key": KEY, "year": y} for y in supported], + full_supported_grid_complete=not skip, + ) + (event_dir / "RUN_MANIFEST.json").write_bytes(compute.canonical_bytes(manifest)) + rows, tally = stage.stage_event( + registry, manifest, artifact_dir=event_dir, registry_sha256=registry_sha + ) + staged = b"".join((json.dumps(r, sort_keys=True) + "\n").encode() for r in rows) + (event_dir / "STAGED.jsonl").write_bytes(staged) + (event_dir / "STAGING_MANIFEST.json").write_bytes( + compute.canonical_bytes( + { + **tally, + "event": EVENT, + "staged_path": f"ns/{name}/{EVENT}/STAGED.jsonl", + "staged_sha256": hashlib.sha256(staged).hexdigest(), + } + ) + ) + compare.write_comparison( + EVENT, + registry_path=canonical, + staged_path=event_dir / "STAGED.jsonl", + output_dir=event_dir, + artifact_root=tmp_path, + ) + return diff.load_side(root, artifact_root=tmp_path) + + +@pytest.fixture +def ns(tmp_path, monkeypatch): + monkeypatch.setattr(compute, "ROOT", tmp_path) + + def build(name, **kwargs): + return make_namespace(tmp_path, name, **kwargs) + + return build + + +def flags_by_key(result): + return { + (r["level"], r["key"], r["measure_key"], r["year"]): r["flags"] + for r in result["rows"] + } + + +def test_committed_legacy_namespace_has_no_flags_against_itself(tmp_path): + side = diff.load_side(diff.ROOT / "results" / "uk" / "events") + result = diff.diff_sides(side, side) + assert set(result["events"]) == { + "autumn_budget_2024", + "autumn_statement_2023", + "spring_budget_2024", + "spring_statement_2025", + "spring_budget_2023", + } + assert not any(r["flags"] for r in result["rows"]) + assert result["identity_changes"] == {} + files = diff.render(side, side) + diff.write(files, tmp_path / "out") + assert diff.check(files, tmp_path / "out") == [] + with open(tmp_path / "out" / "DIFF.json", "ab") as handle: + handle.write(b" ") + assert diff.check(files, tmp_path / "out") == ["DIFF.json"] + + +def test_identity_change_alone_raises_no_row_flags(ns): + result = diff.diff_sides(ns("a"), ns("b", variant="pe6.2.6_uk2.125.1_build")) + assert not any(r["flags"] for r in result["rows"]) + assert result["identity_changes"]["variant_id"]["head"] == "pe6.2.6_uk2.125.1_build" + assert result["identity_changes"]["policyengine_uk_version"] == { + "base": "2.89.2", + "head": "2.125.1", + } + + +def test_numerical_flag_only_strictly_above_the_threshold(ns): + base = ns("a") + within = ns( + "b", + effects={ + 2024: {"income_tax": 100e6 + 1e6, "universal_credit": -30e6}, + 2025: {"income_tax": 110e6, "universal_credit": -33e6}, + }, + ) + above = ns( + "c", + effects={ + 2024: {"income_tax": 100e6 + 1.5e6, "universal_credit": -30e6}, + 2025: {"income_tax": 110e6, "universal_credit": -33e6}, + }, + ) + assert not any(r["flags"] for r in diff.diff_sides(base, within)["rows"]) + flagged = flags_by_key(diff.diff_sides(base, above)) + assert flagged[("head", "income_tax", KEY, 2024)] == ["numerical"] + assert flagged[("measure_year", "total", KEY, 2024)] == ["numerical"] + assert flagged[("source_row", "tax_2024", KEY, 2024)] == ["numerical"] + assert flagged[("head", "universal_credit", KEY, 2024)] == [] + + +def test_sign_reversal_and_zero_crossing_are_structural_regardless_of_size(ns): + base = ns( + "a", + effects={ + 2024: {"income_tax": 0.2, "universal_credit": -30e6}, + 2025: {"income_tax": 0.4, "universal_credit": -33e6}, + }, + ) + head = ns( + "b", + effects={ + 2024: {"income_tax": 0.9, "universal_credit": -30e6}, + 2025: {"income_tax": -0.6, "universal_credit": -33e6}, + }, + ) + flagged = flags_by_key(diff.diff_sides(base, head)) + assert flagged[("head", "income_tax", KEY, 2024)] == ["structural"] + assert flagged[("head", "income_tax", KEY, 2025)] == ["structural"] + codes = { + (r["year"], d["code"]) + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "head" and r["key"] == "income_tax" + for d in r["details"] + } + assert codes == {(2024, "zero_crossing"), (2025, "zero_crossing")} + flipped = ns( + "c", + effects={ + 2024: {"income_tax": -5.0, "universal_credit": -30e6}, + 2025: {"income_tax": 0.4, "universal_credit": -33e6}, + }, + ) + base5 = ns( + "d", + effects={ + 2024: {"income_tax": 5.0, "universal_credit": -30e6}, + 2025: {"income_tax": 0.4, "universal_credit": -33e6}, + }, + ) + rows = diff.diff_sides(base5, flipped)["rows"] + assert any(d["code"] == "sign_reversal" for r in rows for d in r["details"]) + + +def test_unsupported_year_is_support_never_coverage(ns): + result = diff.diff_sides( + ns("a"), ns("b", supported=(2025,), variant="pe6_uk2_build") + ) + for record in result["rows"]: + if record["key"].startswith("corporation_"): + # Outside household scope in every year: support cannot move it. + assert record["flags"] == [] + elif record["year"] == 2024: + assert record["flags"] == ["support"], record + else: + assert record["flags"] == [] + assert result["flag_counts"]["coverage"] == 0 + assert result["flag_counts"]["support"] > 0 + + +def test_missing_artifact_in_a_supported_year_is_coverage(ns): + result = diff.diff_sides(ns("a"), ns("b", skip=(2025,), variant="pe6_uk2_build")) + flagged = flags_by_key(result) + assert flagged[("measure_year", "total", KEY, 2025)] == ["coverage"] + assert flagged[("source_row", "tax_2025", KEY, 2025)] == ["coverage"] + + +def test_newly_inert_construction_is_flagged(ns): + result = diff.diff_sides(ns("a"), ns("b", identical=(2025,))) + flagged = flags_by_key(result) + assert "inertness" in flagged[("measure_year", "total", KEY, 2025)] + assert "inertness" in flagged[("source_row", "tax_2025", KEY, 2025)] + + +def test_changed_construction_and_mapping_are_flagged(ns): + changed = flags_by_key(diff.diff_sides(ns("a"), ns("b", modifier=0.25))) + assert "construction" in changed[("measure_year", "total", KEY, 2024)] + heads = [ + { + "obr_head": "Income tax", + "pe_variables": ["income_tax", "national_insurance"], + "channel": "tax", + }, + { + "obr_head": "Welfare inside cap", + "pe_variables": ["universal_credit"], + "channel": "spending", + }, + ] + effects = { + y: {"income_tax": 100e6, "national_insurance": 0.0, "universal_credit": -30e6} + for y in (2024, 2025) + } + remapped = flags_by_key( + diff.diff_sides(ns("c"), ns("d", heads=heads, effects=effects)) + ) + assert "mapping" in remapped[("measure_year", "total", KEY, 2024)] + assert remapped[("head", "national_insurance", KEY, 2024)] == ["mapping"] + assert "mapping" in remapped[("source_row", "tax_2024", KEY, 2024)] + + +def test_premise_changes_flag_construction_and_current_law_is_context(ns): + base, head = ns("a"), ns("b") + premise = { + "events": { + EVENT: { + "measures": { + KEY: { + "checks": { + "reversal_differs_from_current_law": {"holds": True} + }, + "years": { + "2024": { + "paths": { + "gov.tax.rate": { + "current_law": 0.3, + "baseline_world": 0.2, + "reform_world": 0.3, + } + } + } + }, + } + } + } + } + } + base["premises"] = premise + + def record_for(head_premises): + head["premises"] = head_premises + return next( + r + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "measure_year" and r["year"] == 2024 + ) + + def changed(**values): + out = copy.deepcopy(premise) + out["events"][EVENT]["measures"][KEY]["years"]["2024"]["paths"][ + "gov.tax.rate" + ].update(values) + return out + + # The reform world of a reversal is current law: when current law moves, + # that world moves with it, which is context and not a construction change. + record = record_for(changed(current_law=0.35, reform_world=0.35)) + assert record["flags"] == [] + assert record["context"]["current_law_changed"] == { + "gov.tax.rate": {"base": 0.3, "head": 0.35} + } + # An executed override that changes (a re-derived counterfactual) is. + record = record_for(changed(baseline_world=0.25)) + assert record["flags"] == ["construction"] + assert [d["code"] for d in record["details"]] == ["baseline_world_value_changed"] + # So is an override on one side that the other side no longer applies. + record = record_for(changed(baseline_world=0.3)) + assert record["flags"] == ["construction"] + only_current_law = copy.deepcopy(premise) + only_current_law["events"][EVENT]["measures"][KEY]["years"]["2024"]["paths"][ + "gov.tax.rate" + ]["current_law"] = 0.4 + head["premises"] = only_current_law + record = next( + r + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "measure_year" and r["year"] == 2024 + ) + assert record["flags"] == [] + assert "current_law_changed" in record["context"] + flipped = copy.deepcopy(premise) + flipped["events"][EVENT]["measures"][KEY]["checks"][ + "reversal_differs_from_current_law" + ]["holds"] = False + head["premises"] = flipped + assert any( + d["code"] == "premise_check_changed" + for r in diff.diff_sides(base, head)["rows"] + for d in r["details"] + ) + + +def test_swapping_sides_swaps_values_and_keeps_flags(ns): + a = ns("a") + b = ns( + "b", + effects={ + 2024: {"income_tax": -40e6, "universal_credit": -30e6}, + 2025: {"income_tax": 0.0, "universal_credit": -60e6}, + }, + variant="pe6_uk2_build", + ) + forward = { + (r["level"], r["key"], r["year"]): r for r in diff.diff_sides(a, b)["rows"] + } + backward = { + (r["level"], r["key"], r["year"]): r for r in diff.diff_sides(b, a)["rows"] + } + assert forward.keys() == backward.keys() + for key, record in forward.items(): + other = backward[key] + assert record["flags"] == other["flags"] + assert (record["base_gbp"], record["head_gbp"]) == ( + other["head_gbp"], + other["base_gbp"], + ) + if record["delta_gbp"] is not None: + assert record["delta_gbp"] == -other["delta_gbp"] + + +def test_measure_year_change_equals_the_sum_of_head_changes(ns): + a = ns("a") + b = ns( + "b", + effects={ + 2024: {"income_tax": 123.456e6, "universal_credit": -7.25e6}, + 2025: {"income_tax": 1e9, "universal_credit": -2e9}, + }, + ) + rows = diff.diff_sides(a, b)["rows"] + for year in (2024, 2025): + total = next( + r for r in rows if r["level"] == "measure_year" and r["year"] == year + ) + heads = [r for r in rows if r["level"] == "head" and r["year"] == year] + assert total["delta_gbp"] == pytest.approx( + sum(h["delta_gbp"] for h in heads), abs=1e-3 + ) + + +def test_output_bytes_do_not_depend_on_input_order(ns): + a = ns("a") + b = ns( + "b", + effects={ + 2024: {"income_tax": 50e6, "universal_credit": -30e6}, + 2025: {"income_tax": 0.0, "universal_credit": -3e6}, + }, + ) + first = diff.render(a, b) + for seed in range(5): + rng = random.Random(seed) + shuffled = copy.copy(b) + shuffled["events"] = {} + for event, data in b["events"].items(): + data = dict(data) + items = list(data["rows"].items()) + rng.shuffle(items) + data["rows"] = dict(items) + artifacts = list(data["artifacts"].items()) + rng.shuffle(artifacts) + data["artifacts"] = dict(artifacts) + shuffled["events"][event] = data + assert diff.render(a, shuffled) == first + + +def test_packets_cover_flagged_moves_but_not_support_only_rows(ns): + a = ns("a") + b = ns( + "b", + supported=(2025,), + variant="pe6_uk2_build", + effects={2025: {"income_tax": 300e6, "universal_credit": -33e6}}, + ) + files = diff.render(a, b) + names = sorted(p for p in files if p.startswith("packets/") and p.endswith(".json")) + assert names == [f"packets/{EVENT}__tax_and_benefit__2025.json"] + packet = json.loads(files[names[0]]) + assert packet["flags"] == ["numerical"] + assert [c["source_row_id"] for c in packet["source_cells"]] == [ + "tax_2025", + "benefit_2025", + "corporation_2025", + ] + assert "--variant pe6_uk2_build" in packet["reproduce"]["head"] + assert "--variant pe5.0.2_uk2.89.2_build" in packet["reproduce"]["base"] + assert packet["cited_sources"] == ["https://www.gov.uk/example"] + assert packet["minimal_case"]["status"] == diff.PACKET_PENDING + assert "files no issues" in packet["filing"] + markdown = files["DIFF.md"].decode() + assert "Years supported on one side only" in markdown + assert f"| {EVENT} | 2024 | source_row | 2 |" in markdown + + +def test_unverifiable_side_is_refused(ns, tmp_path): + ns("a") + staged = tmp_path / "ns" / "a" / EVENT / "STAGED.jsonl" + staged.write_bytes(staged.read_bytes() + b"\n") + with pytest.raises(diff.DiffError, match="does not verify"): + diff.load_side(tmp_path / "ns" / "a", artifact_root=tmp_path) + + +finite = st.floats(min_value=-1e13, max_value=1e13, allow_nan=False) + + +@settings(max_examples=500, deadline=None) +@given(finite, finite) +def test_threshold_is_symmetric_and_strict(base, head): + assert diff.numerically_moved(base, head) == diff.numerically_moved(head, base) + threshold = diff.numerical_threshold(base, head) + assert threshold >= diff.NUMERICAL_FLOOR_GBP + assert diff.numerically_moved(base, head) == (abs(head - base) > threshold) + + +@settings(max_examples=300, deadline=None) +@given(st.floats(min_value=0, max_value=1e12, allow_nan=False)) +def test_no_flag_at_the_threshold_one_just_above(magnitude): + threshold = diff.numerical_threshold(magnitude, magnitude) + assert not diff.numerically_moved(magnitude, magnitude) + # A move of exactly the floor never flags while the floor dominates. + if magnitude < 99e6: + assert not diff.numerically_moved( + magnitude, magnitude + diff.NUMERICAL_FLOOR_GBP + ) + assert diff.numerically_moved( + magnitude, magnitude + 1.000001 * diff.NUMERICAL_FLOOR_GBP + ) + assert threshold == max(diff.NUMERICAL_FLOOR_GBP, 0.01 * magnitude) + + +@settings(max_examples=300, deadline=None) +@given(finite, finite) +def test_sign_details_are_symmetric(base, head): + assert [d["code"] for d in diff.sign_details(base, head)] == [ + d["code"] for d in diff.sign_details(head, base) + ] diff --git a/tests/test_uk_replay_qualify.py b/tests/test_uk_replay_qualify.py new file mode 100644 index 0000000..1a9516c --- /dev/null +++ b/tests/test_uk_replay_qualify.py @@ -0,0 +1,340 @@ +"""Qualification requests, the persistent queue, receipts and seals. + +Engine-free: environments, simulations and downloads are not run; the +pieces that decide identity, rejection and sealing are exercised directly. +""" + +import copy +import hashlib +import json +from pathlib import Path + +import jsonschema +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from pipeline import uk_replay_interface as interface +from pipeline import uk_replay_qualify as qualify +from pipeline import uk_replay_queue as queue +from pipeline import uk_replay_variants as rv + +FIXTURES = Path(__file__).parent / "fixtures" / "uk_replay" +CURRENT = "pe6.2.6_uk2.125.1_policyengine-uk-data-1.58.0" +RUN = "https://github.com/PolicyEngine/policyengine-scorecard/actions/runs/1" + + +def golden_request(): + return json.loads((FIXTURES / "qualification_request_golden.json").read_text()) + + +# --- the queue -------------------------------------------------------------------- + + +def test_enqueue_is_idempotent_and_records_the_request(tmp_path): + request = golden_request() + rid, status = queue.enqueue( + tmp_path, request, run_url=RUN, now="2026-10-11T12:00:00Z" + ) + assert status == "queued" and rid == interface.request_id(request) + stored = json.loads((tmp_path / "requests" / rid / "REQUEST.json").read_text()) + assert stored == request + again = queue.enqueue(tmp_path, request, run_url=RUN, now="2026-10-11T13:00:00Z") + assert again == (rid, "queued") + status_doc = queue.read_status(tmp_path, rid) + assert len(status_doc["history"]) == 1 + + +def test_an_invalid_request_is_still_recorded_for_its_rejection(tmp_path): + request = {"schema": "nonsense"} + rid, status = queue.enqueue( + tmp_path, request, run_url=RUN, now="2026-10-11T12:00:00Z" + ) + assert rid == interface.jcs_sha256(request) and status == "queued" + + +def test_status_transitions_and_terminal_states(tmp_path): + rid, _ = queue.enqueue( + tmp_path, golden_request(), run_url=RUN, now="2026-10-11T12:00:00Z" + ) + queue.set_status(tmp_path, rid, "running", run_url=RUN, now="2026-10-11T12:01:00Z") + with pytest.raises(ValueError, match="terminal status names its receipt"): + queue.set_status( + tmp_path, rid, "complete", run_url=RUN, now="2026-10-11T12:30:00Z" + ) + done = queue.set_status( + tmp_path, + rid, + "complete", + run_url=RUN, + now="2026-10-11T12:30:00Z", + receipt_sha256="a" * 64, + receipt_commit="b" * 40, + ) + assert [h["status"] for h in done["history"]] == ["queued", "running", "complete"] + interface.validate(done, interface.STATUS_SCHEMA_ID) + with pytest.raises(ValueError, match="terminal states are final"): + queue.set_status( + tmp_path, rid, "running", run_url=RUN, now="2026-10-11T13:00:00Z" + ) + + +def test_pending_is_oldest_first_and_recovers_lost_runs(tmp_path): + first = golden_request() + second = copy.deepcopy(first) + second["attempt"] = 2 + third = copy.deepcopy(first) + third["attempt"] = 3 + a, _ = queue.enqueue(tmp_path, second, run_url=RUN, now="2026-10-11T10:00:00Z") + b, _ = queue.enqueue(tmp_path, first, run_url=RUN, now="2026-10-11T09:00:00Z") + c, _ = queue.enqueue(tmp_path, third, run_url=RUN, now="2026-10-11T11:00:00Z") + queue.set_status(tmp_path, c, "running", run_url=RUN, now="2026-10-11T11:05:00Z") + assert queue.pending(tmp_path, now="2026-10-11T12:00:00Z") == [b, a] + # A running request whose job vanished is retried after the stale bound. + assert queue.pending(tmp_path, now="2026-10-11T18:06:00Z") == [b, a, c] + + +def test_request_ids_must_be_digests(tmp_path): + with pytest.raises(ValueError): + queue.request_dir(tmp_path, "../escape") + + +# --- requests to environments ----------------------------------------------------- + + +def test_golden_candidate_matches_the_committed_certified_pair(): + document = rv.load_document() + match = qualify.matching_certified(golden_request(), document) + assert match is not None and match["variant_id"] == CURRENT + other = golden_request() + other["candidate"]["policyengine_uk"]["wheel_sha256"] = "0" * 64 + assert qualify.matching_certified(other, document) is None + assert qualify.latest_certified(document)["variant_id"] == CURRENT + + +def test_install_uses_exact_pins_for_a_certified_pair_and_constraints_otherwise( + tmp_path, +): + wheels = tmp_path / "wheels" + wheels.mkdir() + (wheels / "policyengine_uk-2.125.1-py3-none-any.whl").write_bytes(b"x") + request = golden_request() + env = tmp_path / "env.txt" + exact = qualify.install_command( + request, tmp_path / "venv", wheels, environment=env, exact=True + ) + assert exact[exact.index(str(env)) - 1] == "-r" + loose = qualify.install_command( + request, tmp_path / "venv", wheels, environment=env, exact=False + ) + assert loose[loose.index(str(env)) - 1] == "-c" + assert any( + part.startswith("policyengine @ git+") + and part.endswith( + request["candidate"]["policyengine"]["source"]["git"]["commit"] + ) + for part in exact + ) + lines = qualify.environment_lines(rv.get_variant(CURRENT)) + assert lines and not any(line.split("==")[0] in rv.PACKAGES for line in lines) + + +def test_downloaded_wheels_must_match_the_request(tmp_path): + request = golden_request() + wheels = tmp_path / "wheels" + wheels.mkdir() + (wheels / "policyengine_uk-2.125.1-py3-none-any.whl").write_bytes(b"not the wheel") + (wheels / "policyengine_core-3.33.0-py3-none-any.whl").write_bytes(b"core") + with pytest.raises(qualify.Rejected) as raised: + qualify.verify_wheels(request, wheels) + assert raised.value.code == "artifact_hash_mismatch" + request["candidate"]["policyengine_uk"]["wheel_sha256"] = hashlib.sha256( + b"not the wheel" + ).hexdigest() + request["candidate"]["policyengine_core"]["wheel_sha256"] = hashlib.sha256( + b"core" + ).hexdigest() + digests = qualify.verify_wheels(request, wheels) + assert set(digests) == {"policyengine_uk", "policyengine_core"} + + +def test_schema_failures_are_rejections(): + with pytest.raises(qualify.Rejected) as raised: + qualify.request_id({"schema": "x"}) + assert raised.value.code == "schema_invalid" + + +def test_freeze_normalisation(): + versions = { + "policyengine": "6.2.6", + "policyengine-uk": "2.125.1", + "policyengine-core": "3.33.0", + } + freeze = "\n".join( + [ + "numpy==2.5.4", + "policyengine @ git+https://github.com/PolicyEngine/policyengine.py@" + + "a" * 40, + "policyengine-core @ file:///tmp/wheels/policyengine_core-3.33.0-py3-none-any.whl", + "policyengine-uk==2.125.1", + ] + ) + text = rv.normalized_freeze(freeze, versions) + assert rv.requirements_pins(text) == { + "numpy": "2.5.4", + "policyengine": "6.2.6", + "policyengine-core": "3.33.0", + "policyengine-uk": "2.125.1", + } + assert ( + "# policyengine-core installed from policyengine_core-3.33.0-py3-none-any.whl" + in text + ) + with pytest.raises(rv.VariantError, match="not an exact pin"): + rv.normalized_freeze(freeze + "\nother @ file:///x.whl", versions) + with pytest.raises(rv.VariantError, match="lacks"): + rv.normalized_freeze("numpy==2.5.4", versions) + + +def test_variant_ids_take_a_digest_suffix_only_on_a_collision(): + identity = rv.get_variant(CURRENT)["identity"] + digest = rv.identity_sha256(identity) + slug = rv.variant_slug(identity) + assert rv.variant_id_for(identity, {}) == slug + assert rv.variant_id_for(identity, {slug: digest}) == slug + assert rv.variant_id_for(identity, {slug: "f" * 64}) == f"{slug}--{digest[:10]}" + + +def test_every_pinned_distribution_must_be_installed_at_its_version(): + pins = {"numpy": "2.5.4", "pandas": "3.0.6"} + assert ( + rv.environment_mismatches( + pins, {"distributions": {"numpy": "2.5.4", "pandas": "3.0.6", "extra": "1"}} + ) + == [] + ) + assert rv.environment_mismatches( + pins, {"distributions": {"numpy": "2.5.5", "pandas": "3.0.6"}} + ) + assert rv.environment_mismatches(pins, {"distributions": {"numpy": "2.5.4"}}) + + +# --- receipts and seals ------------------------------------------------------------- + + +def test_rejected_and_preidentity_failed_receipts_validate(): + request = golden_request() + rejected = interface.seal_receipt( + qualify.build_receipt( + request, + record=None, + rejection=qualify.Rejected("data_access_failed", "401"), + results_path="requests/x/results/", + ) + ) + interface.validate(rejected, interface.RECEIPT_SCHEMA_ID) + assert rejected["status"] == "rejected" and rejected["corpus"] is None + failed = interface.seal_receipt( + qualify.build_receipt( + request, + record=None, + failures=[ + {"code": "qualification_step_failed", "detail": "pip download: exit 1"} + ], + results_path="requests/x/results/", + ) + ) + interface.validate(failed, interface.RECEIPT_SCHEMA_ID) + assert failed["status"] == "failed" and failed["integrity"]["status"] == "failed" + assert failed["execution"] is None + + +def test_a_complete_receipt_needs_its_execution(): + receipt = interface.seal_receipt( + { + **json.loads((FIXTURES / "replay_receipt_golden.json").read_text()), + "execution": None, + } + ) + with pytest.raises(jsonschema.ValidationError): + interface.validate(receipt, interface.RECEIPT_SCHEMA_ID) + + +def sealed_namespace(tmp_path): + namespace = tmp_path / "ns" + (namespace / "event").mkdir(parents=True) + (namespace / "event" / "RUN_MANIFEST.json").write_text("{}") + (namespace / "SUMMARY.md").write_text("# summary\n") + diff = tmp_path / "diff" + (diff / "packets").mkdir(parents=True) + (diff / "DIFF.json").write_text("{}") + golden = json.loads((FIXTURES / "replay_receipt_golden.json").read_text()) + receipt = {k: v for k, v in golden.items() if k != "receipt_sha256"} + receipt["results"] = {k: v for k, v in receipt["results"].items() if k != "commit"} + sealed = qualify.seal(namespace, receipt, diff_dir=diff) + return namespace, diff, sealed + + +def test_seal_binds_every_file_and_detects_tampering(tmp_path): + namespace, diff, sealed = sealed_namespace(tmp_path) + assert qualify.verify_sealed(namespace, diff_dir=diff) == [] + manifest = json.loads((namespace / qualify.MANIFEST_NAME).read_text()) + assert set(manifest["files"]) == {"event/RUN_MANIFEST.json", "SUMMARY.md"} + assert ( + sealed["results"]["manifest_sha256"] + == hashlib.sha256((namespace / qualify.MANIFEST_NAME).read_bytes()).hexdigest() + ) + (namespace / "SUMMARY.md").write_text("# edited\n") + assert qualify.verify_sealed(namespace, diff_dir=diff) + + +def test_seal_detects_added_files_and_diff_edits(tmp_path): + namespace, diff, _ = sealed_namespace(tmp_path) + (namespace / "event" / "extra.json").write_text("{}") + assert any( + "files differ" in p for p in qualify.verify_sealed(namespace, diff_dir=diff) + ) + (namespace / "event" / "extra.json").unlink() + (diff / "DIFF.json").write_text("{ }") + assert any("diff" in p for p in qualify.verify_sealed(namespace, diff_dir=diff)) + + +def test_sealed_namespaces_refuse_writes(tmp_path): + namespace, _, _ = sealed_namespace(tmp_path) + with pytest.raises(rv.VariantError, match="sealed"): + rv.assert_not_sealed(namespace / "event", root=tmp_path) + + +@settings(max_examples=50, deadline=None) +@given(st.lists(st.sampled_from(["queued", "running"]), min_size=1, max_size=5)) +def test_pending_never_returns_terminal_requests(tmp_path_factory, statuses): + state = tmp_path_factory.mktemp("state") + ids = [] + for attempt, status in enumerate(statuses, start=1): + request = golden_request() + request["attempt"] = attempt + rid, _ = queue.enqueue( + state, request, run_url=RUN, now=f"2026-10-11T0{attempt}:00:00Z" + ) + if status == "running": + queue.set_status( + state, rid, "running", run_url=RUN, now=f"2026-10-11T0{attempt}:10:00Z" + ) + ids.append(rid) + finished = ids[0] + if queue.read_status(state, finished)["status"] == "queued": + queue.set_status( + state, finished, "running", run_url=RUN, now="2026-10-11T09:00:00Z" + ) + queue.set_status( + state, + finished, + "failed", + run_url=RUN, + now="2026-10-11T09:30:00Z", + receipt_sha256="c" * 64, + receipt_commit="d" * 40, + ) + later = queue.pending(state, now="2026-10-12T09:00:00Z") + assert finished not in later + assert set(later) == set(ids[1:]) diff --git a/tests/test_uk_replay_variants.py b/tests/test_uk_replay_variants.py index 7637a64..0e0ba24 100644 --- a/tests/test_uk_replay_variants.py +++ b/tests/test_uk_replay_variants.py @@ -178,9 +178,37 @@ def installed_for(r): }, "packaged_bundle_manifest_sha256": identity["packaged_bundle_manifest_sha256"], "python": identity["python"], + "distributions": rv.requirements_pins( + (rv.ROOT / r["requirements"]["path"]).read_text() + ), } +def test_every_pinned_distribution_is_part_of_the_installed_check(): + r = record() + exact = installed_for(r) + for name in ("numpy", "pandas", "h5py"): + changed = copy.deepcopy(exact) + changed["distributions"][name] = "0.0.0" + with pytest.raises(rv.VariantError, match=name): + rv.verify_installed(r, installed=changed) + missing = copy.deepcopy(exact) + del missing["distributions"][name] + with pytest.raises(rv.VariantError, match=name): + rv.verify_installed(r, installed=missing) + extra = copy.deepcopy(exact) + extra["distributions"]["rfc8785"] = "0.1.4" + rv.verify_installed(r, installed=extra) + + +def test_the_identity_binds_the_requirements_freeze(): + r = record() + r["identity"]["requirements_sha256"] = "5" * 64 + reseal(r) + with pytest.raises(rv.VariantError, match="another requirements file"): + rv.validate_record(r) + + def test_installed_environment_must_match_every_field(): r = record() exact = installed_for(r)