From a70d02dcc9634b8881e0223ded1f7e9ff1e8efc3 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 11:51:19 -0400 Subject: [PATCH 1/5] Add the release diff for event replays and its issue packets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pipeline/diff_uk_event_runs.py compares two complete replay namespaces after re-verifying each side's staging, comparison and artifact receipts. It flags, at source-row, measure-year and PE-head level: numerical moves above max(£1m, 1% of the larger magnitude); sign reversals and zero crossings; coverage, mapping, support, inertness and construction changes regardless of size. A year one side cannot support is a support flag, never coverage, and only on rows that could carry a counterpart. Ratio bins are reported as transitions. Flags are investigation prompts; nothing passes or fails. Each flagged (event, measure, year) gets a packet with the OBR source cells, both execution identities, artifact digests, executed reform dictionaries, focused rerun commands and the sources the registry cites. --check re-renders and compares bytes for CI. Co-Authored-By: Claude Opus 5.5 --- pipeline/diff_uk_event_runs.py | 1008 ++++++++++++++++++++++++++++++++ tests/test_uk_event_diff.py | 599 +++++++++++++++++++ 2 files changed, 1607 insertions(+) create mode 100644 pipeline/diff_uk_event_runs.py create mode 100644 tests/test_uk_event_diff.py diff --git a/pipeline/diff_uk_event_runs.py b/pipeline/diff_uk_event_runs.py new file mode 100644 index 0000000..edba223 --- /dev/null +++ b/pipeline/diff_uk_event_runs.py @@ -0,0 +1,1008 @@ +"""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"): + if 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/tests/test_uk_event_diff.py b/tests/test_uk_event_diff.py new file mode 100644 index 0000000..388bd02 --- /dev/null +++ b/tests/test_uk_event_diff.py @@ -0,0 +1,599 @@ +"""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 + moved = copy.deepcopy(premise) + moved["events"][EVENT]["measures"][KEY]["years"]["2024"]["paths"][ + "gov.tax.rate" + ].update(current_law=0.35, reform_world=0.35) + head["premises"] = moved + record = next( + r + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "measure_year" and r["year"] == 2024 + ) + assert record["flags"] == ["construction"] + assert [d["code"] for d in record["details"]] == ["reform_world_value_changed"] + assert record["context"]["current_law_changed"] == { + "gov.tax.rate": {"base": 0.3, "head": 0.35} + } + 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) + ] From 062093fe38a82581ce803a0db259bac532f80d65 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 16:06:49 -0400 Subject: [PATCH 2/5] Diff: a world that is current law on both sides is context, not construction On the first real premise diff (2.89.2 to 2.125.1) every reversal's reform world, which is current law, moved with current law: the employer-NIC secondary threshold (96 to 96.153846), the Class 2 small-profits threshold, float noise in the blended CGT rates. Those were raised as construction flags, which release qualification proposes to fail on. A baseline or reform world value change is now a construction flag only where one side's world overrides that side's current law; otherwise the move is recorded in current_law_changed context. Premise-check changes (the CGT annualisation premise) still flag. Co-Authored-By: Claude Opus 5.5 --- pipeline/diff_uk_event_runs.py | 7 +++++- tests/test_uk_event_diff.py | 39 +++++++++++++++++++++++----------- 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/pipeline/diff_uk_event_runs.py b/pipeline/diff_uk_event_runs.py index edba223..03b86bd 100644 --- a/pipeline/diff_uk_event_runs.py +++ b/pipeline/diff_uk_event_runs.py @@ -389,7 +389,12 @@ def measure(premises): for path in sorted(set(by) | set(hy)): bp, hp = by.get(path, {}), hy.get(path, {}) for world in ("baseline_world", "reform_world"): - if bp.get(world) != hp.get(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", diff --git a/tests/test_uk_event_diff.py b/tests/test_uk_event_diff.py index 388bd02..07ec8c6 100644 --- a/tests/test_uk_event_diff.py +++ b/tests/test_uk_event_diff.py @@ -414,21 +414,36 @@ def test_premise_changes_flag_construction_and_current_law_is_context(ns): } } base["premises"] = premise - moved = copy.deepcopy(premise) - moved["events"][EVENT]["measures"][KEY]["years"]["2024"]["paths"][ - "gov.tax.rate" - ].update(current_law=0.35, reform_world=0.35) - head["premises"] = moved - record = next( - r - for r in diff.diff_sides(base, head)["rows"] - if r["level"] == "measure_year" and r["year"] == 2024 - ) - assert record["flags"] == ["construction"] - assert [d["code"] for d in record["details"]] == ["reform_world_value_changed"] + + 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" From 41586ddbac7105f436d9675713eb3a79600c5868 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 20:26:31 -0400 Subject: [PATCH 3/5] Revalidate release diff inputs and preserve investigation evidence Co-Authored-By: GPT-6.1 Sol (Subfleet lane) --- pipeline/compare_uk_event.py | 14 +- pipeline/diff_uk_event_runs.py | 340 ++++++++++++++++----- tests/test_uk_event_diff.py | 533 ++++++++++++++++++++++++++++++++- 3 files changed, 806 insertions(+), 81 deletions(-) diff --git a/pipeline/compare_uk_event.py b/pipeline/compare_uk_event.py index c5bf710..1c84953 100644 --- a/pipeline/compare_uk_event.py +++ b/pipeline/compare_uk_event.py @@ -1319,8 +1319,14 @@ def render_summary( return "\n".join(lines) -def load_verified_comparison(path: Path, *, artifact_root: Path = ROOT) -> list[dict]: - """Bind summary-only reads to current source, stage, axes and artifact bytes.""" +def load_verified_comparison( + path: Path, *, artifact_root: Path = ROOT, registry_path: Path | None = None +) -> list[dict]: + """Verify receipts against the current registry or an explicit snapshot. + + Historical readers may select the receipt-bound registry snapshot. The + default preserves summaries' binding to the current event registry. + """ provenance_path = path.parent / "COMPARISON_PROVENANCE.json" if not provenance_path.exists(): raise EventComparisonError("summary requires COMPARISON_PROVENANCE.json") @@ -1352,7 +1358,9 @@ def load_verified_comparison(path: Path, *, artifact_root: Path = ROOT) -> list[ raise EventComparisonError("comparison replay-grid receipt differs") if registry.get("event_slug") != path.parent.name: raise EventComparisonError("summary registry event differs") - canonical_registry = ( + if registry_path is not None and registry_path.resolve() != paths["registry"]: + raise EventComparisonError("comparison registry snapshot path differs") + canonical_registry = registry_path or ( artifact_root / "data" / "uk" / "events" / f"{path.parent.name}_measures.json" ) if ( diff --git a/pipeline/diff_uk_event_runs.py b/pipeline/diff_uk_event_runs.py index 03b86bd..f4c3006 100644 --- a/pipeline/diff_uk_event_runs.py +++ b/pipeline/diff_uk_event_runs.py @@ -18,11 +18,20 @@ 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; +* mapping: changed head variables, channels, definitions or registry digest; * 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. + +Invariants: each side uses its receipt-bound registry and staging manifest; +every inventory artifact is revalidated, and the run's identity, support and +grid reconcile with regenerated staging rows and tallies. Identical sides +have no flags. Swapping sides preserves flag sets and mirrors values. Every +source row on either side appears once at source_row level; coverage requires +shared year support, and rows incapable of counterparts never get support +flags. Every flagged record belongs to exactly one measure-year packet, +including support records. Input ordering cannot change output bytes. """ from __future__ import annotations @@ -32,6 +41,7 @@ import json import math import re +import shlex import sys from collections import defaultdict from pathlib import Path @@ -42,6 +52,7 @@ sys.path.insert(0, str(ROOT)) from pipeline import compare_uk_event as compare # noqa: E402 +from pipeline import stage_uk_event as stage # noqa: E402 from pipeline.compare_uk_obr_costings import atomic_write_bytes # noqa: E402 NUMERICAL_FLOOR_GBP = 1_000_000.0 @@ -57,7 +68,7 @@ "inertness", "construction", ) -PACKET_FLAGS = frozenset(FLAGS) - {"support"} +PACKET_FLAGS = frozenset(FLAGS) COMPUTABLE_CLASSES = ("expressible", "partial") LEVELS = ("source_row", "measure_year", "head") DIFF_ROOT = ROOT / "results" / "uk" / "replays" / "diffs" @@ -150,9 +161,148 @@ def resolve_root(value: str, *, artifact_root: Path = ROOT) -> Path: raise DiffError(f"{value!r} is neither a replay namespace nor a variant id") +def _revalidate_run( + event_dir: Path, manifest: dict, provenance: dict, *, artifact_root: Path +) -> tuple[dict, dict, dict, Path]: + """Reconcile unbound run metadata with the receipt-bound staging inputs.""" + registry_path = compare._receipt_path(provenance["registry_path"], artifact_root) + registry = _json(registry_path) + registry_years = sorted(registry["calendar_years"]) + if ( + "registry_path" in manifest + and compare._receipt_path(manifest["registry_path"], artifact_root) + != registry_path + ): + raise DiffError("run manifest registry path differs from verified snapshot") + if manifest.get("variant_id") is not None: + data_years = manifest["data_years"] + if ( + not isinstance(data_years, list) + or not data_years + or any(type(year) is not int for year in data_years) + or data_years != sorted(set(data_years)) + ): + raise DiffError("run manifest data years must be nonempty and ordered") + supported, unsupported = stage.compute.variants.supported_years( + manifest, registry_years + ) + if ( + manifest["supported_years"] != supported + or manifest["unsupported_years"] != unsupported + ): + raise DiffError("run manifest year support differs from its data years") + elif manifest.get( + "supported_years", registry_years + ) != registry_years or manifest.get("unsupported_years", []): + raise DiffError("legacy run manifest year support differs from registry") + + # stage_event resolves inventory references against compute.ROOT. Absolute + # paths let it validate a separate artifact root without changing globals; + # restore receipt-relative paths before comparing its rows with STAGED. + inventory = { + str(compare._receipt_path(relative, artifact_root)): relative + for relative in manifest["artifacts"] + } + if len(inventory) != len(manifest["artifacts"]): + raise DiffError("run manifest repeats an artifact path") + staging_manifest = dict(manifest) + staging_manifest["artifacts"] = { + absolute: manifest["artifacts"][relative] + for absolute, relative in inventory.items() + } + rows, tally = stage.stage_event( + registry, + staging_manifest, + artifact_dir=event_dir, + registry_sha256=provenance["registry_sha256"], + ) + pairs = set() + for absolute in inventory: + artifact = _json(Path(absolute)) + pairs.add((artifact["measure_key"], artifact["year"])) + planned = stage.compute.worlds.computable( + {measure["measure_key"]: measure for measure in registry["measures"]}, + years=registry_years, + ) + full_pairs = { + (key, year) + for key, world in planned.items() + if "alias_of" not in world + for year in registry_years + } + supported_pairs = { + pair + for pair in full_pairs + if pair[1] in manifest.get("supported_years", registry_years) + } + for field, expected in ( + ("artifact_grid", pairs), + ("full_event_grid", full_pairs), + ("full_supported_grid", supported_pairs), + ): + if field in manifest and manifest[field] != stage.compute.grid_rows(expected): + raise DiffError(f"run manifest {field} differs from verified grid") + for field, expected in ( + ("years", sorted({year for _, year in pairs})), + ("measures", sorted({key for key, _ in pairs})), + ("full_event_complete", full_pairs <= pairs), + ("full_supported_grid_complete", supported_pairs <= pairs), + ): + if field in manifest and manifest[field] != expected: + raise DiffError(f"run manifest {field} differs from verified inventory") + for row in rows: + if row.get("artifact_path"): + row["artifact_path"] = inventory[row["artifact_path"]] + staged_path = compare._receipt_path(provenance["staged_path"], artifact_root) + if rows != compare.load_jsonl(staged_path): + raise DiffError("run manifest staging rows differ from verified staging") + staging_path = compare._receipt_path( + provenance["staging_manifest_path"], artifact_root + ) + staging = _json(staging_path) + if staging.get("variant_id") != manifest.get("variant_id"): + raise DiffError("run manifest variant differs from verified staging") + if staging.get("event") != event_dir.name or any( + staging.get(field) != value for field, value in tally.items() + ): + raise DiffError("run manifest staging tally differs from verified staging") + if staging.get("inert_measure_years", []) != tally.get("inert_measure_years", []): + raise DiffError("verified staging inertness differs from run artifacts") + + artifacts = {} + for absolute, relative in sorted(inventory.items()): + payload = Path(absolute).read_bytes() + digest = manifest["artifacts"][relative] + if sha256_bytes(payload) != digest: + raise DiffError(f"{relative}: artifact bytes differ from RUN_MANIFEST") + artifact = json.loads(payload) + compare.validate_artifact(artifact, relative) + for field in ( + "policyengine_version", + "engine_version", + "policyengine_core_version", + "data_bundle", + ): + if artifact.get(field) != manifest[field]: + raise DiffError(f"{relative}: {field} differs from run manifest") + for field in ("variant_id", "variant_identity_sha256"): + if artifact.get(field) != manifest.get(field): + raise DiffError(f"{relative}: {field} differs from run manifest") + for field in ("dataset_sha256_before", "dataset_sha256_after"): + if artifact.get(field) != manifest["certified_dataset_sha256"]: + raise DiffError(f"{relative}: {field} differs from certified digest") + artifacts[(artifact["measure_key"], artifact["year"])] = { + "path": relative, + "sha256": digest, + "artifact": artifact, + } + return registry, staging, artifacts, staging_path + + def load_side(root: Path, *, artifact_root: Path = ROOT) -> dict: """Every event of a namespace, re-verified through its receipts.""" root = root.resolve() + artifact_root = artifact_root.resolve() events = {} identities = set() for event_dir in sorted(p for p in root.iterdir() if p.is_dir()): @@ -160,40 +310,35 @@ def load_side(root: Path, *, artifact_root: Path = ROOT) -> dict: 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: + manifest = _json(manifest_path) + if manifest.get("event") != event: + raise DiffError(f"{manifest_path}: event differs from its directory") + provenance_path = event_dir / "COMPARISON_PROVENANCE.json" + provenance = _json(provenance_path) + registry_path = compare._receipt_path( + provenance["registry_path"], artifact_root + ) rows = compare.load_verified_comparison( - event_dir / "COMPARISON.json", artifact_root=artifact_root + event_dir / "COMPARISON.json", + artifact_root=artifact_root, + registry_path=registry_path, + ) + registry, staging, artifacts, staging_path = _revalidate_run( + event_dir, manifest, provenance, artifact_root=artifact_root ) - except (compare.EventComparisonError, FileNotFoundError, ValueError) as exc: + identity = side_identity(manifest) + except (ValueError, KeyError, TypeError, OSError) 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_path": provenance["registry_path"], "registry_sha256": provenance["registry_sha256"], "registry_years": registry_years, "supported_years": sorted(supported), @@ -217,6 +362,7 @@ def load_side(root: Path, *, artifact_root: Path = ROOT) -> dict: premises = _json(premises_path) if premises_path.exists() else None return { "root": root, + "artifact_root": artifact_root, "identity": json.loads(identities.pop()), "events": events, "premises": premises, @@ -284,7 +430,7 @@ def compare_source_rows(event: str, base: dict | None, head: dict | None) -> lis 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 not details and supported_both: if b is None or h is None: details.append( { @@ -434,7 +580,11 @@ def compare_artifacts( bv = b["measure_total_gbp"] if b else None hv = h["measure_total_gbp"] if h else None head_records = [] - if not details: + supported_both = all( + side is not None and year in side["supported_years"] + for side in (base, head) + ) + if not details and supported_both: if b is None or h is None: details.append( { @@ -555,12 +705,29 @@ def diff_sides(base: dict, head: dict) -> dict: event_records = compare_source_rows(event, b, h) + compare_artifacts( event, b, h, premises ) + registry_changed = bool( + b and h and b["registry_sha256"] != h["registry_sha256"] + ) + if registry_changed: + for record in event_records: + record["details"].append( + { + "flag": "mapping", + "code": "registry_sha256_changed", + "base": b["registry_sha256"], + "head": h["registry_sha256"], + } + ) + record["details"].sort( + key=lambda detail: (FLAGS.index(detail["flag"]), detail["code"]) + ) + record["flags"] = sorted( + set(record["flags"]) | {"mapping"}, key=FLAGS.index + ) 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"] - ), + "registry_changed": registry_changed, "supported_years": { "base": b["supported_years"] if b else None, "head": h["supported_years"] if h else None, @@ -626,12 +793,30 @@ def _cited_sources(measure: dict | None) -> list[str]: def _reproduce(side: dict, event: str, measure_key: str, year: int) -> str: + data = side["events"].get(event) + if data is None: + return "# This event is absent from this namespace." 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}" - ) + registry = (side["artifact_root"] / data["registry_path"]).resolve() + arguments = [ + "python", + "pipeline/compute_uk_event.py", + "--event", + event, + "--variant", + variant, + "--registry", + str(registry), + "--measures", + measure_key, + "--years", + str(year), + "--workers", + "1", + "--output-dir", + f".venv-replay-checks/reproductions/{variant}/{event}/{measure_key}_{year}", + ] + return "PYTHONPATH=. " + shlex.join(arguments) def build_packets(diff: dict, base: dict, head: dict) -> dict[str, dict]: @@ -644,42 +829,52 @@ def build_packets(diff: dict, base: dict, head: dict) -> dict[str, dict]: 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 - ) + measures = { + side: next( + ( + measure + for measure in (data or {}).get("registry", {}).get("measures", []) + if measure["measure_key"] == measure_key + ), + None, + ) + for side, data in sides.items() + } + measure = measures["head"] or measures["base"] 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 [], - } - ) + for side, side_measure in measures.items(): + for source in (side_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( + { + "side": side, + "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)) @@ -713,6 +908,7 @@ def premises_for(side_premises): "event": event, "measure_key": measure_key, "title": (measure or {}).get("title"), + "measures": measures, "year": year, "fy": f"{year}-{str(year + 1)[-2:]}", "flags": sorted({f for r in records for f in r["flags"]}, key=FLAGS.index), @@ -732,7 +928,11 @@ def premises_for(side_premises): side: _reproduce(data, event, measure_key, year) for side, data in (("base", base), ("head", head)) }, - "cited_sources": _cited_sources(measure), + "cited_sources": [ + {"side": side, "url": url} + for side, side_measure in measures.items() + for url in _cited_sources(side_measure) + ], "minimal_case": {"status": PACKET_PENDING}, "statutory_evidence": {"status": PACKET_PENDING}, "filing": ( @@ -782,12 +982,12 @@ def render_packet_markdown(name: str, packet: dict) -> str: "", "## OBR source cells", "", - "| Source row | Table / column | FY | Head | OBR £m | Base £m | Head £m |", - "|---|---|---|---|---:|---:|---:|", + "| Side | 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"| {cell['side']} | {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'])} |" @@ -809,7 +1009,9 @@ def render_packet_markdown(name: str, packet: dict) -> str: 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 += [ + f"* {source['side']}: {source['url']}" for source in packet["cited_sources"] + ] or ["* none recorded"] lines += [ "", "## Minimal case", diff --git a/tests/test_uk_event_diff.py b/tests/test_uk_event_diff.py index 07ec8c6..c024aa8 100644 --- a/tests/test_uk_event_diff.py +++ b/tests/test_uk_event_diff.py @@ -9,6 +9,7 @@ import hashlib import json import random +import shlex from pathlib import Path import pytest @@ -109,6 +110,7 @@ def make_namespace( heads=None, skip=(), identical=(), + remove_measure=False, ): """A complete namespace for one event; effects maps year -> {variable: gain}.""" effects = effects or { @@ -120,7 +122,7 @@ def make_namespace( "event_slug": EVENT, "event": EVENT, "calendar_years": [2024, 2025], - "measures": [m], + "measures": [] if remove_measure else [m], } registry_path = ( tmp_path / "data" / "uk" / "events" / f"{EVENT}_{name}_measures.json" @@ -128,7 +130,7 @@ def make_namespace( 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. + # Keep the current registry as well as each run's receipt-bound snapshot. canonical = tmp_path / "data" / "uk" / "events" / f"{EVENT}_measures.json" canonical.write_bytes(registry_path.read_bytes()) root = tmp_path / "ns" / name @@ -140,7 +142,7 @@ def make_namespace( {"variant_id": variant, "variant_identity_sha256": "e" * 64} if variant else {} ) artifacts = {} - for year in supported: + for year in () if remove_measure else supported: if year in skip: continue gains = effects[year] @@ -162,6 +164,11 @@ def make_namespace( event=EVENT, variant_fields=variant_fields, ) + artifact.update( + 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", + ) path = event_dir / f"{KEY}_{year}.json" path.write_bytes(compute.canonical_bytes(artifact)) artifacts[str(path.relative_to(tmp_path))] = hashlib.sha256( @@ -169,6 +176,7 @@ def make_namespace( ).hexdigest() manifest = { "event": EVENT, + "registry_path": str(registry_path.relative_to(tmp_path)), "registry_sha256": registry_sha, "certified_dataset_sha256": "a" * 64, "artifacts": artifacts, @@ -177,6 +185,20 @@ def make_namespace( "policyengine_core_version": "3.33.0" if variant else "3.27.1", "data_bundle": "build", } + pairs = ( + set() + if remove_measure + else {(KEY, year) for year in supported if year not in skip} + ) + full_pairs = set() if remove_measure else {(KEY, year) for year in (2024, 2025)} + manifest.update( + artifact_grid=compute.grid_rows(pairs), + requested_grid=compute.grid_rows(pairs), + full_event_grid=compute.grid_rows(full_pairs), + full_event_complete=full_pairs <= pairs, + years=sorted({year for _, year in pairs}), + measures=sorted({key for key, _ in pairs}), + ) if variant: manifest.update( variant_fields, @@ -187,7 +209,9 @@ def make_namespace( 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=[] + if remove_measure + else [{"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)) @@ -208,7 +232,7 @@ def make_namespace( ) compare.write_comparison( EVENT, - registry_path=canonical, + registry_path=registry_path, staged_path=event_dir / "STAGED.jsonl", output_dir=event_dir, artifact_root=tmp_path, @@ -542,7 +566,7 @@ def test_output_bytes_do_not_depend_on_input_order(ns): assert diff.render(a, shuffled) == first -def test_packets_cover_flagged_moves_but_not_support_only_rows(ns): +def test_packets_cover_flagged_moves_including_support_only_rows(ns): a = ns("a") b = ns( "b", @@ -552,17 +576,29 @@ def test_packets_cover_flagged_moves_but_not_support_only_rows(ns): ) 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 names == [ + f"packets/{EVENT}__tax_and_benefit__2024.json", + f"packets/{EVENT}__tax_and_benefit__2025.json", + ] + support_packet = json.loads(files[names[0]]) + assert support_packet["flags"] == ["support"] + packet = json.loads(files[names[1]]) assert packet["flags"] == ["numerical"] assert [c["source_row_id"] for c in packet["source_cells"]] == [ "tax_2025", "benefit_2025", "corporation_2025", + "tax_2025", + "benefit_2025", + "corporation_2025", ] + assert [c["side"] for c in packet["source_cells"]] == ["base"] * 3 + ["head"] * 3 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["cited_sources"] == [ + {"side": "base", "url": "https://www.gov.uk/example"}, + {"side": "head", "url": "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() @@ -578,6 +614,398 @@ def test_unverifiable_side_is_refused(ns, tmp_path): diff.load_side(tmp_path / "ns" / "a", artifact_root=tmp_path) +@pytest.mark.parametrize( + ("field", "value"), + [ + ("certified_dataset_sha256", "b" * 64), + ("supported_years", []), + ("unsupported_years", []), + ("data_years", [2024, 2025]), + ("data_years", []), + ("full_supported_grid", []), + ("registry_sha256", "b" * 64), + ("engine_version", "forged"), + ("variant_identity_sha256", "b" * 64), + ("artifact_grid", []), + ("requested_grid", [{"measure_key": KEY, "year": 2024}]), + ("full_event_grid", []), + ("years", []), + ("measures", []), + ("full_event_complete", True), + ("full_supported_grid_complete", False), + ], +) +def test_run_manifest_must_reconcile_with_verified_receipts(ns, tmp_path, field, value): + ns("a", supported=(2025,), variant="pe6_uk2_build") + path = tmp_path / "ns" / "a" / EVENT / "RUN_MANIFEST.json" + manifest = json.loads(path.read_bytes()) + manifest[field] = value + path.write_bytes(compute.canonical_bytes(manifest)) + with pytest.raises(diff.DiffError): + diff.load_side(path.parent.parent, artifact_root=tmp_path) + + +@pytest.mark.parametrize( + "corruption", ["raw_delta", "dataset_sha256_before", "dataset_sha256_after"] +) +def test_unreferenced_artifact_is_revalidated(ns, tmp_path, corruption): + ns("a", identical=(2024,)) + event_dir = tmp_path / "ns" / "a" / EVENT + artifact_path = event_dir / f"{KEY}_2024.json" + artifact = json.loads(artifact_path.read_bytes()) + if corruption == "raw_delta": + artifact["head_effects"]["income_tax"] = 2e9 + artifact["measure_total_gbp"] = 2e9 + else: + artifact[corruption] = "b" * 64 + artifact_path.write_bytes(compute.canonical_bytes(artifact)) + manifest_path = event_dir / "RUN_MANIFEST.json" + manifest = json.loads(manifest_path.read_bytes()) + manifest["artifacts"][str(artifact_path.relative_to(tmp_path))] = hashlib.sha256( + artifact_path.read_bytes() + ).hexdigest() + manifest_path.write_bytes(compute.canonical_bytes(manifest)) + with pytest.raises(diff.DiffError): + diff.load_side(event_dir.parent, artifact_root=tmp_path) + + +def test_variant_identity_cannot_be_removed_from_run_manifest(ns, tmp_path): + ns("a", variant="pe6_uk2_build") + path = tmp_path / "ns" / "a" / EVENT / "RUN_MANIFEST.json" + manifest = json.loads(path.read_bytes()) + for field in ( + "variant_id", + "variant_identity_sha256", + "data_years", + "supported_years", + "unsupported_years", + "full_supported_grid", + "full_supported_grid_complete", + ): + del manifest[field] + path.write_bytes(compute.canonical_bytes(manifest)) + with pytest.raises(diff.DiffError): + diff.load_side(path.parent.parent, artifact_root=tmp_path) + + +def test_unreferenced_artifact_cannot_be_removed_from_inventory(ns, tmp_path): + ns("a", identical=(2024,)) + event_dir = tmp_path / "ns" / "a" / EVENT + manifest_path = event_dir / "RUN_MANIFEST.json" + manifest = json.loads(manifest_path.read_bytes()) + del manifest["artifacts"][f"ns/a/{EVENT}/{KEY}_2024.json"] + manifest_path.write_bytes(compute.canonical_bytes(manifest)) + with pytest.raises(diff.DiffError): + diff.load_side(event_dir.parent, artifact_root=tmp_path) + + +def test_both_historical_registries_load_after_both_runs_are_built(ns, tmp_path): + ns("base") + ns("head", modifier=0.3) + base = diff.load_side(tmp_path / "ns" / "base", artifact_root=tmp_path) + head = diff.load_side(tmp_path / "ns" / "head", artifact_root=tmp_path) + assert ( + base["events"][EVENT]["registry_sha256"] + != head["events"][EVENT]["registry_sha256"] + ) + # Existing summary callers still require the current registry. + with pytest.raises(compare.EventComparisonError, match="current event registry"): + compare.load_verified_comparison( + tmp_path / "ns" / "base" / EVENT / "COMPARISON.json", + artifact_root=tmp_path, + ) + + +@pytest.mark.parametrize("field", ["by_status", "computed_measure_years"]) +def test_verified_staging_tally_must_reconcile_with_run(ns, tmp_path, field): + ns("a") + event_dir = tmp_path / "ns" / "a" / EVENT + staging_path = event_dir / "STAGING_MANIFEST.json" + staging = json.loads(staging_path.read_bytes()) + staging[field] = {} if field == "by_status" else 0 + staging_path.write_bytes(compute.canonical_bytes(staging)) + provenance_path = event_dir / "COMPARISON_PROVENANCE.json" + provenance = json.loads(provenance_path.read_bytes()) + provenance["staging_manifest_sha256"] = hashlib.sha256( + staging_path.read_bytes() + ).hexdigest() + if field in provenance["replay_grid"]: + provenance["replay_grid"][field] = staging[field] + provenance_path.write_bytes(compute.canonical_bytes(provenance)) + with pytest.raises(diff.DiffError): + diff.load_side(event_dir.parent, artifact_root=tmp_path) + + +def test_inertness_uses_provenance_bound_staging_manifest(ns, tmp_path): + original = ns("a") + event_dir = tmp_path / "ns" / "a" / EVENT + relocated = tmp_path / "receipts" / EVENT + relocated.mkdir(parents=True) + for name in ("STAGED.jsonl", "STAGING_MANIFEST.json"): + (relocated / name).write_bytes((event_dir / name).read_bytes()) + provenance_path = event_dir / "COMPARISON_PROVENANCE.json" + provenance = json.loads(provenance_path.read_bytes()) + provenance["staged_path"] = str((relocated / "STAGED.jsonl").relative_to(tmp_path)) + provenance["staging_manifest_path"] = str( + (relocated / "STAGING_MANIFEST.json").relative_to(tmp_path) + ) + provenance_path.write_bytes(compute.canonical_bytes(provenance)) + local_manifest = event_dir / "STAGING_MANIFEST.json" + staging = json.loads(local_manifest.read_bytes()) + staging["inert_measure_years"] = [{"measure_key": KEY, "year": 2025}] + local_manifest.write_bytes(compute.canonical_bytes(staging)) + reloaded = diff.load_side(event_dir.parent, artifact_root=tmp_path) + assert not reloaded["events"][EVENT]["inert_pairs"] + assert not any(row["flags"] for row in diff.diff_sides(original, reloaded)["rows"]) + + +def test_registry_digest_alone_flags_each_affected_record(ns): + base = ns("registry_base") + head = copy.deepcopy(base) + head_event = head["events"][EVENT] + head_event["registry"]["measures"][0]["title"] += " (revised registry note)" + head_event["registry_sha256"] = hashlib.sha256( + compute.canonical_bytes(head_event["registry"]) + ).hexdigest() + + result = diff.diff_sides(base, head) + assert result["events"][EVENT]["registry_changed"] + assert {record["level"] for record in result["rows"]} == set(diff.LEVELS) + for record in result["rows"]: + assert record["flags"] == ["mapping"] + assert record["details"] == [ + { + "flag": "mapping", + "code": "registry_sha256_changed", + "base": base["events"][EVENT]["registry_sha256"], + "head": head_event["registry_sha256"], + } + ] + + +@pytest.mark.parametrize( + "classification", ["out_of_household_scope", "not_expressible"] +) +@pytest.mark.parametrize("head_supported", [(2025,), (2024, 2025)]) +def test_missing_noncomputable_row_requires_shared_year_support( + ns, classification, head_supported +): + base = ns("coverage_base", supported=(2025,), variant="pe6_uk2_build") + head = ns("coverage_head", supported=head_supported, variant="pe6_uk2_build") + base_event = base["events"][EVENT] + head_event = head["events"][EVENT] + base_event["rows"]["corporation_2024"]["classification"] = classification + del head_event["rows"]["corporation_2024"] + + record = next( + row + for row in diff.compare_source_rows(EVENT, base_event, head_event) + if row["key"] == "corporation_2024" + ) + assert record["flags"] == [] + + +def test_missing_noncomputable_row_has_coverage_when_both_support_the_year(ns): + base = ns("shared_base") + head = ns("shared_head") + del head["events"][EVENT]["rows"]["corporation_2024"] + record = next( + row + for row in diff.compare_source_rows( + EVENT, base["events"][EVENT], head["events"][EVENT] + ) + if row["key"] == "corporation_2024" + ) + assert record["flags"] == ["coverage"] + + +def test_missing_event_has_no_coverage_without_shared_support(ns): + base = ns("missing_event_base")["events"][EVENT] + source_records = diff.compare_source_rows(EVENT, base, None) + artifact_records = diff.compare_artifacts(EVENT, base, None, (None, None)) + assert source_records and artifact_records + assert not any(record["flags"] for record in source_records + artifact_records) + + +@st.composite +def replay_sides(draw): + """Small sides for the pure diff's order, symmetry and membership properties.""" + supported = sorted(draw(st.sets(st.sampled_from([2024, 2025]), min_size=1))) + included = draw( + st.sets( + st.sampled_from( + [ + f"{prefix}_{year}" + for year in (2024, 2025) + for prefix in ("tax", "benefit", "corporation") + ] + ) + ) + ) + gains = { + year: { + variable: float(draw(st.integers(-200, 200)) * 1_000_000) + for variable in ("income_tax", "universal_credit") + } + for year in (2024, 2025) + } + m = measure() + m["source_rows"] = [ + source for source in m["source_rows"] if source["source_row_id"] in included + ] + registry = {"calendar_years": [2024, 2025], "measures": [m]} + registry_sha256 = hashlib.sha256(compute.canonical_bytes(registry)).hexdigest() + rows = {} + for source in m["source_rows"]: + source_id = source["source_row_id"] + year = int(source["fy"][:4]) + noncomputable = source.get("classification") == "out_of_household_scope" + variable = ( + "income_tax" if source["tax_head"] == "Income tax" else "universal_credit" + ) + value = ( + None if noncomputable or year not in supported else gains[year][variable] + ) + rows[source_id] = { + "measure_key": KEY, + "year": year, + "fy": source["fy"], + "tax_head": source["tax_head"], + "classification": source.get("classification", "partial"), + "pe_value_gbp": value, + "obr_value_gbp": source["value_gbp"], + "head_variables": [] if noncomputable else [variable], + "status": "out_of_household_scope" if noncomputable else "computed", + "ratio_bin": None if value is None else "synthetic_bin", + } + artifacts = { + (KEY, year): { + "path": f"synthetic/{KEY}_{year}.json", + "sha256": "a" * 64, + "artifact": { + "schema_version": 3, + "measure_total_gbp": sum(gains[year].values()), + "head_effects": gains[year], + "head_channels": {"income_tax": "tax", "universal_credit": "spending"}, + "head_definitions": {}, + "construction": "reversal_on_certified_world", + "baseline_world": {"reform_dict": {"gov.tax.rate": 0.2}}, + "reform_world": {"reform_dict": None}, + }, + } + for year in supported + } + identity = { + "variant_id": "synthetic_variant", + "policyengine_version": "6", + "policyengine_uk_version": "2", + "policyengine_core_version": "3", + "data_bundle": "synthetic_build", + "certified_dataset_sha256": "a" * 64, + } + return { + "root": Path("/tmp/synthetic"), + "artifact_root": Path("/tmp"), + "identity": identity, + "premises": None, + "premises_sha256": None, + "events": { + EVENT: { + "rows": rows, + "artifacts": artifacts, + "registry": registry, + "registry_sha256": registry_sha256, + "registry_path": "registry.json", + "registry_years": [2024, 2025], + "supported_years": supported, + "inert_pairs": set(), + "inputs": {}, + } + }, + } + + +def record_key(record): + return ( + record["event"], + record["measure_key"], + record["year"], + record["level"], + record["key"], + ) + + +@settings(max_examples=100, deadline=None) +@given(replay_sides()) +def test_identical_sides_have_no_flags(side): + result = diff.diff_sides(side, copy.deepcopy(side)) + assert not any(record["flags"] for record in result["rows"]) + + +@settings(max_examples=100, deadline=None) +@given(replay_sides(), replay_sides()) +def test_swapping_sides_preserves_flags_and_mirrors_values(base, head): + forward = {record_key(row): row for row in diff.diff_sides(base, head)["rows"]} + backward = {record_key(row): row for row in diff.diff_sides(head, base)["rows"]} + assert forward.keys() == backward.keys() + for key, row in forward.items(): + reverse = backward[key] + assert row["flags"] == reverse["flags"] + assert row["base_gbp"] == reverse["head_gbp"] + assert row["head_gbp"] == reverse["base_gbp"] + assert row["delta_gbp"] == ( + None if reverse["delta_gbp"] is None else -reverse["delta_gbp"] + ) + + +@settings(max_examples=100, deadline=None) +@given(replay_sides(), replay_sides()) +def test_every_source_row_on_either_side_has_one_source_record(base, head): + result = diff.diff_sides(base, head) + expected = set(base["events"][EVENT]["rows"]) | set(head["events"][EVENT]["rows"]) + source_keys = [row["key"] for row in result["rows"] if row["level"] == "source_row"] + assert set(source_keys) == expected + assert len(source_keys) == len(expected) + + +@settings(max_examples=100, deadline=None) +@given(replay_sides(), replay_sides()) +def test_every_flagged_record_belongs_to_exactly_one_packet(base, head): + result = diff.diff_sides(base, head) + packets = diff.build_packets(result, base, head) + for record in result["rows"]: + if record["flags"]: + assert ( + sum( + record_key(candidate) == record_key(record) + for packet in packets.values() + for candidate in packet["records"] + ) + == 1 + ) + + +@settings(max_examples=100, deadline=None) +@given(replay_sides(), replay_sides()) +def test_coverage_flags_require_shared_year_support(base, head): + for record in diff.diff_sides(base, head)["rows"]: + if "coverage" in record["flags"]: + assert all( + record["year"] in side["events"][record["event"]]["supported_years"] + for side in (base, head) + ) + + +@settings(max_examples=100, deadline=None) +@given(replay_sides(), replay_sides()) +def test_rows_without_counterparts_never_have_support_flags(base, head): + for record in diff.compare_source_rows( + EVENT, base["events"][EVENT], head["events"][EVENT] + ): + if record["key"].startswith("corporation_"): + assert "support" not in record["flags"] + + finite = st.floats(min_value=-1e13, max_value=1e13, allow_nan=False) @@ -612,3 +1040,90 @@ 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) ] + + +def test_removed_measure_packet_retains_base_source_evidence(ns): + base = ns("base") + head = ns("head", remove_measure=True) + packets = diff.build_packets(diff.diff_sides(base, head), base, head) + packet = packets[f"{EVENT}__tax_and_benefit__2024"] + assert "coverage" in packet["flags"] + assert packet["title"] == measure()["title"] + assert packet["measures"]["base"] == measure() + assert packet["measures"]["head"] is None + assert [ + (cell["side"], cell["source_row_id"]) for cell in packet["source_cells"] + ] == [ + ("base", "tax_2024"), + ("base", "benefit_2024"), + ("base", "corporation_2024"), + ] + assert packet["cited_sources"] == [ + {"side": "base", "url": "https://www.gov.uk/example"} + ] + markdown = diff.render_packet_markdown("removed", packet) + assert "| base | tax_2024" in markdown + assert "* base: https://www.gov.uk/example" in markdown + + +def test_packets_keep_each_sides_changed_cells_and_citations(ns): + base = ns("base") + head = ns( + "head", + effects={ + 2024: {"income_tax": 300e6, "universal_credit": -30e6}, + 2025: {"income_tax": 110e6, "universal_credit": -33e6}, + }, + ) + head_measure = head["events"][EVENT]["registry"]["measures"][0] + head_measure["title"] = "Updated announcement: https://www.gov.uk/updated" + head_measure["source_rows"][0]["value_gbp"] = 125 + head_measure["source_rows"][0]["source_column"] = "Updated column" + packets = diff.build_packets(diff.diff_sides(base, head), base, head) + packet = packets[f"{EVENT}__tax_and_benefit__2024"] + tax_cells = [c for c in packet["source_cells"] if c["source_row_id"] == "tax_2024"] + assert [(c["side"], c["obr_value_gbp"], c["source_column"]) for c in tax_cells] == [ + ("base", 100, "C"), + ("head", 125, "Updated column"), + ] + assert packet["cited_sources"] == [ + {"side": "base", "url": "https://www.gov.uk/example"}, + {"side": "head", "url": "https://www.gov.uk/updated"}, + ] + + +def test_reproduction_uses_each_verified_registry_and_variant(ns, tmp_path): + base = ns("base") + head = ns("head", modifier=0.25, variant="pe6_uk2_build") + packet = diff.build_packets(diff.diff_sides(base, head), base, head)[ + f"{EVENT}__tax_and_benefit__2024" + ] + for label, side in (("base", base), ("head", head)): + argv = shlex.split(packet["reproduce"][label]) + registry_path = Path(argv[argv.index("--registry") + 1]) + assert ( + registry_path + == (tmp_path / side["events"][EVENT]["registry_path"]).resolve() + ) + assert ( + hashlib.sha256(registry_path.read_bytes()).hexdigest() + == side["events"][EVENT]["registry_sha256"] + ) + assert argv[argv.index("--variant") + 1] == side["identity"]["variant_id"] + assert packet["reproduce"]["base"] != packet["reproduce"]["head"] + + +def test_committed_reproduction_command_is_accepted_by_compute_cli( + monkeypatch, tmp_path +): + side = diff.load_side(diff.ROOT / "results" / "uk" / "events") + event = "autumn_budget_2024" + key, year = next(iter(side["events"][event]["artifacts"])) + argv = shlex.split(diff._reproduce(side, event, key, year))[3:] + argv[argv.index("--output-dir") + 1] = str(tmp_path / "reproduction") + + def unavailable_engine(): + raise ImportError("This regression checks the engine-free plan only") + + monkeypatch.setattr(compute.worlds, "engine_resolver", unavailable_engine) + assert compute.main(argv + ["--dry-run"]) == 0 From f87b5a263544aeef4b620b8fa932351188e8daa0 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 20:27:16 -0400 Subject: [PATCH 4/5] Document verified release diff inputs and packet invariants Co-Authored-By: GPT-6.1 Sol (Subfleet lane) --- docs/uk_replay/RELEASE_REPLAY.md | 30 ++++++++++++++++++++++++------ 1 file changed, 24 insertions(+), 6 deletions(-) diff --git a/docs/uk_replay/RELEASE_REPLAY.md b/docs/uk_replay/RELEASE_REPLAY.md index 769cdcb..e40a766 100644 --- a/docs/uk_replay/RELEASE_REPLAY.md +++ b/docs/uk_replay/RELEASE_REPLAY.md @@ -222,7 +222,15 @@ Observed on 2.125.1 on 11 October (parameter tree only): `pipeline/diff_uk_event_runs.py --base --head ` compares two complete namespaces, engine-free, after re-verifying each side's -staging and comparison receipts. Keys: +staging and comparison receipts against that side's receipt-bound registry +snapshot. The current event registry does not replace a historical snapshot. +Every inventoried artifact is checked by the staging and comparison validators, +including artifacts that supply no numeric source counterpart. Each run +manifest's execution identity, registry and artifact inventory, supported and +unsupported years, and measure-year grids must agree with the verified staging +rows and tally. Inertness comes from the staging manifest named and verified by +comparison provenance. Inconsistent receipts are refused before a diff is +written. Keys: * source row: `(event, source_row_id)`; * measure-year total: `(event, measure_key, year)`; @@ -236,9 +244,9 @@ bin transitions. |---|---| | `numerical` | Both sides numeric and `abs(head − base) > max(£1,000,000, 0.01 × max(abs(head), abs(base)))`. Strictly greater. | | `structural` | Sign reversal or a zero crossing (zero means `abs(x) ≤ £0.50`, accounting roundoff); artifact schema or orientation changed; non-finite value. Regardless of size. | -| `coverage` | A counterpart, artifact pair or source row present on one side only, in a year both sides support. | -| `mapping` | A row's `head_variables`, an artifact's `head_channels` or `head_definitions`, or the registry SHA-256 differs. | -| `support` | A year supported on one side only. | +| `coverage` | A counterpart, artifact pair or source row present on one side only, in a year both sides support. The shared-support requirement also applies to rows outside household scope or without an expressible counterpart. | +| `mapping` | A row's `head_variables`, an artifact's `head_channels` or `head_definitions`, or the registry SHA-256 differs. Registry changes flag the event's affected records with both digests. | +| `support` | A year supported on one side only. Source rows that cannot carry a counterpart do not receive this flag. | | `inertness` | `inert_construction` on one side only, or identical worlds in an in-force year on one side only. | | `construction` | Executed `baseline_world`/`reform_world` reform dicts differ, or a premise check's result differs. A changed current-law value of a touched parameter is recorded as context on the row, not flagged: it is what a model release is expected to change. | @@ -247,14 +255,24 @@ recorded once per diff; they explain moves and are not flags on rows. Outputs: `DIFF.json` (canonical, sorted), `DIFF.md`, and one issue packet per flagged `(event, measure, year)` under `packets/`. A packet holds the source -cells (table, column, row number, source-row id, OBR value), both execution +cells (table, column, row number, source-row id, OBR value) and measure details +from each registry with base/head attribution, both execution identities, registry and artifact digests, the executed reform dicts, the affected heads with base/head values, focused rerun commands for each -variant, the statutory sources the registry already cites, and empty +variant using that side's verified registry path and variant selection, the +statutory sources each registry already cites with side attribution, and empty `minimal_case` and `statutory_evidence` sections for the investigating lane. +Removed measures retain their base title, source cells and citations. Packets are evidence for the coordinating session, which files upstream issues; this pipeline files none. +Diff invariants are checked with deterministic cases and Hypothesis properties +in `tests/test_uk_event_diff.py`: identical sides raise no flags; swapping sides +preserves each record's flag set and mirrors its values; every source row in +either side appears once at source-row level; every flagged record appears in +exactly one packet, including support-only records; numerical moves flag only +strictly above the stated threshold; and coverage and support do not overlap. + ## Qualification requests and receipts See [QUALIFICATION_INTERFACE.md](QUALIFICATION_INTERFACE.md) for the request From 8700d29a905bc24e9b8f9032813997b262510eb9 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 11 Oct 2026 20:40:04 -0400 Subject: [PATCH 5/5] Diff: check each run's variant against the given variants file After #160, staging checks a manifest's variant against its record in a variants document, and the diff re-stages each run to revalidate it. The diff now takes --variants-file (default data/uk/replay_variants.json) and passes it to staging, so a qualification run can diff a candidate whose record lives only in its work file. The diff tests build synthetic variants (several share an id with different data years), so they stand in a record derived from each manifest; staging's variant checks are tested in test_uk_replay_variants.py, and a new test checks the file reaches staging. Co-Authored-By: Claude Opus 5.5 --- pipeline/diff_uk_event_runs.py | 35 +++++++++++++++++++++----- tests/test_uk_event_diff.py | 45 ++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 6 deletions(-) diff --git a/pipeline/diff_uk_event_runs.py b/pipeline/diff_uk_event_runs.py index f4c3006..c843a4d 100644 --- a/pipeline/diff_uk_event_runs.py +++ b/pipeline/diff_uk_event_runs.py @@ -162,7 +162,12 @@ def resolve_root(value: str, *, artifact_root: Path = ROOT) -> Path: def _revalidate_run( - event_dir: Path, manifest: dict, provenance: dict, *, artifact_root: Path + event_dir: Path, + manifest: dict, + provenance: dict, + *, + artifact_root: Path, + variants_path: Path | None = None, ) -> tuple[dict, dict, dict, Path]: """Reconcile unbound run metadata with the receipt-bound staging inputs.""" registry_path = compare._receipt_path(provenance["registry_path"], artifact_root) @@ -215,6 +220,7 @@ def _revalidate_run( staging_manifest, artifact_dir=event_dir, registry_sha256=provenance["registry_sha256"], + variants_path=variants_path or stage.variants.VARIANTS_PATH, ) pairs = set() for absolute in inventory: @@ -299,8 +305,14 @@ def _revalidate_run( return registry, staging, artifacts, staging_path -def load_side(root: Path, *, artifact_root: Path = ROOT) -> dict: - """Every event of a namespace, re-verified through its receipts.""" +def load_side( + root: Path, *, artifact_root: Path = ROOT, variants_path: Path | None = None +) -> dict: + """Every event of a namespace, re-verified through its receipts. + + A variant run's manifest is checked against its record in + ``variants_path`` (default data/uk/replay_variants.json); a qualification + run passes its work file, which also holds the candidate's record.""" root = root.resolve() artifact_root = artifact_root.resolve() events = {} @@ -325,7 +337,11 @@ def load_side(root: Path, *, artifact_root: Path = ROOT) -> dict: registry_path=registry_path, ) registry, staging, artifacts, staging_path = _revalidate_run( - event_dir, manifest, provenance, artifact_root=artifact_root + event_dir, + manifest, + provenance, + artifact_root=artifact_root, + variants_path=variants_path, ) identity = side_identity(manifest) except (ValueError, KeyError, TypeError, OSError) as exc: @@ -1184,6 +1200,12 @@ def main(argv: list[str] | None = None) -> int: 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( + "--variants-file", + type=Path, + help="variant records the runs' manifests are checked against " + "(default data/uk/replay_variants.json)", + ) parser.add_argument("--max-rows", type=int, default=200) parser.add_argument( "--check", @@ -1191,8 +1213,9 @@ def main(argv: list[str] | None = None) -> int: 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)) + variants_path = args.variants_file.resolve() if args.variants_file else None + base = load_side(resolve_root(args.base), variants_path=variants_path) + head = load_side(resolve_root(args.head), variants_path=variants_path) output_dir = args.output_dir or ( DIFF_ROOT / f"{base['identity']['variant_id']}__{head['identity']['variant_id']}" diff --git a/tests/test_uk_event_diff.py b/tests/test_uk_event_diff.py index c024aa8..b9721c1 100644 --- a/tests/test_uk_event_diff.py +++ b/tests/test_uk_event_diff.py @@ -240,9 +240,34 @@ def make_namespace( return diff.load_side(root, artifact_root=tmp_path) +def synthetic_record(manifest, registry, *, variants_path): + """Stand-in for staging's variant lookup. These sides use synthetic + variants (several share an id with different data years), so no one + variants document can hold them; staging's checks of a manifest against + its record are tested in test_uk_replay_variants.py.""" + if manifest is None or manifest.get("variant_id") is None: + return None + return { + "variant_id": manifest["variant_id"], + "layout": "replay_namespace", + "identity_sha256": manifest["variant_identity_sha256"], + "data_years": manifest["data_years"], + "identity": { + "dataset": { + "path": "synthetic.h5", + "repo_id": "synthetic/data", + "revision": manifest["data_bundle"], + "sha256": manifest["certified_dataset_sha256"], + }, + "release_bundle": {"certified_data_build_id": manifest["data_bundle"]}, + }, + } + + @pytest.fixture def ns(tmp_path, monkeypatch): monkeypatch.setattr(compute, "ROOT", tmp_path) + monkeypatch.setattr(stage, "manifest_variant", synthetic_record) def build(name, **kwargs): return make_namespace(tmp_path, name, **kwargs) @@ -1127,3 +1152,23 @@ def unavailable_engine(): monkeypatch.setattr(compute.worlds, "engine_resolver", unavailable_engine) assert compute.main(argv + ["--dry-run"]) == 0 + + +def test_the_variants_file_reaches_staging(ns, tmp_path, monkeypatch): + # A qualification run diffs a candidate whose record is only in its work + # variants file; the diff must check each manifest against that file. + ns("base", variant="pe6_uk2_build") + seen = [] + real = stage.stage_event + + def spy(*args, **kwargs): + seen.append(kwargs.get("variants_path")) + return real(*args, **kwargs) + + monkeypatch.setattr(stage, "stage_event", spy) + work = tmp_path / "work-variants.json" + diff.load_side(tmp_path / "ns" / "base", artifact_root=tmp_path, variants_path=work) + assert seen == [work] + seen.clear() + diff.load_side(tmp_path / "ns" / "base", artifact_root=tmp_path) + assert seen == [stage.variants.VARIANTS_PATH]