diff --git a/docs/uk_replay/RELEASE_REPLAY.md b/docs/uk_replay/RELEASE_REPLAY.md index 4c8c1f8..4603ef4 100644 --- a/docs/uk_replay/RELEASE_REPLAY.md +++ b/docs/uk_replay/RELEASE_REPLAY.md @@ -275,7 +275,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)`; @@ -289,9 +297,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. | @@ -300,14 +308,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 diff --git a/pipeline/compare_uk_event.py b/pipeline/compare_uk_event.py index e746192..aebd5ef 100644 --- a/pipeline/compare_uk_event.py +++ b/pipeline/compare_uk_event.py @@ -1321,8 +1321,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") @@ -1354,7 +1360,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 new file mode 100644 index 0000000..c843a4d --- /dev/null +++ b/pipeline/diff_uk_event_runs.py @@ -0,0 +1,1238 @@ +"""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, 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 + +import argparse +import hashlib +import json +import math +import re +import shlex +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 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 +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) +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 _revalidate_run( + 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) + 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"], + variants_path=variants_path or stage.variants.VARIANTS_PATH, + ) + 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, 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 = {} + 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 + 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, + registry_path=registry_path, + ) + registry, staging, artifacts, staging_path = _revalidate_run( + event_dir, + manifest, + provenance, + artifact_root=artifact_root, + variants_path=variants_path, + ) + identity = side_identity(manifest) + except (ValueError, KeyError, TypeError, OSError) as exc: + raise DiffError(f"{event_dir}: comparison does not verify: {exc}") from exc + registry_years = sorted(registry["calendar_years"]) + supported = manifest.get("supported_years", registry_years) + 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), + "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, + "artifact_root": artifact_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: + if b is None or h is None: + details.append( + { + "flag": "coverage", + "code": "source_row_on_base_only" + if h is None + else "source_row_on_head_only", + } + ) + else: + inert = { + side: r.get("status") == "inert_construction" + for side, r in (("base", b), ("head", h)) + } + if inert["base"] != inert["head"]: + details.append( + { + "flag": "inertness", + "code": "inert_on_base_only" + if inert["base"] + else "inert_on_head_only", + } + ) + elif (bv is None) != (hv is None): + details.append( + { + "flag": "coverage", + "code": "counterpart_on_base_only" + if hv is None + else "counterpart_on_head_only", + } + ) + if bv is not None and hv is not None: + details += numeric_details(bv, hv) + if b["head_variables"] != h["head_variables"]: + details.append( + { + "flag": "mapping", + "code": "head_variables_changed", + "base": b["head_variables"], + "head": h["head_variables"], + } + ) + out.append( + _record( + event, + "source_row", + source_row_id, + row["measure_key"], + year, + bv, + hv, + details, + { + "fy": row["fy"], + "tax_head": row.get("tax_head"), + "obr_value_gbp": row["obr_value_gbp"], + "status": { + "base": b.get("status") if b else None, + "head": h.get("status") if h else None, + }, + "ratio_bin": { + "base": b.get("ratio_bin") if b else None, + "head": h.get("ratio_bin") if h else None, + }, + }, + ) + ) + return out + + +def _premise_details(event, measure_key, year, base_premises, head_premises): + """Construction flags from premise records; current-law moves are context.""" + + def measure(premises): + return ( + ((premises or {}).get("events", {}).get(event, {})) + .get("measures", {}) + .get(measure_key) + ) + + b, h = measure(base_premises), measure(head_premises) + details, context = [], {} + if b is None or h is None: + return details, context + for name in sorted(set(b.get("checks", {})) | set(h.get("checks", {}))): + before = (b.get("checks", {}).get(name) or {}).get("holds") + after = (h.get("checks", {}).get(name) or {}).get("holds") + if before != after: + details.append( + { + "flag": "construction", + "code": "premise_check_changed", + "check": name, + "base": before, + "head": after, + } + ) + by = (b.get("years", {}) or {}).get(str(year), {}).get("paths", {}) + hy = (h.get("years", {}) or {}).get(str(year), {}).get("paths", {}) + moved = {} + for path in sorted(set(by) | set(hy)): + bp, hp = by.get(path, {}), hy.get(path, {}) + for world in ("baseline_world", "reform_world"): + # A world that is current law on both sides moved with current + # law (context below); only an executed override is construction. + overrides = any( + side.get(world) != side.get("current_law") for side in (bp, hp) + ) + if overrides and bp.get(world) != hp.get(world): + details.append( + { + "flag": "construction", + "code": f"{world}_value_changed", + "path": path, + "base": bp.get(world), + "head": hp.get(world), + } + ) + if bp.get("current_law") != hp.get("current_law"): + moved[path] = {"base": bp.get("current_law"), "head": hp.get("current_law")} + if moved: + context["current_law_changed"] = moved + return details, context + + +def compare_artifacts( + event: str, base: dict | None, head: dict | None, premises +) -> list: + out = [] + pairs = set(base["artifacts"] if base else ()) | set( + head["artifacts"] if head else () + ) + for measure_key, year in sorted(pairs): + b = ( + (base["artifacts"].get((measure_key, year)) or {}).get("artifact") + if base + else None + ) + h = ( + (head["artifacts"].get((measure_key, year)) or {}).get("artifact") + if head + else None + ) + details = _support(event, year, base, head) + context = {} + bv = b["measure_total_gbp"] if b else None + hv = h["measure_total_gbp"] if h else None + head_records = [] + 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( + { + "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 + ) + 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": registry_changed, + "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: + data = side["events"].get(event) + if data is None: + return "# This event is absent from this namespace." + variant = side["identity"]["variant_id"] + 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]: + 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)} + 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 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)) + 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"), + "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), + "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": [ + {"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": ( + "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", + "", + "| Side | Source row | Table / column | FY | Head | OBR £m | Base £m | Head £m |", + "|---|---|---|---|---|---:|---:|---:|", + ] + for cell in packet["source_cells"]: + lines.append( + 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'])} |" + ) + 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"* {source['side']}: {source['url']}" for source 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( + "--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", + action="store_true", + help="compare a fresh render with the files on disk", + ) + args = parser.parse_args(argv) + 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']}" + ) + 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..b9721c1 --- /dev/null +++ b/tests/test_uk_event_diff.py @@ -0,0 +1,1174 @@ +"""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 +import shlex +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=(), + remove_measure=False, +): + """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": [] if remove_measure else [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() + # 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 + 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 () if remove_measure else 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, + ) + 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( + path.read_bytes() + ).hexdigest() + manifest = { + "event": EVENT, + "registry_path": str(registry_path.relative_to(tmp_path)), + "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", + } + 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, + 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=[] + 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)) + 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=registry_path, + staged_path=event_dir / "STAGED.jsonl", + output_dir=event_dir, + artifact_root=tmp_path, + ) + 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) + + return build + + +def flags_by_key(result): + return { + (r["level"], r["key"], r["measure_key"], r["year"]): r["flags"] + for r in result["rows"] + } + + +def test_committed_legacy_namespace_has_no_flags_against_itself(tmp_path): + side = diff.load_side(diff.ROOT / "results" / "uk" / "events") + result = diff.diff_sides(side, side) + assert set(result["events"]) == { + "autumn_budget_2024", + "autumn_statement_2023", + "spring_budget_2024", + "spring_statement_2025", + "spring_budget_2023", + } + assert not any(r["flags"] for r in result["rows"]) + assert result["identity_changes"] == {} + files = diff.render(side, side) + diff.write(files, tmp_path / "out") + assert diff.check(files, tmp_path / "out") == [] + with open(tmp_path / "out" / "DIFF.json", "ab") as handle: + handle.write(b" ") + assert diff.check(files, tmp_path / "out") == ["DIFF.json"] + + +def test_identity_change_alone_raises_no_row_flags(ns): + result = diff.diff_sides(ns("a"), ns("b", variant="pe6.2.6_uk2.125.1_build")) + assert not any(r["flags"] for r in result["rows"]) + assert result["identity_changes"]["variant_id"]["head"] == "pe6.2.6_uk2.125.1_build" + assert result["identity_changes"]["policyengine_uk_version"] == { + "base": "2.89.2", + "head": "2.125.1", + } + + +def test_numerical_flag_only_strictly_above_the_threshold(ns): + base = ns("a") + within = ns( + "b", + effects={ + 2024: {"income_tax": 100e6 + 1e6, "universal_credit": -30e6}, + 2025: {"income_tax": 110e6, "universal_credit": -33e6}, + }, + ) + above = ns( + "c", + effects={ + 2024: {"income_tax": 100e6 + 1.5e6, "universal_credit": -30e6}, + 2025: {"income_tax": 110e6, "universal_credit": -33e6}, + }, + ) + assert not any(r["flags"] for r in diff.diff_sides(base, within)["rows"]) + flagged = flags_by_key(diff.diff_sides(base, above)) + assert flagged[("head", "income_tax", KEY, 2024)] == ["numerical"] + assert flagged[("measure_year", "total", KEY, 2024)] == ["numerical"] + assert flagged[("source_row", "tax_2024", KEY, 2024)] == ["numerical"] + assert flagged[("head", "universal_credit", KEY, 2024)] == [] + + +def test_sign_reversal_and_zero_crossing_are_structural_regardless_of_size(ns): + base = ns( + "a", + effects={ + 2024: {"income_tax": 0.2, "universal_credit": -30e6}, + 2025: {"income_tax": 0.4, "universal_credit": -33e6}, + }, + ) + head = ns( + "b", + effects={ + 2024: {"income_tax": 0.9, "universal_credit": -30e6}, + 2025: {"income_tax": -0.6, "universal_credit": -33e6}, + }, + ) + flagged = flags_by_key(diff.diff_sides(base, head)) + assert flagged[("head", "income_tax", KEY, 2024)] == ["structural"] + assert flagged[("head", "income_tax", KEY, 2025)] == ["structural"] + codes = { + (r["year"], d["code"]) + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "head" and r["key"] == "income_tax" + for d in r["details"] + } + assert codes == {(2024, "zero_crossing"), (2025, "zero_crossing")} + flipped = ns( + "c", + effects={ + 2024: {"income_tax": -5.0, "universal_credit": -30e6}, + 2025: {"income_tax": 0.4, "universal_credit": -33e6}, + }, + ) + base5 = ns( + "d", + effects={ + 2024: {"income_tax": 5.0, "universal_credit": -30e6}, + 2025: {"income_tax": 0.4, "universal_credit": -33e6}, + }, + ) + rows = diff.diff_sides(base5, flipped)["rows"] + assert any(d["code"] == "sign_reversal" for r in rows for d in r["details"]) + + +def test_unsupported_year_is_support_never_coverage(ns): + result = diff.diff_sides( + ns("a"), ns("b", supported=(2025,), variant="pe6_uk2_build") + ) + for record in result["rows"]: + if record["key"].startswith("corporation_"): + # Outside household scope in every year: support cannot move it. + assert record["flags"] == [] + elif record["year"] == 2024: + assert record["flags"] == ["support"], record + else: + assert record["flags"] == [] + assert result["flag_counts"]["coverage"] == 0 + assert result["flag_counts"]["support"] > 0 + + +def test_missing_artifact_in_a_supported_year_is_coverage(ns): + result = diff.diff_sides(ns("a"), ns("b", skip=(2025,), variant="pe6_uk2_build")) + flagged = flags_by_key(result) + assert flagged[("measure_year", "total", KEY, 2025)] == ["coverage"] + assert flagged[("source_row", "tax_2025", KEY, 2025)] == ["coverage"] + + +def test_newly_inert_construction_is_flagged(ns): + result = diff.diff_sides(ns("a"), ns("b", identical=(2025,))) + flagged = flags_by_key(result) + assert "inertness" in flagged[("measure_year", "total", KEY, 2025)] + assert "inertness" in flagged[("source_row", "tax_2025", KEY, 2025)] + + +def test_changed_construction_and_mapping_are_flagged(ns): + changed = flags_by_key(diff.diff_sides(ns("a"), ns("b", modifier=0.25))) + assert "construction" in changed[("measure_year", "total", KEY, 2024)] + heads = [ + { + "obr_head": "Income tax", + "pe_variables": ["income_tax", "national_insurance"], + "channel": "tax", + }, + { + "obr_head": "Welfare inside cap", + "pe_variables": ["universal_credit"], + "channel": "spending", + }, + ] + effects = { + y: {"income_tax": 100e6, "national_insurance": 0.0, "universal_credit": -30e6} + for y in (2024, 2025) + } + remapped = flags_by_key( + diff.diff_sides(ns("c"), ns("d", heads=heads, effects=effects)) + ) + assert "mapping" in remapped[("measure_year", "total", KEY, 2024)] + assert remapped[("head", "national_insurance", KEY, 2024)] == ["mapping"] + assert "mapping" in remapped[("source_row", "tax_2024", KEY, 2024)] + + +def test_premise_changes_flag_construction_and_current_law_is_context(ns): + base, head = ns("a"), ns("b") + premise = { + "events": { + EVENT: { + "measures": { + KEY: { + "checks": { + "reversal_differs_from_current_law": {"holds": True} + }, + "years": { + "2024": { + "paths": { + "gov.tax.rate": { + "current_law": 0.3, + "baseline_world": 0.2, + "reform_world": 0.3, + } + } + } + }, + } + } + } + } + } + base["premises"] = premise + + def record_for(head_premises): + head["premises"] = head_premises + return next( + r + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "measure_year" and r["year"] == 2024 + ) + + def changed(**values): + out = copy.deepcopy(premise) + out["events"][EVENT]["measures"][KEY]["years"]["2024"]["paths"][ + "gov.tax.rate" + ].update(values) + return out + + # The reform world of a reversal is current law: when current law moves, + # that world moves with it, which is context and not a construction change. + record = record_for(changed(current_law=0.35, reform_world=0.35)) + assert record["flags"] == [] + assert record["context"]["current_law_changed"] == { + "gov.tax.rate": {"base": 0.3, "head": 0.35} + } + # An executed override that changes (a re-derived counterfactual) is. + record = record_for(changed(baseline_world=0.25)) + assert record["flags"] == ["construction"] + assert [d["code"] for d in record["details"]] == ["baseline_world_value_changed"] + # So is an override on one side that the other side no longer applies. + record = record_for(changed(baseline_world=0.3)) + assert record["flags"] == ["construction"] + only_current_law = copy.deepcopy(premise) + only_current_law["events"][EVENT]["measures"][KEY]["years"]["2024"]["paths"][ + "gov.tax.rate" + ]["current_law"] = 0.4 + head["premises"] = only_current_law + record = next( + r + for r in diff.diff_sides(base, head)["rows"] + if r["level"] == "measure_year" and r["year"] == 2024 + ) + assert record["flags"] == [] + assert "current_law_changed" in record["context"] + flipped = copy.deepcopy(premise) + flipped["events"][EVENT]["measures"][KEY]["checks"][ + "reversal_differs_from_current_law" + ]["holds"] = False + head["premises"] = flipped + assert any( + d["code"] == "premise_check_changed" + for r in diff.diff_sides(base, head)["rows"] + for d in r["details"] + ) + + +def test_swapping_sides_swaps_values_and_keeps_flags(ns): + a = ns("a") + b = ns( + "b", + effects={ + 2024: {"income_tax": -40e6, "universal_credit": -30e6}, + 2025: {"income_tax": 0.0, "universal_credit": -60e6}, + }, + variant="pe6_uk2_build", + ) + forward = { + (r["level"], r["key"], r["year"]): r for r in diff.diff_sides(a, b)["rows"] + } + backward = { + (r["level"], r["key"], r["year"]): r for r in diff.diff_sides(b, a)["rows"] + } + assert forward.keys() == backward.keys() + for key, record in forward.items(): + other = backward[key] + assert record["flags"] == other["flags"] + assert (record["base_gbp"], record["head_gbp"]) == ( + other["head_gbp"], + other["base_gbp"], + ) + if record["delta_gbp"] is not None: + assert record["delta_gbp"] == -other["delta_gbp"] + + +def test_measure_year_change_equals_the_sum_of_head_changes(ns): + a = ns("a") + b = ns( + "b", + effects={ + 2024: {"income_tax": 123.456e6, "universal_credit": -7.25e6}, + 2025: {"income_tax": 1e9, "universal_credit": -2e9}, + }, + ) + rows = diff.diff_sides(a, b)["rows"] + for year in (2024, 2025): + total = next( + r for r in rows if r["level"] == "measure_year" and r["year"] == year + ) + heads = [r for r in rows if r["level"] == "head" and r["year"] == year] + assert total["delta_gbp"] == pytest.approx( + sum(h["delta_gbp"] for h in heads), abs=1e-3 + ) + + +def test_output_bytes_do_not_depend_on_input_order(ns): + a = ns("a") + b = ns( + "b", + effects={ + 2024: {"income_tax": 50e6, "universal_credit": -30e6}, + 2025: {"income_tax": 0.0, "universal_credit": -3e6}, + }, + ) + first = diff.render(a, b) + for seed in range(5): + rng = random.Random(seed) + shuffled = copy.copy(b) + shuffled["events"] = {} + for event, data in b["events"].items(): + data = dict(data) + items = list(data["rows"].items()) + rng.shuffle(items) + data["rows"] = dict(items) + artifacts = list(data["artifacts"].items()) + rng.shuffle(artifacts) + data["artifacts"] = dict(artifacts) + shuffled["events"][event] = data + assert diff.render(a, shuffled) == first + + +def test_packets_cover_flagged_moves_including_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__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"] == [ + {"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() + 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) + + +@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) + + +@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) + ] + + +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 + + +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]