From 0e741e3546355c28e9e06cf172ad260aa468a4be Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 19 Aug 2026 22:22:43 -0400 Subject: [PATCH 1/4] Add the UK national calibration step over ledger-backed target references Part of #623 and #701: a named national stage that resolves activated references against pinned chronicle facts (any unresolvable activated reference aborts the build), builds the benefit-unit target matrix, and calibrates household weights through microcosm-calibrate's public front door. Activates the UC caseload and UC-in-cap expenditure references at period 2025. Post-calibration aggregate scalers stay out of scope per the open adjudication on #623. --- .../623-uk-national-calibration.added.md | 1 + .../src/microcosm/build/country_spec.py | 1 + .../src/microcosm/build/uk/gates.json | 8 + .../microcosm/build/uk/target_references.json | 32 +++ .../build/uk_runtime/battery_bindings.py | 28 +++ .../build/uk_runtime/national_build.py | 13 + .../build/uk_runtime/national_calibration.py | 227 ++++++++++++++++++ .../tests/test_country_spec.py | 1 + .../tests/test_uk_national_build.py | 5 +- .../tests/test_uk_national_calibration.py | 144 +++++++++++ .../tests/test_uk_target_references.py | 32 ++- tools/build_uk_national_dataset.py | 45 ++++ 12 files changed, 522 insertions(+), 15 deletions(-) create mode 100644 changelog.d/623-uk-national-calibration.added.md create mode 100644 packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py create mode 100644 packages/microcosm-build/tests/test_uk_national_calibration.py diff --git a/changelog.d/623-uk-national-calibration.added.md b/changelog.d/623-uk-national-calibration.added.md new file mode 100644 index 000000000..e1a1a3c29 --- /dev/null +++ b/changelog.d/623-uk-national-calibration.added.md @@ -0,0 +1 @@ +Added a fail-closed UK national calibration stage that resolves activated Chronicle-backed target references, prepares their declared measures, calibrates household weights and records target diagnostics and solve evidence. diff --git a/packages/microcosm-build/src/microcosm/build/country_spec.py b/packages/microcosm-build/src/microcosm/build/country_spec.py index fcf306daf..d12348341 100644 --- a/packages/microcosm-build/src/microcosm/build/country_spec.py +++ b/packages/microcosm-build/src/microcosm/build/country_spec.py @@ -106,6 +106,7 @@ ALLOWED_GATE_FUNCTIONS = frozenset( { "aggregate_admin", + "calibration_reference_coverage", "degenerate_release_surface", "enum_domain", "export_surface", diff --git a/packages/microcosm-build/src/microcosm/build/uk/gates.json b/packages/microcosm-build/src/microcosm/build/uk/gates.json index d397abbf1..b258907ed 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/gates.json +++ b/packages/microcosm-build/src/microcosm/build/uk/gates.json @@ -253,6 +253,14 @@ }, "notes": "The household BRMA assignment must remain inside the PolicyEngine-UK brma enum domain." }, + { + "id": "uk_calibration_reference_coverage", + "gate": "calibration_reference_coverage", + "phase": "terminal", + "criticality": "release_blocking", + "parameters": {}, + "notes": "Every activated UK national target reference must resolve and enter the calibration matrix; count mismatch blocks release." + }, { "id": "uk_target_surface", "gate": "target_surface", diff --git a/packages/microcosm-build/src/microcosm/build/uk/target_references.json b/packages/microcosm-build/src/microcosm/build/uk/target_references.json index eb524850a..10c1ac1e4 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/target_references.json +++ b/packages/microcosm-build/src/microcosm/build/uk/target_references.json @@ -252,6 +252,38 @@ "contract_target_id": "ons.public_sector_employment", "measure_kind": "prepared_column" } + }, + { + "name": "dwp.uc.households", + "ledger_selector": { + "source_name": "dwp", + "source_concept": "dwp.uc_benefit_units", + "geography_level": "country" + }, + "entity": "benunit", + "measure": "dwp/uc/households", + "family": "dwp_universal_credit", + "period": 2025, + "metadata": { + "contract_target_id": "dwp.uc.households", + "measure_kind": "prepared_column" + } + }, + { + "name": "obr.universal_credit_in_cap", + "ledger_selector": { + "source_name": "obr", + "source_concept": "obr.universal_credit_in_cap", + "geography_level": "country" + }, + "entity": "benunit", + "measure": "obr/universal_credit_in_cap", + "family": "obr", + "period": 2025, + "metadata": { + "contract_target_id": "obr.universal_credit_in_cap", + "measure_kind": "prepared_column" + } } ] } diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py index b4f0330d5..d163abdf4 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py @@ -209,6 +209,28 @@ def _evaluate_source_coverage( ) +def _evaluate_calibration_reference_coverage( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> GateResult: + if parameters: + raise ValueError("calibration_reference_coverage takes no parameters.") + evidence = context.artifacts["national_calibration"] + declared = int(evidence["activated_reference_count"]) + resolved = int(evidence["resolved_reference_count"]) + matrix = int(evidence["matrix_target_count"]) + passed = declared == resolved == matrix + return GateResult( + name="calibration_reference_coverage", + passed=passed, + failures=() + if passed + else ( + f"Activated/resolved/matrix target counts differ: {declared}/{resolved}/{matrix}.", + ), + details={"activated": declared, "resolved": resolved, "matrix": matrix}, + ) + + def _evaluate_nonnegative_columns( context: EvidenceContext, parameters: Mapping[str, Any] ) -> GateResult: @@ -729,6 +751,12 @@ def _evaluate_tail_concentration( needs_frame=False, evidence=_stage_names_evidence, ), + "calibration_reference_coverage": UKGateBinding( + name="calibration_reference_coverage", + evaluator=_evaluate_calibration_reference_coverage, + artifact_keys=frozenset({"national_calibration"}), + needs_frame=False, + ), "nonnegative_columns": UKGateBinding( name="nonnegative_columns", evaluator=_evaluate_nonnegative_columns, diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py index f02eea039..9c51d23ef 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py @@ -544,6 +544,9 @@ def build_uk_national_dataset( fit_weight_records = _stage_fit_weight_records(materialized_stages) if fit_weight_records is not None: artifacts["fit_weight_records"] = fit_weight_records + calibration_evidence = _stage_calibration_evidence(materialized_stages) + if calibration_evidence is not None: + artifacts["national_calibration"] = calibration_evidence if input_mass_reference is not None: artifacts["input_mass_reference"] = input_mass_reference if reviewed_input_mass_exclusions is not None: @@ -811,6 +814,16 @@ def _stage_fit_weight_records( return tuple(collected) +def _stage_calibration_evidence( + stages: tuple[PlanStage, ...], +) -> Mapping[str, object] | None: + for stage in stages: + if stage.name == "national_calibration": + manifest = getattr(stage.transform, "manifest", None) + return dict(manifest) if isinstance(manifest, Mapping) else {} + return None + + def _brma_enum_domain(engine: object) -> tuple[str, ...] | None: variable_getter = getattr(engine, "_variable", None) if not callable(variable_getter): diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py new file mode 100644 index 000000000..510a48e69 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py @@ -0,0 +1,227 @@ +"""Ledger-backed calibration stage for the UK national build.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from importlib import resources +from typing import Any + +import pandas as pd + +from microcosm.build.ledger_targets import compile_ledger_target_references +from microcosm.build.plan import Stage +from microcosm.calibrate import calibrate, effective_sample_size +from microcosm.frame import Frame + +__all__ = ["UKNationalCalibrationStage", "uk_national_calibration_stage"] + + +class UKNationalCalibrationStage: + """Fail-closed national calibration transform and its manifest evidence.""" + + def __init__( + self, + facts: Sequence[Mapping[str, Any]], + *, + references: Sequence[object] | None = None, + epochs: int = 256, + learning_rate: float = 0.02, + max_weight_ratio: float = 10.0, + seed: int = 0, + ) -> None: + from microcosm.build.country_spec import load_country_spec + + self.facts = tuple(facts) + self.references = tuple( + load_country_spec("uk").target_references + if references is None + else references + ) + self.epochs = epochs + self.learning_rate = learning_rate + self.max_weight_ratio = max_weight_ratio + self.seed = seed + self.manifest: dict[str, object] | None = None + self.diagnostics: tuple[dict[str, object], ...] = () + + def __call__(self, frame: Frame) -> Frame: + registry = compile_ledger_target_references( + self.facts, self.references, country="uk" + ) + declared = len(self.references) + resolved = len(registry.specs) + if resolved != declared: + raise RuntimeError( + "UK national calibration resolved " + f"{resolved} of {declared} activated target references." + ) + prepared = _prepare_target_columns(frame, registry.specs) + result = calibrate( + prepared, + registry.to_target_set(), + weight_entity="household", + epochs=self.epochs, + learning_rate=self.learning_rate, + max_weight_ratio=self.max_weight_ratio, + seed=self.seed, + ) + if result.skipped or len(result.problem.names) != declared: + skipped = [item.name for item in result.skipped] + raise RuntimeError( + "UK national calibration matrix did not contain every activated " + f"reference: declared={declared}, rows={len(result.problem.names)}, " + f"skipped={skipped}." + ) + self.diagnostics = tuple( + { + "name": row.name, + "estimate": row.final_estimate, + "target": row.target, + "relative_error": row.relative_error, + } + for row in result.diagnostics + ) + ratios = result.weights / result.initial_weights + self.manifest = { + "activated_reference_count": declared, + "resolved_reference_count": resolved, + "matrix_target_count": len(result.problem.names), + "loss": result.final_loss, + "effective_sample_size": effective_sample_size(result.weights), + "max_weight_ratio": float(ratios.max()), + "max_weight_ratio_bound": self.max_weight_ratio, + } + return result.frame + + def checkpoint_metadata(self) -> Mapping[str, object]: + if self.manifest is None: + raise RuntimeError("UK national calibration has not run.") + return {"calibration": self.manifest, "diagnostics": self.diagnostics} + + +def uk_national_calibration_stage( + facts: Sequence[Mapping[str, Any]], **kwargs: Any +) -> Stage: + """Return the named ordered-stage entry for national calibration.""" + + transform = UKNationalCalibrationStage(facts, **kwargs) + return Stage(name="national_calibration", transform=transform) + + +def _contract_targets() -> dict[str, Mapping[str, Any]]: + payload = json.loads( + resources.files("microcosm.build.uk") + .joinpath("uk_national_targets.json") + .read_text() + ) + return {row["target_id"]: row for row in payload["targets"]} + + +def _prepare_target_columns(frame: Frame, specs: Sequence[object]) -> Frame: + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables.update({name: frame.link(name).copy() for name in frame.links}) + contracts = _contract_targets() + for spec in specs: + target_id = spec.metadata["contract_target_id"] + binding = contracts[target_id]["bindings"]["policyengine"] + entity = spec.entity + table = tables[entity] + if spec.measure in table: + continue + value_name = binding["value_variable"] + if value_name in {"person_count", "household_count", "benunit_count"}: + values = pd.Series(1.0, index=table.index) + else: + if value_name not in table: + raise ValueError( + f"Activated UK target {target_id!r} requires missing " + f"{entity} column {value_name!r}." + ) + values = table[value_name].astype(float) + mask = pd.Series(True, index=table.index) + for condition in binding.get("filters", ()): + variable = condition["variable"] + if variable not in table: + raise ValueError( + f"Activated UK target {target_id!r} requires missing " + f"{entity} filter column {variable!r}." + ) + mask &= _compare(table[variable], condition) + if binding.get("household_conditions"): + if entity != "household": + raise ValueError( + f"Activated UK target {target_id!r} declares household " + f"conditions on entity {entity!r}." + ) + for condition in binding["household_conditions"]: + mask &= _household_condition(tables, condition, table) + table[spec.measure] = values.where(mask, 0.0) + weights = {entity: frame.weights_for(entity) for entity in frame.weighted_entities} + return Frame( + tables, + frame.schema, + weights, + frame.strata, + mass_log=frame.mass_log, + metadata=frame.metadata, + ) + + +def _compare(values: pd.Series, condition: Mapping[str, Any]) -> pd.Series: + operator = condition.get("operator") + if operator is None: + operator, expected = "==", condition["equals"] + else: + expected = condition["value"] + operations = { + "==": values.eq, + ">": values.gt, + ">=": values.ge, + "<": values.lt, + "<=": values.le, + } + if operator not in operations: + raise ValueError(f"Unsupported UK target condition operator {operator!r}.") + return operations[operator](expected) + + +def _household_condition( + tables: Mapping[str, pd.DataFrame], + condition: Mapping[str, Any], + households: pd.DataFrame, +) -> pd.Series: + entity = condition["entity"] + source = tables[entity] + if entity == "household": + household_ids = source["household_id"] + else: + people = tables["person"] + entity_membership = f"person_{entity}_id" + if entity_membership not in people: + raise ValueError(f"UK person table is missing {entity_membership!r}.") + group_to_household = ( + people[[entity_membership, "person_household_id"]] + .drop_duplicates() + .set_index(entity_membership)["person_household_id"] + ) + if group_to_household.index.has_duplicates: + raise ValueError(f"UK {entity} groups span multiple households.") + household_ids = source[f"{entity}_id"].map(group_to_household) + reduce = condition["reduce"] + if reduce == "any": + matched = _compare(source[condition["variable"]], condition) + aggregate = matched.groupby(household_ids).any().astype(float) + expected_condition = {"operator": "==", "value": True} + elif reduce == "sum": + aggregate = source[condition["variable"]].groupby(household_ids).sum() + expected_condition = condition + elif reduce == "count": + aggregate = source[condition["variable"]].groupby(household_ids).count() + expected_condition = condition + else: + raise ValueError(f"Unsupported UK household reduction {reduce!r}.") + ids = households["household_id"] + return _compare(ids.map(aggregate).fillna(0.0), expected_condition).set_axis( + households.index + ) diff --git a/packages/microcosm-build/tests/test_country_spec.py b/packages/microcosm-build/tests/test_country_spec.py index de592a46c..943655020 100644 --- a/packages/microcosm-build/tests/test_country_spec.py +++ b/packages/microcosm-build/tests/test_country_spec.py @@ -582,6 +582,7 @@ def test_declares_the_full_june_battery(self, manifest) -> None: "uk_export_surface", "uk_take_up_signal", "uk_brma_enum_domain", + "uk_calibration_reference_coverage", "uk_target_surface", "uk_target_fit", "uk_input_mass_parity", diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index 45d77959f..9614b7ae6 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -1008,8 +1008,9 @@ def test_national_build_real_terminal_batch_blocks_incomplete_qrf_before_staging "uk_brma_enum_domain": "passed", # The legacy report omitted unevidenced gates; the battery names # every gap — non-blocking off the release-candidate posture. - "uk_export_surface": "evidence_absent", - "uk_target_surface": "evidence_absent", + "uk_export_surface": "evidence_absent", + "uk_calibration_reference_coverage": "evidence_absent", + "uk_target_surface": "evidence_absent", "uk_target_fit": "evidence_absent", "uk_input_mass_parity": "evidence_absent", "uk_qrf_tail_concentration": "failed", diff --git a/packages/microcosm-build/tests/test_uk_national_calibration.py b/packages/microcosm-build/tests/test_uk_national_calibration.py new file mode 100644 index 000000000..c28281a42 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_national_calibration.py @@ -0,0 +1,144 @@ +"""The UK national Ledger-backed calibration stage.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.ledger_targets import LedgerTargetReference +from microcosm.build.uk_runtime.national_calibration import ( + UKNationalCalibrationStage, +) +from microcosm.build.uk_runtime.national_frame import validate_uk_national_frame +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights + + +def _uc_reference(**overrides) -> LedgerTargetReference: + values = { + "name": "dwp.uc.households", + "ledger_selector": { + "source_name": "dwp", + "source_concept": "dwp.uc_benefit_units", + "geography_level": "country", + }, + "entity": "benunit", + "measure": "dwp/uc/households", + "family": "dwp_uc", + "period": 2025, + "metadata": {"contract_target_id": "dwp.uc.households"}, + } + values.update(overrides) + return LedgerTargetReference(**values) + + +def _fact( + *, + concept: str = "dwp.uc_benefit_units", + source_name: str = "dwp", + value: float = 30.0, +) -> dict: + return { + "aggregate_fact_key": "ledger.aggregate_fact.v2:uc-fixture", + "aggregation": {"method": "sum"}, + "assertion": "observation", + "geography": {"level": "country", "id": "K02000001"}, + "observed_measure": { + "source_name": source_name, + "source_concept": concept, + "source_measure_id": "total_units", + "unit": "count", + }, + "period": {"type": "month", "value": "2025-12"}, + "value": value, + } + + +def _frame() -> Frame: + ids = np.arange(4, dtype="int64") + return Frame( + { + "person": pd.DataFrame( + { + "person_id": ids, + "person_benunit_id": ids, + "person_household_id": ids, + } + ), + "benunit": pd.DataFrame( + {"benunit_id": ids, "universal_credit": [1.0, 1.0, 0.0, 0.0]} + ), + "household": pd.DataFrame({"household_id": ids}), + }, + EntitySchema(group_entities=("benunit", "household")), + {"household": Weights(np.full(4, 10.0), WeightKind.DESIGN)}, + metadata={"time_period": "2023"}, + ) + + +def test_uc_calibration_compiles_and_moves_weighted_count_towards_fact() -> None: + frame = _frame() + stage = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=200, learning_rate=0.05 + ) + + result = stage(frame) + + before = 20.0 + after = float(result.weights_for("household").values[:2].sum()) + assert abs(after - 30.0) < abs(before - 30.0) + assert stage.manifest["activated_reference_count"] == 1 + assert stage.manifest["resolved_reference_count"] == 1 + assert stage.manifest["matrix_target_count"] == 1 + assert stage.diagnostics[0]["target"] == 30.0 + + +def test_activated_unresolvable_reference_aborts_loudly() -> None: + stage = UKNationalCalibrationStage( + [_fact(concept="different")], references=[_uc_reference()], epochs=1 + ) + + with pytest.raises(ValueError, match="did not match a Ledger fact selector"): + stage(_frame()) + + +def test_chronicle_184_uc_and_obr_references_compile_fail_closed() -> None: + from microcosm.build.country_spec import load_country_spec + from microcosm.build.ledger_targets import compile_ledger_target_references + + references = tuple( + reference + for reference in load_country_spec("uk").target_references + if reference.name in {"dwp.uc.households", "obr.universal_credit_in_cap"} + ) + registry = compile_ledger_target_references( + [ + _fact(), + _fact( + concept="obr.universal_credit_in_cap", + source_name="obr", + value=40_000_000_000, + ), + ], + references, + country="uk", + ) + + assert {spec.name for spec in registry.specs} == { + "dwp.uc.households", + "obr.universal_credit_in_cap", + } + + +def test_calibration_preserves_entity_ids_and_national_integrity() -> None: + frame = _frame() + stage = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=5 + ) + + result = stage(frame) + + for entity in frame.entities: + id_column = f"{entity}_id" + assert result.table(entity)[id_column].equals(frame.table(entity)[id_column]) + validate_uk_national_frame(result) diff --git a/packages/microcosm-build/tests/test_uk_target_references.py b/packages/microcosm-build/tests/test_uk_target_references.py index 02bff1893..6a5ae5804 100644 --- a/packages/microcosm-build/tests/test_uk_target_references.py +++ b/packages/microcosm-build/tests/test_uk_target_references.py @@ -19,7 +19,7 @@ from microcosm.calibrate.matrix import build_constraint_matrix from microcosm.frame import EntitySchema, Frame, WeightKind, Weights -ACTIVE_REFERENCE_COUNT = 13 +ACTIVE_REFERENCE_COUNT = 15 FIXTURE_FEED_ROWS = ( Path(__file__).parent / "fixtures" / "uk_target_reference_feed_rows.jsonl" @@ -71,7 +71,16 @@ def test_uk_target_references_follow_contract_derivation_rules() -> None: assert reference["entity"] == _expected_reference_entity(target) assert reference["measure"] == binding["metric_name"] assert reference["family"] == target["family"] - assert reference["period"] == 2023 + expected_period = ( + 2025 + if contract_target_id + in { + "dwp.uc.households", + "obr.universal_credit_in_cap", + } + else 2023 + ) + assert reference["period"] == expected_period assert reference["metadata"] == { "contract_target_id": contract_target_id, "measure_kind": "prepared_column", @@ -124,8 +133,7 @@ def test_uk_target_references_compile_from_real_staged_feed_rows() -> None: assert savings_interest.measure == "ons/savings_interest_income" assert savings_interest.family == "ons_national_accounts" assert ( - savings_interest.metadata["contract_target_id"] - == "ons.savings_interest_income" + savings_interest.metadata["contract_target_id"] == "ons.savings_interest_income" ) assert ( savings_interest.metadata["ledger_aggregate_fact_key"] @@ -145,7 +153,9 @@ def test_uk_target_references_constrain_a_frame_with_prepared_columns() -> None: """ spec = load_country_spec("uk") - references = spec.target_references + references = tuple( + reference for reference in spec.target_references if reference.period == 2023 + ) feed_rows = [ json.loads(line) for line in FIXTURE_FEED_ROWS.read_text().splitlines() @@ -153,7 +163,7 @@ def test_uk_target_references_constrain_a_frame_with_prepared_columns() -> None: ] registry = compile_ledger_target_references(feed_rows, references, country="uk") - assert len(registry.specs) == ACTIVE_REFERENCE_COUNT + assert len(registry.specs) == 13 n_households = 3 weights = np.array([10.0, 20.0, 30.0]) @@ -162,9 +172,7 @@ def test_uk_target_references_constrain_a_frame_with_prepared_columns() -> None: expected_aggregates: dict[str, float] = {} for index, compiled in enumerate(registry.specs): column = np.array([index + 1.0, 2.0 * (index + 1.0), 0.0]) - columns = ( - person_columns if compiled.entity == "person" else household_columns - ) + columns = person_columns if compiled.entity == "person" else household_columns columns[compiled.measure] = column expected_aggregates[f"{compiled.name}@{compiled.period}"] = float( (column * weights).sum() @@ -191,7 +199,7 @@ def test_uk_target_references_constrain_a_frame_with_prepared_columns() -> None: problem = build_constraint_matrix(frame, registry.to_target_set()) assert problem.skipped == () - assert len(problem.names) == ACTIVE_REFERENCE_COUNT + assert len(problem.names) == 13 achieved = problem.matrix @ problem.initial_weights.values for name, estimate in zip(problem.names, achieved, strict=True): assert estimate == expected_aggregates[name], name @@ -199,9 +207,7 @@ def test_uk_target_references_constrain_a_frame_with_prepared_columns() -> None: f"{compiled.name}@{compiled.period}": compiled.value for compiled in registry.specs } - for name, target_value in zip( - problem.names, problem.target_vector, strict=True - ): + for name, target_value in zip(problem.names, problem.target_vector, strict=True): assert target_value == fact_values_by_name[name], name diff --git a/tools/build_uk_national_dataset.py b/tools/build_uk_national_dataset.py index 8839ce9f4..f8ecee0b6 100644 --- a/tools/build_uk_national_dataset.py +++ b/tools/build_uk_national_dataset.py @@ -15,6 +15,7 @@ from microcosm.build.country_spec import country_stage_plan, load_country_spec from microcosm.build.gate_battery import GateBatteryBlockedError +from microcosm.build.ledger_artifact import load_ledger_consumer_artifact from microcosm.build.logbook import canonical_json_bytes from microcosm.build.logbook_adoption import ( AttemptState, @@ -30,6 +31,7 @@ sha256_argument, write_error_receipt, ) +from microcosm.build.plan import Stage as PlanStage from microcosm.build.uk_runtime.cgt_imputation import ( uk_capital_gains_imputation_stage, ) @@ -42,6 +44,9 @@ verify_certified_uk_candidate, ) from microcosm.build.uk_runtime.national_build import build_uk_national_dataset +from microcosm.build.uk_runtime.national_calibration import ( + UKNationalCalibrationStage, +) from microcosm.build.uk_runtime.national_frame import ( uk_household_weight_kind, uk_time_period, @@ -140,6 +145,16 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: "that will ship with this release." ), ) + parser.add_argument( + "--ledger-facts", + type=Path, + help="Pinned Chronicle consumer artifact used to resolve UK targets.", + ) + parser.add_argument( + "--national-calibration-diagnostics-json", + type=Path, + help="Per-target diagnostics emitted by the national calibration stage.", + ) parser.add_argument( "--frs-raw-dir", type=Path, @@ -821,6 +836,26 @@ def _main_recording( "checkpoint_dir": args.checkpoint_dir, "run_config": run_config, } + if (args.ledger_facts is None) != ( + args.national_calibration_diagnostics_json is None + ): + raise ValueError( + "--ledger-facts and --national-calibration-diagnostics-json must " + "be supplied together." + ) + if args.release_candidate and args.ledger_facts is None: + raise ValueError("a release candidate requires the national calibration stage.") + calibration_transform = None + calibration_stages: tuple[PlanStage, ...] = () + if args.ledger_facts is not None: + ledger_artifact = load_ledger_consumer_artifact(args.ledger_facts) + calibration_transform = UKNationalCalibrationStage(ledger_artifact.facts) + calibration_stages = ( + PlanStage( + name="national_calibration", + transform=calibration_transform, + ), + ) result = build_uk_national_dataset( input_h5=args.input_h5, staging_h5=args.staging_h5, @@ -843,6 +878,7 @@ def _main_recording( # bespoke uk/cgt_source_stages.json; absorbing it into the # canonical source_stages.json is WS-E follow-up work. uk_capital_gains_imputation_stage(args.cgt_ods), + *calibration_stages, ), **gate_path_argument, **weighted_integrity_arguments, @@ -851,6 +887,15 @@ def _main_recording( sample_seed=args.sample_seed, release_candidate=args.release_candidate, ) + if calibration_transform is not None: + _write_json( + args.national_calibration_diagnostics_json, + { + "schema_version": 1, + "targets": list(calibration_transform.diagnostics), + "calibration": calibration_transform.manifest, + }, + ) append_phase(state, "build_completed") _write_stage_reports( evidence_path=evidence_path, From 9dcc83b418a6144c3ed09eda2855a8d13f4451a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Thu, 20 Aug 2026 16:32:49 +0200 Subject: [PATCH 2/4] Reconcile the coverage gate with main's unarmed-battery enumeration Rebasing over merged #727 (and #730 beneath it): main's TestUnevidencedArms now enumerates the closed set of gates that report evidence_absent when the battery runs unarmed, so the new uk_calibration_reference_coverage gate joins that set. Also squares the indentation of the earlier enumeration update in the national-build terminal-batch test. Co-Authored-By: Claude Fable 5 --- packages/microcosm-build/tests/test_uk_battery_bindings.py | 1 + packages/microcosm-build/tests/test_uk_national_build.py | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index 0feec00f2..fe4d5bda2 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -503,6 +503,7 @@ def test_battery_records_evidence_absent(self, uk_gates) -> None: assert set(absent) == { "uk_weights_audit", "uk_export_surface", + "uk_calibration_reference_coverage", "uk_target_surface", "uk_target_fit", "uk_input_mass_parity", diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index 9614b7ae6..932454c9c 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -1008,9 +1008,9 @@ def test_national_build_real_terminal_batch_blocks_incomplete_qrf_before_staging "uk_brma_enum_domain": "passed", # The legacy report omitted unevidenced gates; the battery names # every gap — non-blocking off the release-candidate posture. - "uk_export_surface": "evidence_absent", - "uk_calibration_reference_coverage": "evidence_absent", - "uk_target_surface": "evidence_absent", + "uk_export_surface": "evidence_absent", + "uk_calibration_reference_coverage": "evidence_absent", + "uk_target_surface": "evidence_absent", "uk_target_fit": "evidence_absent", "uk_input_mass_parity": "evidence_absent", "uk_qrf_tail_concentration": "failed", From b5c926e703f722405511f98f021a43afe9eaecb5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Thu, 20 Aug 2026 17:55:16 +0200 Subject: [PATCH 3/4] Re-pin the UK gate-battery vintage digests for the coverage gate Adding uk_calibration_reference_coverage to uk/gates.json moves the three vintage pins (policy sha256, gates-manifest sha256, spec fingerprint) and the entry-gates mirror in the data contract, plus the schema-3-style local copies in the data shard's contract tests. The new entry carries no legacy detail schema and contributes no evidence digest, so the legacy-name and evidence-id mirrors are unchanged; the all-passing fixture mirrors the evaluator's activated/resolved/matrix detail block. Co-Authored-By: Claude Fable 5 --- .../microcosm-data/src/microcosm/data/contract.py | 10 +++++++--- packages/microcosm-data/tests/test_contract.py | 14 +++++++++++--- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/packages/microcosm-data/src/microcosm/data/contract.py b/packages/microcosm-data/src/microcosm/data/contract.py index 2bdd2f18b..fc0790570 100644 --- a/packages/microcosm-data/src/microcosm/data/contract.py +++ b/packages/microcosm-data/src/microcosm/data/contract.py @@ -344,13 +344,13 @@ # fingerprint derives from the manifest digest. Editing the spec moves all # three here in the same reviewed change. _UK_GATE_BATTERY_POLICY_SHA256 = ( - "91ba70060b87eeca1e35d2aebe2ad79da61e33105b8f1352a2a05846e0780d4b" + "728a5fe2f543f59e2797e4227269fe4516508a274b4fa5fe49559387d6b9d686" ) _UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( - "4092f5012cddc4c878ea3a727c09f23212475ca246a4555ecea6f39219656a98" + "f01e5459debc6f3ebfa097591749377b640a5d633e3df3825575b4ec15eeacb2" ) _UK_GATE_BATTERY_SPEC_FINGERPRINT = ( - "59f050a3a1ef1364107140083d548a873a9494b11472d4e4fd2a86f64ea8bb6b" + "f358121fefc6e0e2371956dc0628c1d997ee3735e558ca3dd4526326a6add78b" ) #: Spec entry id -> the legacy gate name whose observable detail checks #: apply unchanged (the battery re-keys the report by entry id; the gate @@ -395,6 +395,10 @@ "uk_export_surface": ("export_surface", "terminal"), "uk_take_up_signal": ("take_up_signal", "terminal"), "uk_brma_enum_domain": ("enum_domain", "terminal"), + "uk_calibration_reference_coverage": ( + "calibration_reference_coverage", + "terminal", + ), "uk_target_surface": ("target_surface", "terminal"), "uk_target_fit": ("target_fit", "terminal"), "uk_input_mass_parity": ("input_mass_parity", "terminal"), diff --git a/packages/microcosm-data/tests/test_contract.py b/packages/microcosm-data/tests/test_contract.py index 77ddad51d..6d5cc69cf 100644 --- a/packages/microcosm-data/tests/test_contract.py +++ b/packages/microcosm-data/tests/test_contract.py @@ -144,13 +144,13 @@ def _trusted_terminal_gate_signing_key(monkeypatch) -> None: UK_GATE_BATTERY_PRODUCER = "microcosm.build.gate_battery" UK_GATE_BATTERY_SIGNING_KEY_ENV = "MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY" UK_GATE_BATTERY_POLICY_SHA256 = ( - "91ba70060b87eeca1e35d2aebe2ad79da61e33105b8f1352a2a05846e0780d4b" + "728a5fe2f543f59e2797e4227269fe4516508a274b4fa5fe49559387d6b9d686" ) UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( - "4092f5012cddc4c878ea3a727c09f23212475ca246a4555ecea6f39219656a98" + "f01e5459debc6f3ebfa097591749377b640a5d633e3df3825575b4ec15eeacb2" ) UK_GATE_BATTERY_SPEC_FINGERPRINT = ( - "59f050a3a1ef1364107140083d548a873a9494b11472d4e4fd2a86f64ea8bb6b" + "f358121fefc6e0e2371956dc0628c1d997ee3735e558ca3dd4526326a6add78b" ) UK_GATE_BATTERY_DEGENERATE_EVIDENCE_SHA256 = ( "d0d024043132fa07c378c393dbe2b24fe99bf19e876bcc39997d2c80cc9bd4f6" @@ -190,6 +190,11 @@ def _trusted_terminal_gate_signing_key(monkeypatch) -> None: "uk_export_surface": ("export_surface", "terminal", "export_surface"), "uk_take_up_signal": ("take_up_signal", "terminal", "take_up_signal"), "uk_brma_enum_domain": ("enum_domain", "terminal", "enum_domain"), + "uk_calibration_reference_coverage": ( + "calibration_reference_coverage", + "terminal", + None, + ), "uk_target_surface": ("target_surface", "terminal", "target_surface"), "uk_target_fit": ("target_fit", "terminal", "target_fit"), "uk_input_mass_parity": ("input_mass_parity", "terminal", "input_mass_parity"), @@ -1033,6 +1038,9 @@ def _gate_battery_payload( details: dict = {"check": "manifest_current"} elif entry_id == "uk_release_family_build_stages": details = {"stage_names": list(stage_names)} + elif entry_id == "uk_calibration_reference_coverage": + # Mirrors _evaluate_calibration_reference_coverage's detail block. + details = {"activated": 15, "resolved": 15, "matrix": 15} else: details = _terminal_gate_details(detail_name) gates[entry_id] = { From 6c8a95aafc3a2bd1e30ef140fa84be85a7444cb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Thu, 20 Aug 2026 22:47:02 +0200 Subject: [PATCH 4/4] Restore calibration evidence on checkpoint resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The calibration stage exposed checkpoint_metadata but no resume_from_checkpoint, so a resumed build left manifest None, the evidence getter handed the coverage gate an empty mapping, and the release-blocking uk_calibration_reference_coverage entry failed closed on a KeyError despite a completed calibration checkpoint (the driver sidecar wrote null diagnostics). Mirror the retained-leaves and SPI restoration precedents: record the output frame's content identity in the checkpoint record, and rehydrate manifest/diagnostics on resume with fail-closed validation of the three coverage counts and a drifted-record refusal. Tests cover the JSON round-trip and its refusals, the resumed evidence feeding the real coverage evaluator, and an end-to-end crash-resume build asserting the gate passes without re-executing calibration. The e2e cleanup stage drops the stage's prepared scratch column before the staging write — the pandas staging writer refuses "/" in column names, a latent issue for real armed builds recorded on the PR. Implemented via Codex (gpt-5.4) from the reviewed plan. Co-Authored-By: Claude Fable 5 --- .../build/uk_runtime/national_calibration.py | 48 +++- .../tests/test_uk_national_build.py | 243 ++++++++++++++++-- .../tests/test_uk_national_calibration.py | 48 ++++ 3 files changed, 323 insertions(+), 16 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py index 510a48e69..e56cd725c 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_calibration.py @@ -11,6 +11,7 @@ from microcosm.build.ledger_targets import compile_ledger_target_references from microcosm.build.plan import Stage +from microcosm.build.uk_runtime.content_identity import uk_frame_content_identity from microcosm.calibrate import calibrate, effective_sample_size from microcosm.frame import Frame @@ -44,6 +45,7 @@ def __init__( self.seed = seed self.manifest: dict[str, object] | None = None self.diagnostics: tuple[dict[str, object], ...] = () + self.output_content_identity: str | None = None def __call__(self, frame: Frame) -> Frame: registry = compile_ledger_target_references( @@ -92,12 +94,56 @@ def __call__(self, frame: Frame) -> Frame: "max_weight_ratio": float(ratios.max()), "max_weight_ratio_bound": self.max_weight_ratio, } + self.output_content_identity = uk_frame_content_identity(result.frame) return result.frame def checkpoint_metadata(self) -> Mapping[str, object]: if self.manifest is None: raise RuntimeError("UK national calibration has not run.") - return {"calibration": self.manifest, "diagnostics": self.diagnostics} + return { + "calibration": self.manifest, + "diagnostics": self.diagnostics, + "output_content_identity": self.output_content_identity, + } + + def resume_from_checkpoint( + self, + metadata: Mapping[str, object], + frame: Frame, + ) -> None: + """Rehydrate completed calibration evidence from its checkpoint record.""" + + calibration = metadata.get("calibration") + diagnostics = metadata.get("diagnostics") + output_identity = metadata.get("output_content_identity") + count_keys = ( + "activated_reference_count", + "resolved_reference_count", + "matrix_target_count", + ) + if ( + not isinstance(calibration, Mapping) + or not all(key in calibration for key in count_keys) + or not isinstance(diagnostics, list) + or not all(isinstance(row, Mapping) for row in diagnostics) + or not isinstance(output_identity, str) + or not output_identity + ): + raise RuntimeError( + "UK national calibration resume requires the checkpoint record " + "to carry calibration counts, diagnostics, and output content " + "identity; a record without them cannot feed the calibration " + "reference coverage gate or the drift check." + ) + if uk_frame_content_identity(frame) != output_identity: + raise RuntimeError( + "UK national calibration checkpoint content does not match its " + "recorded output identity; refusing to resume from a drifted " + "record." + ) + self.manifest = dict(calibration) + self.diagnostics = tuple(dict(row) for row in diagnostics) + self.output_content_identity = output_identity def uk_national_calibration_stage( diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index 932454c9c..e9c101110 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -3,6 +3,7 @@ import json from datetime import date, datetime from pathlib import Path +from types import SimpleNamespace import numpy as np import pandas as pd @@ -10,17 +11,26 @@ from microcosm.build.country_spec import country_stage_plan, load_country_spec from microcosm.build.gate_battery import ( + EvidenceContext, GateBatteryBlockedError, gate_signing_key_env, ) from microcosm.build.gates import FitWeightRecord, GateResult +from microcosm.build.ledger_targets import LedgerTargetReference from microcosm.build.plan import Stage, StagePlan -from microcosm.build.uk_runtime.battery_bindings import UKGateBinding +from microcosm.build.uk_runtime.battery_bindings import ( + UK_GATE_REGISTRY, + UKGateBinding, + _evaluate_calibration_reference_coverage, +) from microcosm.build.uk_runtime.national_build import ( UKNationalStage, build_uk_national_dataset, load_uk_national_frame, ) +from microcosm.build.uk_runtime.national_calibration import ( + UKNationalCalibrationStage, +) from microcosm.build.uk_runtime.national_frame import ( _uk_gate_surface, uk_household_weight_kind, @@ -205,6 +215,7 @@ def _write_two_row_h5( path: Path, *, employment_income: tuple[float, float] = (40_000.0, 55_000.0), + include_calibration_columns: bool = False, ) -> None: n = 100 household_ids = np.arange(1, n + 1) @@ -233,22 +244,25 @@ def flags(true_count: int) -> list[bool]: format="table", data_columns=True, ) + benunit = pd.DataFrame( + { + "benunit_id": benunit_ids, + "would_claim_child_benefit": flags(89), + "child_benefit_opts_out": flags(23), + "would_claim_pc": flags(70), + "would_claim_uc": flags(55), + "would_claim_tfc": flags(59), + "would_claim_extended_childcare": flags(81), + "would_claim_universal_childcare": flags(56), + "would_claim_targeted_childcare": flags(60), + "maximum_extended_childcare_hours_usage": np.linspace(1.0, 30.0, n), + } + ) + if include_calibration_columns: + benunit["universal_credit"] = flags(55) store.put( "benunit", - pd.DataFrame( - { - "benunit_id": benunit_ids, - "would_claim_child_benefit": flags(89), - "child_benefit_opts_out": flags(23), - "would_claim_pc": flags(70), - "would_claim_uc": flags(55), - "would_claim_tfc": flags(59), - "would_claim_extended_childcare": flags(81), - "would_claim_universal_childcare": flags(56), - "would_claim_targeted_childcare": flags(60), - "maximum_extended_childcare_hours_usage": np.linspace(1.0, 30.0, n), - } - ), + benunit, format="table", data_columns=True, ) @@ -333,6 +347,49 @@ def evaluator(context, parameters): return registry +def _registry_with_calibration() -> dict[str, UKGateBinding]: + registry = _registry_with_coverage(_passing_gate) + registry["calibration_reference_coverage"] = UK_GATE_REGISTRY[ + "calibration_reference_coverage" + ] + return registry + + +def _uc_reference(**overrides) -> LedgerTargetReference: + values = { + "name": "dwp.uc.households", + "ledger_selector": { + "source_name": "dwp", + "source_concept": "dwp.uc_benefit_units", + "geography_level": "country", + }, + "entity": "benunit", + "measure": "dwp/uc/households", + "family": "dwp_uc", + "period": 2025, + "metadata": {"contract_target_id": "dwp.uc.households"}, + } + values.update(overrides) + return LedgerTargetReference(**values) + + +def _calibration_fact(value: float = 60.0) -> dict: + return { + "aggregate_fact_key": "ledger.aggregate_fact.v2:uc-build-fixture", + "aggregation": {"method": "sum"}, + "assertion": "observation", + "geography": {"level": "country", "id": "K02000001"}, + "observed_measure": { + "source_name": "dwp", + "source_concept": "dwp.uc_benefit_units", + "source_measure_id": "total_units", + "unit": "count", + }, + "period": {"type": "month", "value": "2025-12"}, + "value": value, + } + + def test_driver_validates_the_uk_residue_after_each_stage( monkeypatch, tmp_path ) -> None: @@ -453,6 +510,45 @@ def test_weights_audit_details_carry_the_was_fit_records() -> None: assert resolved["uk_was_2018_20_wealth:cash_isa"] == "explicit" +def test_resumed_national_calibration_feeds_reference_coverage_gate( + tmp_path, +) -> None: + from microcosm.build.uk_runtime.national_build import ( + _stage_calibration_evidence, + ) + + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + _write_two_row_h5(input_h5, include_calibration_columns=True) + frame, _provenance = load_uk_national_frame(input_h5) + stage = UKNationalCalibrationStage( + [_calibration_fact()], + references=[_uc_reference()], + epochs=1, + ) + staged = stage(frame) + metadata = json.loads(json.dumps(stage.checkpoint_metadata())) + resumed = UKNationalCalibrationStage( + [_calibration_fact()], + references=[_uc_reference()], + epochs=1, + ) + + resumed.resume_from_checkpoint(metadata, staged) + evidence = _stage_calibration_evidence( + (SimpleNamespace(name="national_calibration", transform=resumed),) + ) + result = _evaluate_calibration_reference_coverage( + EvidenceContext(artifacts={"national_calibration": evidence}), + {}, + ) + + assert evidence == stage.manifest + assert result.passed + assert result.details == {"activated": 1, "resolved": 1, "matrix": 1} + + def test_national_build_runs_preflight_stages_gate_then_staging_write( monkeypatch, tmp_path ) -> None: @@ -1400,6 +1496,60 @@ def transform(frame: Frame) -> Frame: return UKNationalStage(name=name, transform=transform) +def _counting_cleanup_stage( + name: str, + calls: list[str] | None = None, +) -> UKNationalStage: + def transform(frame: Frame) -> Frame: + if calls is not None: + calls.append(name) + person = frame.table("person").copy() + person["employment_income"] = person["employment_income"] + 1.0 + benunit = frame.table("benunit").drop( + columns=["dwp/uc/households"], + errors="ignore", + ) + return uk_national_frame( + person=person, + benunit=benunit, + household=frame.table("household"), + time_period=uk_time_period(frame), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + + return UKNationalStage(name=name, transform=transform) + + +class _CountingCalibrationStage: + def __init__(self, calls: list[str]) -> None: + self.calls = calls + self.inner = UKNationalCalibrationStage( + [_calibration_fact()], + references=[_uc_reference()], + epochs=1, + ) + + @property + def manifest(self) -> dict[str, object] | None: + return self.inner.manifest + + def __call__(self, frame: Frame) -> Frame: + self.calls.append("national_calibration") + return self.inner(frame) + + def checkpoint_metadata(self) -> dict[str, object]: + return dict(self.inner.checkpoint_metadata()) + + def resume_from_checkpoint( + self, + metadata: dict[str, object], + frame: Frame, + ) -> None: + self.inner.resume_from_checkpoint(metadata, frame) + + def _assert_same_staging_payload(left: Path, right: Path) -> None: left_frame, _ = load_uk_national_frame(left) right_frame, _ = load_uk_national_frame(right) @@ -1508,6 +1658,69 @@ def exploding(frame: Frame) -> Frame: assert calls == ["two"] +def test_checkpointed_build_resumes_completed_calibration_evidence( + tmp_path, +) -> None: + pytest.importorskip("tables") + pytest.importorskip("h5py") + + registry = _registry_with_calibration() + input_h5 = tmp_path / "base.h5" + _write_two_row_h5(input_h5, include_calibration_columns=True) + run_config = {"input_sha256": "a" * 64, "seed": 42} + + def exploding(frame: Frame) -> Frame: + raise RuntimeError("boom") + + calibration_calls: list[str] = [] + with pytest.raises(RuntimeError, match="boom"): + _run_national_build( + coverage_engine=object(), + input_h5=input_h5, + staging_h5=tmp_path / "crashed.h5", + stages=( + UKNationalStage( + "national_calibration", + _CountingCalibrationStage(calibration_calls), + ), + UKNationalStage(name="after", transform=exploding), + ), + checkpoint_dir=tmp_path / "checkpoints", + run_config=run_config, + gate_registry=registry, + ) + assert calibration_calls == ["national_calibration"] + + resumed_calibration_calls: list[str] = [] + after_calls: list[str] = [] + result = _run_national_build( + coverage_engine=object(), + input_h5=input_h5, + staging_h5=tmp_path / "recovered.h5", + stages=( + UKNationalStage( + "national_calibration", + _CountingCalibrationStage(resumed_calibration_calls), + ), + _counting_cleanup_stage("after", after_calls), + ), + checkpoint_dir=tmp_path / "checkpoints", + run_config=run_config, + gate_registry=registry, + terminal_gate_path=tmp_path / "terminal_gates.json", + ) + + assert resumed_calibration_calls == [] + assert after_calls == ["after"] + calibration_gate = result.gate_report["gates"]["uk_calibration_reference_coverage"] + assert calibration_gate["status"] == "passed" + assert calibration_gate["details"] == { + "activated": 1, + "resolved": 1, + "matrix": 1, + } + + def test_checkpointed_build_pins_the_run_config(tmp_path) -> None: """Resuming under a different configuration is refused, never blended.""" diff --git a/packages/microcosm-build/tests/test_uk_national_calibration.py b/packages/microcosm-build/tests/test_uk_national_calibration.py index c28281a42..bff827548 100644 --- a/packages/microcosm-build/tests/test_uk_national_calibration.py +++ b/packages/microcosm-build/tests/test_uk_national_calibration.py @@ -2,6 +2,8 @@ from __future__ import annotations +import json + import numpy as np import pandas as pd import pytest @@ -142,3 +144,49 @@ def test_calibration_preserves_entity_ids_and_national_integrity() -> None: id_column = f"{entity}_id" assert result.table(entity)[id_column].equals(frame.table(entity)[id_column]) validate_uk_national_frame(result) + + +def test_checkpoint_metadata_round_trips_calibration_evidence() -> None: + frame = _frame() + stage = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=5 + ) + + staged = stage(frame) + metadata = json.loads(json.dumps(stage.checkpoint_metadata())) + + resumed = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=5 + ) + resumed.resume_from_checkpoint(metadata, staged) + + assert resumed.manifest == stage.manifest + assert resumed.diagnostics == stage.diagnostics + assert resumed.output_content_identity == metadata["output_content_identity"] + + drifted = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=5 + ) + with pytest.raises(RuntimeError, match="drifted record"): + drifted.resume_from_checkpoint(metadata, frame) + + empty = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=5 + ) + with pytest.raises(RuntimeError, match="calibration counts"): + empty.resume_from_checkpoint({}, staged) + + missing_count = dict(metadata) + missing_count["calibration"] = { + key: value + for key, value in metadata["calibration"].items() + if key != "activated_reference_count" + } + with pytest.raises(RuntimeError, match="calibration counts"): + empty.resume_from_checkpoint(missing_count, staged) + + unrun = UKNationalCalibrationStage( + [_fact()], references=[_uc_reference()], epochs=5 + ) + with pytest.raises(RuntimeError, match="has not run"): + unrun.checkpoint_metadata()