From 656c22a298d85ad90d8f292bff2c028838a384de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:19:51 +0200 Subject: [PATCH 1/7] Give the battery report a release_evidence slot Digests of release inputs the gates themselves do not consume (for the UK, the calibration-diagnostics digest) need a signed home once the legacy schema-3 attestation retires. The slot rides in the report body and the signed attestation, defaults to an empty mapping so the envelope key set is stable, and refuses non-string entries. Co-Authored-By: Claude Fable 5 --- .../src/microcosm/build/gate_battery.py | 17 +++++ .../tests/test_gate_battery.py | 67 +++++++++++++++++++ 2 files changed, 84 insertions(+) diff --git a/packages/microcosm-build/src/microcosm/build/gate_battery.py b/packages/microcosm-build/src/microcosm/build/gate_battery.py index dd48e3ca8..5a0ef54da 100644 --- a/packages/microcosm-build/src/microcosm/build/gate_battery.py +++ b/packages/microcosm-build/src/microcosm/build/gate_battery.py @@ -749,6 +749,11 @@ class GateBatteryRun: cannot excuse absent evidence; dev builds record it and continue. registry: Gate bindings; defaults to :data:`DEFAULT_REGISTRY`. + release_evidence: Digests of release inputs the gates themselves do + not consume but the release contract links (for the UK, the + calibration-diagnostics digest). Carried in the report and the + signed attestation, so the linkage survives the build that + attested it. """ def __init__( @@ -759,15 +764,25 @@ def __init__( report_path: Path | str, release_candidate: bool, registry: Mapping[str, GateBinding] = DEFAULT_REGISTRY, + release_evidence: Mapping[str, str] | None = None, ) -> None: if not isinstance(release_id, str) or not release_id.strip(): raise ValueError("release_id must be a non-empty string.") validate_gate_parameters(gates, registry) + evidence = dict(release_evidence or {}) + for key, value in evidence.items(): + if not isinstance(key, str) or not key.strip(): + raise ValueError("release_evidence keys must be non-empty strings.") + if not isinstance(value, str) or not value.strip(): + raise ValueError( + f"release_evidence[{key!r}] must be a non-empty string digest." + ) self._gates = gates self._release_id = release_id self._report_path = Path(report_path) self._release_candidate = bool(release_candidate) self._registry = dict(registry) + self._release_evidence = dict(sorted(evidence.items())) self._gates_manifest_sha256 = _canonical_sha256( _gates_manifest_payload(self._gates) ) @@ -936,6 +951,7 @@ def report_payload(self) -> dict[str, object]: "phases": list(self._gates.phases), "phases_evaluated": list(self._phase_reports), "blocked_at_phase": self._blocked_at_phase, + "release_evidence": dict(self._release_evidence), "evidence_sha256": dict(evidence_sha256), "gate_outcomes_sha256": _canonical_sha256(gates_payload), "signature_algorithm": GATE_BATTERY_SIGNATURE_ALGORITHM, @@ -961,6 +977,7 @@ def report_payload(self) -> dict[str, object]: "shippable": shippable, "gates": gates_payload, "policy_sha256": policy_sha256, + "release_evidence": dict(self._release_evidence), "evidence_sha256": dict(evidence_sha256), "attestation": attestation, } diff --git a/packages/microcosm-build/tests/test_gate_battery.py b/packages/microcosm-build/tests/test_gate_battery.py index bb10a958b..8b6920533 100644 --- a/packages/microcosm-build/tests/test_gate_battery.py +++ b/packages/microcosm-build/tests/test_gate_battery.py @@ -549,6 +549,73 @@ def payload_for(tolerance: float) -> dict: ), "a threshold outside the policy hash is not attested" +class TestReleaseEvidence: + """Digests of release inputs the gates do not consume ride in the report. + + The slot exists so a linkage like the UK calibration-diagnostics digest + keeps a signed home once the legacy schema-3 attestation retires: the + verifier reads it from the attestation, so it must be covered by the + signature and present (empty) even when unused. + """ + + def test_release_evidence_rides_in_the_report_and_is_signed( + self, tmp_path, signing_env + ): + manifest = _manifest([_entry("t", gate="support")], ["terminal"]) + digest = "ab" * 32 + run = GateBatteryRun( + manifest, + release_id="xx-test-build", + report_path=tmp_path / "terminal_gates.json", + release_candidate=True, + registry={"support": _binding("support")}, + release_evidence={"calibration_diagnostics_sha256": digest}, + ) + run.run_phase("terminal", EvidenceContext()) + report = json.loads((tmp_path / "terminal_gates.json").read_text()) + expected = {"calibration_diagnostics_sha256": digest} + assert report["release_evidence"] == expected + assert report["attestation"]["release_evidence"] == expected + signature = report["attestation"]["signature"] + report["attestation"]["signature"] = None + report["release_evidence"]["calibration_diagnostics_sha256"] = "cd" * 32 + report["attestation"]["release_evidence"] = report["release_evidence"] + tampered = hmac.new( + base64.b64decode(KEY), + json.dumps( + report, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + assert tampered != signature, "release_evidence sits outside the signature" + + def test_release_evidence_defaults_to_an_empty_mapping(self, tmp_path, signing_env): + manifest = _manifest([_entry("t", gate="support")], ["terminal"]) + run = GateBatteryRun( + manifest, + release_id="xx-test-build", + report_path=tmp_path / "terminal_gates.json", + release_candidate=True, + registry={"support": _binding("support")}, + ) + payload = run.report_payload() + assert payload["release_evidence"] == {} + assert payload["attestation"]["release_evidence"] == {} + + def test_release_evidence_refuses_non_string_entries(self, tmp_path): + manifest = _manifest([_entry("t", gate="support")], ["terminal"]) + for bad in ({"calibration_diagnostics_sha256": 7}, {"": "ab" * 32}): + with pytest.raises(ValueError, match="release_evidence"): + GateBatteryRun( + manifest, + release_id="xx-test-build", + report_path=tmp_path / "terminal_gates.json", + release_candidate=True, + registry={"support": _binding("support")}, + release_evidence=bad, + ) + + class TestBelgianCompatibility: def test_the_be_spec_runs_as_declared_with_named_gaps(self, tmp_path, monkeypatch): monkeypatch.setenv(gate_signing_key_env("be"), KEY) From 2ce3157902e1edbff735c91a04354331cc7a96e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:25:06 +0200 Subject: [PATCH 2/7] Thread one exclusion clock and a loud override through the UK bindings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exclusion receipts carry approval and expiry dates, and the release contract requires every gate in one report to evaluate them on the same date. The three exclusion-consuming bindings (degenerate surface, input mass, tail concentration) now require an exclusions_evaluated_on artifact — one date, computed once by the caller — instead of each gate defaulting its own clock across a possible midnight. The degenerate binding also gains the review-time override the driver offers (--degenerate-exclusions): a supplied artifact replaces the committed register for that run, and a new evidence hook digests whichever records actually ran into the signed report's evidence_sha256, so an overridden run self-describes. The committed register stays the policy of record; the spec pin test now covers its declared resource name. Co-Authored-By: Claude Fable 5 --- .../build/uk_runtime/battery_bindings.py | 71 +++++++++++-- .../tests/test_country_spec.py | 9 +- .../tests/test_uk_battery_bindings.py | 100 +++++++++++++++++- 3 files changed, 166 insertions(+), 14 deletions(-) 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 f91f81c89..944c94a0b 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 @@ -30,6 +30,7 @@ from collections.abc import Callable, Mapping from dataclasses import dataclass, replace +from datetime import date, datetime from typing import Any import pandas as pd @@ -67,7 +68,9 @@ UK_INPUT_MASS_EXCLUSION_REGISTER_RESOURCE, UK_INPUT_MASS_REFERENCE_EVIDENCE_SHA256, UK_QRF_TAIL_EXCLUSION_REGISTER_RESOURCE, + UKReviewedExclusion, _input_mass_reference_evidence_sha256, + coerce_reviewed_exclusions, uk_dataset_input_mass_totals, uk_input_mass_parity_gate, uk_qrf_tail_concentration_columns, @@ -230,6 +233,45 @@ def _stage_names_evidence( } +def _exclusion_clock(context: EvidenceContext) -> date: + """The one expiry clock every exclusion-consuming gate shares. + + Exclusion receipts carry approval and expiry dates, and the release + contract requires every gate in one report to evaluate them on the same + date. A per-gate default could straddle midnight, so the clock is a + required artifact and anything but a plain ``date`` is refused. + """ + + clock = context.artifacts["exclusions_evaluated_on"] + if isinstance(clock, datetime) or not isinstance(clock, date): + raise ValueError( + "exclusions_evaluated_on must be a datetime.date, got " + f"{type(clock).__name__}; expiry must be evaluated on one " + "shared clock." + ) + return clock + + +def _resolve_degenerate_exclusions( + context: EvidenceContext, +) -> tuple[Mapping[str, UKReviewedExclusion], str]: + """The exclusion records the degenerate gate runs, and their source. + + The committed register is the reviewed policy of record (#630/#610); + a supplied ``reviewed_degenerate_exclusions`` artifact is the loud + review-time override — the evidence hook digests whichever resolved, + so an overridden run self-describes in the signed report. + """ + + override = context.artifacts.get("reviewed_degenerate_exclusions") + if override is None: + return uk_default_degenerate_reviewed_exclusions(), "committed" + return ( + coerce_reviewed_exclusions(override, label="UK degenerate-surface policy"), + "override", + ) + + def _evaluate_degenerate_release_surface( context: EvidenceContext, parameters: Mapping[str, Any] ) -> GateResult: @@ -240,16 +282,27 @@ def _evaluate_degenerate_release_surface( f"uk/gates.json names exclusion register {register!r} but the " f"runtime loads {UK_DEGENERATE_EXCLUSION_REGISTER_RESOURCE!r}." ) - # The committed register is the reviewed policy of record (#630/#610): - # the battery must run the same exclusions the legacy terminal report - # resolves for None, or the two paths diverge on dormant/expired state. + resolved, _source = _resolve_degenerate_exclusions(context) return uk_degenerate_release_surface_gate( _uk_gate_surface(context.frame), - reviewed_exclusions=uk_default_degenerate_reviewed_exclusions(), + reviewed_exclusions=resolved, + now=_exclusion_clock(context), **kwargs, ) +def _degenerate_exclusions_evidence( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> object: + resolved, source = _resolve_degenerate_exclusions(context) + return { + "exclusions_register": source, + "reviewed_exclusions": { + name: record.policy_payload() for name, record in sorted(resolved.items()) + }, + } + + def _evaluate_zero_weight_strata( context: EvidenceContext, parameters: Mapping[str, Any] ) -> GateResult: @@ -366,6 +419,7 @@ def _evaluate_input_mass_parity( uk_dataset_input_mass_totals(_uk_gate_surface(context.frame)), context.artifacts["input_mass_reference"], policy=context.artifacts["input_mass_policy"], + now=_exclusion_clock(context), **kwargs, ) @@ -397,6 +451,7 @@ def _evaluate_tail_concentration( weights, policy=context.artifacts["qrf_tail_policy"], surface=surface, + now=_exclusion_clock(context), **kwargs, ) @@ -442,6 +497,8 @@ def _evaluate_tail_concentration( name="degenerate_release_surface", evaluator=_evaluate_degenerate_release_surface, parameter_keys=frozenset({"reviewed_exclusions_resource"}), + artifact_keys=frozenset({"exclusions_evaluated_on"}), + evidence=_degenerate_exclusions_evidence, ), "zero_weight_strata": UKGateBinding( name="zero_weight_strata", @@ -489,14 +546,16 @@ def _evaluate_tail_concentration( "candidate_name", } ), - artifact_keys=frozenset({"input_mass_reference", "input_mass_policy"}), + artifact_keys=frozenset( + {"input_mass_reference", "input_mass_policy", "exclusions_evaluated_on"} + ), evidence=_input_mass_reference_evidence, ), "tail_concentration": UKGateBinding( name="tail_concentration", evaluator=_evaluate_tail_concentration, parameter_keys=frozenset({"reviewed_exclusions_resource"}), - artifact_keys=frozenset({"qrf_tail_policy"}), + artifact_keys=frozenset({"qrf_tail_policy", "exclusions_evaluated_on"}), legacy_name="qrf_tail_concentration", ), } diff --git a/packages/microcosm-build/tests/test_country_spec.py b/packages/microcosm-build/tests/test_country_spec.py index 9b476cbec..7139b9761 100644 --- a/packages/microcosm-build/tests/test_country_spec.py +++ b/packages/microcosm-build/tests/test_country_spec.py @@ -339,9 +339,7 @@ def test_export_surface_registers_match_the_reviewed_constants( == terminal_gates.UK_REVIEWED_EXPORT_EXCLUSIONS ) - def test_input_mass_reference_is_a_declared_pinned_input( - self, manifest - ) -> None: + def test_input_mass_reference_is_a_declared_pinned_input(self, manifest) -> None: # The microcosm#327 rule: a parity gate's reference and exclusion # register are declared per-country inputs, never implicit code. params = {gate.id: gate.parameters for gate in manifest.gates} @@ -362,6 +360,11 @@ def test_input_mass_reference_is_a_declared_pinned_input( qrf["reviewed_exclusions_resource"] == weighted_integrity.UK_QRF_TAIL_EXCLUSION_REGISTER_RESOURCE ) + degenerate = params["uk_degenerate_release_surface"] + assert ( + degenerate["reviewed_exclusions_resource"] + == weighted_integrity.UK_DEGENERATE_EXCLUSION_REGISTER_RESOURCE + ) class TestRefusals: diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index 0e0a12a97..6e7b09a59 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -16,6 +16,7 @@ from __future__ import annotations import base64 +from datetime import date, datetime from types import SimpleNamespace from unittest.mock import patch @@ -61,6 +62,10 @@ KEY = base64.b64encode(b"\x07" * 32).decode("ascii") RELEASE_ID = "populace-uk-2023-frs-k535080" DIAGNOSTICS_SHA256 = "c" * 64 +#: The shared exclusion-expiry clock, fixed inside the committed register's +#: validity window (approved 2026-08-10, expires 2027-02-10) so the suite +#: never drifts across an expiry boundary. +CLOCK = date(2026, 9, 1) #: Neutral declared name -> the legacy result name the bindings re-mint. LEGACY_NAMES = { @@ -165,12 +170,14 @@ def _fixture_coverage_registry(): } -def _run_both(tables, *, parity=None, fit_records=None, armed=True): +def _run_both(tables, *, parity=None, fit_records=None, armed=True, clock=CLOCK): """Run the legacy battery and the declared battery over one evidence set. Both sides are built from the same tables and the same evidence objects in one place — evidence asymmetry between the sides would read as a - false differential failure. + false differential failure. That includes the exclusion-expiry clock: + the legacy aggregator threads ``now`` and the battery threads the + ``exclusions_evaluated_on`` artifact, both set to the same date here. """ person, benunit, household = tables @@ -178,8 +185,11 @@ def _run_both(tables, *, parity=None, fit_records=None, armed=True): frame = uk_national_frame( person=person, benunit=benunit, household=household, time_period="2023" ) - artifacts: dict[str, object] = {"coverage_engine": object()} - legacy_kwargs: dict[str, object] = {} + artifacts: dict[str, object] = { + "coverage_engine": object(), + "exclusions_evaluated_on": clock, + } + legacy_kwargs: dict[str, object] = {"now": clock} if fit_records is not None: artifacts["fit_weight_records"] = fit_records legacy_kwargs["fit_weight_records"] = fit_records @@ -289,7 +299,11 @@ def test_missing_evidence_names_its_keys(self, uk_gates) -> None: reasons = {o.entry.id: o.reason for o in phase.outcomes} assert reasons["uk_weights_audit"] == ("missing evidence: fit_weight_records") assert reasons["uk_input_mass_parity"] == ( - "missing evidence: frame, input_mass_policy, input_mass_reference" + "missing evidence: frame, exclusions_evaluated_on, " + "input_mass_policy, input_mass_reference" + ) + assert reasons["uk_degenerate_release_surface"] == ( + "missing evidence: frame, exclusions_evaluated_on" ) @@ -427,6 +441,82 @@ def test_absent_but_required_fit_evidence_is_the_named_delta(self) -> None: assert audit.reason == "missing evidence: fit_weight_records" +class TestExclusionDiscipline: + """One expiry clock, a committed register of record, a loud override.""" + + EXCLUSION_GATES = ( + "uk_degenerate_release_surface", + "uk_input_mass_parity", + "uk_qrf_tail_concentration", + ) + + def test_every_exclusion_gate_shares_the_injected_clock(self) -> None: + _legacy, battery = _run_both( + _tables(), + parity=_parity(), + fit_records=(FitWeightRecord("spi_qrf", "importance"),), + ) + by_id = {o.entry.id: o for o in battery.outcomes} + stamps = { + entry_id: by_id[entry_id].result.details["exclusions_evaluated_on"] + for entry_id in self.EXCLUSION_GATES + } + assert set(stamps.values()) == {CLOCK.isoformat()}, stamps + + def test_an_expired_register_behaves_identically_on_both_sides(self) -> None: + # Past the committed register's expiry the exclusion is out of + # force on both paths; whatever the verdict, it must be the same + # verdict — the differential contract holds at every clock value. + legacy, battery = _run_both( + _tables(), + parity=_parity(), + fit_records=(FitWeightRecord("spi_qrf", "importance"),), + clock=date(2027, 3, 1), + ) + _assert_identical_verdicts(legacy, battery) + + def test_review_override_is_loud_in_the_evidence_payload(self) -> None: + binding = UK_GATE_REGISTRY["degenerate_release_surface"] + committed = binding.evidence_payload( + EvidenceContext(artifacts={"exclusions_evaluated_on": CLOCK}), {} + ) + assert committed["exclusions_register"] == "committed" + assert "household.source_year" in committed["reviewed_exclusions"] + + overridden = binding.evidence_payload( + EvidenceContext( + artifacts={ + "exclusions_evaluated_on": CLOCK, + "reviewed_degenerate_exclusions": {}, + } + ), + {}, + ) + assert overridden["exclusions_register"] == "override" + assert overridden["reviewed_exclusions"] == {} + assert overridden != committed, "an override must move the evidence digest" + + def test_a_datetime_clock_is_refused(self, uk_gates) -> None: + person, benunit, household = _tables() + frame = uk_national_frame( + person=person, benunit=benunit, household=household, time_period="2023" + ) + entry = {e.id: e for e in uk_gates.gates}["uk_degenerate_release_surface"] + binding = UK_GATE_REGISTRY["degenerate_release_surface"] + result = _evaluate_gate( + "degenerate_release_surface", + lambda: binding.evaluate( + EvidenceContext( + frame=frame, + artifacts={"exclusions_evaluated_on": datetime(2026, 9, 1, 12, 0)}, + ), + entry.parameters, + ), + ) + assert result.passed is False + assert "shared clock" in result.details["evaluation_error"]["message"] + + class _TerminalCoverageEngine: """Minimal engine surface the coverage gate consults.""" From a22941f93a0ec6ee18b1f30d1692c0ef59d9c6f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:46:04 +0200 Subject: [PATCH 3/7] Swap the UK national build onto the shared gate battery executor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consumer flip the tooling was built for: build_uk_national_dataset constructs one GateBatteryRun over the declared uk/gates.json spec and runs both phases under BLOCKS_ARTIFACT — preflight before the frame loads (the two raising assertions become declared, persisted verdicts), terminal after the last stage and immediately before the staging writer. Every declared entry now appears in the schema-4 report at the existing diagnostic path; evidence the build cannot supply is a named evidence_absent gap that blocks only release candidates, which is the chartered semantic change: omission stops vanishing. release_candidate (default False, refused on a sampled rung — the #627 non-publishability coupling), a shared now clock, and a gate_registry test seam replace the module-attribute monkeypatch seams. A full-scale build still refuses to stage unsigned (report on disk first); a rung may proceed with an honest shippable: false. The schema-1 input-coverage alias keeps its byte-compatible last-write order through a try/finally around enforce. Deleted: _UKGateEvidence/_uk_gate_evidence (battery_bindings' _UKGateSurface is the one surviving copy) and terminal_gates' verbatim _evaluate_gate (the shared executor's copy is now imported by the retained uk_terminal_gate_report oracle). GateBatteryRun gains a public phase_report accessor so the build result carries its phase reports without reaching into private state. Co-Authored-By: Claude Fable 5 --- .../src/microcosm/build/gate_battery.py | 21 +- .../build/uk_runtime/national_build.py | 250 +++-- .../build/uk_runtime/terminal_gates.py | 40 +- .../tests/test_uk_battery_bindings.py | 30 +- .../tests/test_uk_national_build.py | 876 +++++++++--------- 5 files changed, 627 insertions(+), 590 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/gate_battery.py b/packages/microcosm-build/src/microcosm/build/gate_battery.py index 5a0ef54da..26ad06834 100644 --- a/packages/microcosm-build/src/microcosm/build/gate_battery.py +++ b/packages/microcosm-build/src/microcosm/build/gate_battery.py @@ -520,11 +520,12 @@ def failures(self) -> tuple[str, ...]: def _evaluate_gate(name: str, evaluator: Callable[[], GateResult]) -> GateResult: """Run one evaluator, failing closed on any misbehaviour. - Lifted verbatim from the UK terminal battery: a raising evaluator - becomes a failed result (the batch must keep evaluating — a crash that - masked the remaining gates would hide exactly the failures the battery - exists to surface), and a result under the wrong name fails rather than - letting one gate impersonate another. + The one fail-closed wrapper for every battery, shared with the legacy UK + terminal report: a raising evaluator becomes a failed result (the batch + must keep evaluating — a crash that masked the remaining gates would + hide exactly the failures the battery exists to surface), and a result + under the wrong name fails rather than letting one gate impersonate + another. """ try: @@ -804,6 +805,16 @@ def blocked_at_phase(self) -> str | None: def phases_evaluated(self) -> tuple[str, ...]: return tuple(self._phase_reports) + def phase_report(self, phase: str) -> GatePhaseReport: + """The evaluated report for ``phase``; refuses a phase that has not run.""" + + try: + return self._phase_reports[phase] + except KeyError: + raise ValueError( + f"phase {phase!r} has not run; evaluated: {list(self._phase_reports)}." + ) from None + def _next_phase(self) -> str | None: for phase in self._gates.phases: if phase not in self._phase_reports: 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 594cfe7ad..682a3ff74 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 @@ -1,4 +1,4 @@ -"""National UK build orchestration with batched terminal release gates. +"""National UK build orchestration over the shared gate battery. UK source stages run ``Frame -> Frame`` on the national carrier assembled by :mod:`microcosm.build.uk_runtime.national_frame`; the staging H5 persists the @@ -6,6 +6,14 @@ ``household_weight`` as a real export column materialized from the frame's typed weights. The local-geography clone remains a separate downstream build product with its own carrier. + +Gates run through :class:`microcosm.build.gate_battery.GateBatteryRun` over +the declared ``uk/gates.json`` spec: the preflight phase before the frame +loads, the terminal phase after the last stage and immediately before the +staging writer. Every declared entry appears in the persisted schema-4 +report — evidence the build cannot supply is a named ``evidence_absent`` +gap, blocking release candidates only — and the report is on disk before +any blocking decision raises. """ from __future__ import annotations @@ -14,18 +22,27 @@ import uuid from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass +from datetime import date from pathlib import Path from typing import Any import pandas as pd import microcosm.build.uk_runtime.national_frame as _national_frame -import microcosm.build.uk_runtime.release_input_coverage as _release_input_coverage +from microcosm.build.country_spec import load_country_spec from microcosm.build.frame_sampling import ( validate_sample_fraction, validate_sample_seed, ) -from microcosm.build.gates import GateReport, GateResult +from microcosm.build.gate_battery import ( + BlockingMode, + EvidenceContext, + GateBatteryRun, + GateBinding, + GatePhaseReport, +) +from microcosm.build.gates import GateResult +from microcosm.build.uk_runtime.battery_bindings import UK_GATE_REGISTRY from microcosm.build.uk_runtime.national_frame import ( UKStagingProvenance, uk_household_weight_kind, @@ -39,8 +56,6 @@ ) from microcosm.build.uk_runtime.release_input_coverage import ( PolicyEngineUKCoverageEngine, - assert_uk_release_input_coverage_build_stages, - assert_uk_release_input_coverage_manifest_current, ) from microcosm.build.uk_runtime.terminal_gates import ( UKInputMassParityPolicy, @@ -48,16 +63,10 @@ UKQRFTailConcentrationPolicy, UKReleaseParityEvidence, UKReviewedExclusion, - uk_terminal_gate_report, - write_uk_terminal_gate_report, ) +from microcosm.build.uk_runtime.weighted_integrity import exclusion_evaluation_date from microcosm.frame import Frame, MassChangeRecord, WeightKind, engine_tables -# Retained as the existing library-test monkeypatch seam. Production terminal -# evaluation resolves the same function inside terminal_gates so its policy -# attestation can identify the builtin evaluator. -uk_release_input_coverage_gate = _release_input_coverage.uk_release_input_coverage_gate - __all__ = [ "UKNationalBuildResult", "UKNationalStage", @@ -108,40 +117,6 @@ def run(self, frame: Frame) -> Frame: return result -@dataclass(frozen=True) -class _UKGateEvidence: - """The duck-attr evidence surface the UK gate battery consumes today. - - Exactly the shadow carrier's read surface — the three entity tables plus - the weight-kind, period, and mass-log metadata — materialized from the - frame. The gate modules stay deliberately duck-typed (#611 owns their - Frame typing); until that lands, this adapter is the one place the legacy - evidence shape survives, so a gate that reads ``household_weight_kind`` - or ``time_period`` sees the frame's real values rather than a fallback. - """ - - person: pd.DataFrame - benunit: pd.DataFrame - household: pd.DataFrame - time_period: str - household_weight_kind: WeightKind - mass_log: tuple[MassChangeRecord, ...] - - -def _uk_gate_evidence(frame: Frame) -> _UKGateEvidence: - """Materialize the gate battery's evidence surface from the frame.""" - - tables = engine_tables(frame) - return _UKGateEvidence( - person=tables["person"], - benunit=tables["benunit"], - household=tables["household"], - time_period=uk_time_period(frame), - household_weight_kind=uk_household_weight_kind(frame), - mass_log=frame.mass_log, - ) - - @dataclass(frozen=True) class UKNationalBuildResult: """A gated national staging artifact and its execution evidence.""" @@ -151,20 +126,26 @@ class UKNationalBuildResult: input_h5: Path staging_h5: Path stage_names: tuple[str, ...] - terminal_gates: GateReport + #: The in-memory phase reports, declared order (preflight, terminal). + phase_reports: tuple[GatePhaseReport, ...] + #: The exact schema-4 payload persisted at ``terminal_gate_path``. + gate_report: Mapping[str, object] terminal_gate_path: Path #: The #627 rung receipt; ``None`` on a full-scale (fraction 1.0) build. sampling_receipt: Mapping[str, object] | None = None @property def input_coverage(self) -> GateResult: - """Backward-compatible projection of the consolidated gate report.""" + """Backward-compatible projection of the coverage gate's verdict.""" - return next( - result - for result in self.terminal_gates.results - if result.name == "uk_release_input_coverage" - ) + for report in self.phase_reports: + for outcome in report.outcomes: + if ( + outcome.entry.id == "uk_release_input_coverage" + and outcome.result is not None + ): + return outcome.result + raise LookupError("uk_release_input_coverage did not evaluate in this build.") @property def input_coverage_path(self) -> Path: @@ -337,6 +318,9 @@ def build_uk_national_dataset( run_config: Mapping[str, object] | None = None, sample_fraction: float = 1.0, sample_seed: int = UK_SAMPLE_SEED_DEFAULT, + release_candidate: bool = False, + now: date | None = None, + gate_registry: Mapping[str, GateBinding] | None = None, ) -> UKNationalBuildResult: """Run ordered national stages, hard-gate the result, and stage an H5. @@ -359,6 +343,14 @@ def build_uk_national_dataset( sampled run must carry the fraction and seed inside ``run_config`` (the driver does); otherwise two rungs pointed at one checkpoint directory would silently resume across each other. + + ``release_candidate`` is the battery's second blocking axis: a candidate + build treats every ``evidence_absent`` gap as blocking, a dev build + records the gap and continues. A sampled rung is structurally + non-releasable, so requesting both is refused. ``now`` is the shared + exclusion-expiry clock (default: today, UTC), threaded to every + exclusion-consuming gate so one report carries one evaluation date. + ``gate_registry`` overrides the binding registry (tests only). """ requested_input_path = Path(input_h5).expanduser() @@ -392,20 +384,8 @@ def build_uk_national_dataset( materialized_stages = tuple(stages) _validate_stages(materialized_stages) - staging_path.unlink(missing_ok=True) - diagnostic_path.unlink(missing_ok=True) - - engine = ( - coverage_engine - if coverage_engine is not None - else PolicyEngineUKCoverageEngine() - ) - # Mirrors the US cheap preflight: graph or reference drift aborts before - # source stages and, once added, before national target-registry compilation. - assert_uk_release_input_coverage_manifest_current(engine=engine) - assert_uk_release_input_coverage_build_stages( - tuple(stage.name for stage in materialized_stages) - ) + # Configuration refusals precede the battery and the sidecar unlinks: a + # misconfigured run must not delete a previous report or write a new one. if checkpoint_dir is not None and run_config is None: raise ValueError( "a checkpointed UK national build requires run_config: the " @@ -423,6 +403,47 @@ def build_uk_national_dataset( "run_config: two rungs pointed at one checkpoint directory must " "refuse, never cross-resume." ) + if release_candidate and sample_fraction != 1.0: + raise ValueError( + "a sampled rung build is structurally non-releasable (#627); " + "release_candidate requires sample_fraction == 1.0." + ) + if (input_mass_reference is None) != (input_mass_policy is None): + raise ValueError( + "input_mass_parity arms with a frozen reference and reviewed " + "thresholds together; supply both or neither." + ) + staging_path.unlink(missing_ok=True) + diagnostic_path.unlink(missing_ok=True) + + engine = ( + coverage_engine + if coverage_engine is not None + else PolicyEngineUKCoverageEngine() + ) + evaluation_date = exclusion_evaluation_date(now) + battery = GateBatteryRun( + load_country_spec("uk").gates, + release_id=release_id, + report_path=diagnostic_path, + release_candidate=release_candidate, + registry=UK_GATE_REGISTRY if gate_registry is None else gate_registry, + release_evidence={ + "calibration_diagnostics_sha256": calibration_diagnostics_sha256 + }, + ) + # Mirrors the US cheap preflight: graph or reference drift blocks before + # source stages — now with the refusal persisted as a schema-4 report. + battery.run_phase( + "preflight", + EvidenceContext( + artifacts={ + "coverage_engine": engine, + "build_stage_names": tuple(stage.name for stage in materialized_stages), + } + ), + ) + battery.enforce("preflight", mode=BlockingMode.BLOCKS_ARTIFACT) frame, provenance = load_uk_national_frame(requested_input_path) sampling_receipt: Mapping[str, object] | None = None if sample_fraction != 1.0: @@ -453,35 +474,58 @@ def build_uk_national_dataset( run_config=run_config, ) - # Mirrors the US final-export placement: evaluate every evidenced gate in + # Mirrors the US final-export placement: evaluate every declared gate in # one batch after all stages and immediately before the staging writer. - fit_weight_records, require_fit_weight_records = _stage_fit_weight_records( - materialized_stages + artifacts: dict[str, object] = { + "coverage_engine": engine, + "exclusions_evaluated_on": evaluation_date, + } + fit_weight_records = _stage_fit_weight_records(materialized_stages) + if fit_weight_records is not None: + artifacts["fit_weight_records"] = fit_weight_records + if parity_evidence is not None: + artifacts["parity_evidence"] = parity_evidence + if input_mass_reference is not None: + artifacts["input_mass_reference"] = input_mass_reference + artifacts["input_mass_policy"] = input_mass_policy + if qrf_tail_policy is not None: + artifacts["qrf_tail_policy"] = qrf_tail_policy + if reviewed_degenerate_exclusions is not None: + artifacts["reviewed_degenerate_exclusions"] = reviewed_degenerate_exclusions + terminal = battery.run_phase( + "terminal", EvidenceContext(frame=frame, artifacts=artifacts) ) - terminal_gates = uk_terminal_gate_report( - _uk_gate_evidence(frame), - engine, - release_id=release_id, - calibration_diagnostics_sha256=calibration_diagnostics_sha256, - fit_weight_records=fit_weight_records, - require_fit_weight_records=require_fit_weight_records, - parity_evidence=parity_evidence, - input_mass_reference=input_mass_reference, - input_mass_policy=input_mass_policy, - qrf_tail_policy=qrf_tail_policy, - reviewed_degenerate_exclusions=reviewed_degenerate_exclusions, + coverage_outcome = next( + outcome + for outcome in terminal.outcomes + if outcome.entry.id == "uk_release_input_coverage" ) - write_uk_terminal_gate_report(terminal_gates, diagnostic_path) - if legacy_input_coverage_output: - input_coverage = next( - gate - for gate in terminal_gates.results - if gate.name == "uk_release_input_coverage" + if legacy_input_coverage_output and coverage_outcome.result is None: + raise RuntimeError( + "uk_release_input_coverage did not evaluate; the schema-1 " + "compatibility alias has no verdict to serialize." ) - _write_input_coverage_diagnostic(diagnostic_path, input_coverage) - if not terminal_gates.passed: + try: + battery.enforce("terminal", mode=BlockingMode.BLOCKS_ARTIFACT) + finally: + # The alias consumer reads the schema-1 shape at this exact path, in + # the blocked case too — same last-write order as the legacy flow. + if legacy_input_coverage_output and coverage_outcome.result is not None: + _write_input_coverage_diagnostic(diagnostic_path, coverage_outcome.result) + gate_report = battery.report_payload() + attestation = gate_report["attestation"] + signing_error = ( + attestation.get("signing_error") if isinstance(attestation, Mapping) else None + ) + if signing_error is not None and sample_fraction == 1.0: + # A rung build may proceed unsigned (its report honestly says + # shippable: false, and a rung is structurally non-releasable); a + # full-scale build keeps the legacy guarantee — no staging artifact + # without an attested report. The unsigned report is already on disk. raise RuntimeError( - "Release gates failed: " + "; ".join(terminal_gates.failures) + "UK terminal gate report is unsigned and this is a full-scale " + f"build; refusing to stage. {signing_error} The unsigned report " + f"was written to {diagnostic_path}." ) write_uk_national_frame(frame, staging_path) @@ -491,7 +535,10 @@ def build_uk_national_dataset( input_h5=input_path, staging_h5=staging_path, stage_names=tuple(stage.name for stage in materialized_stages), - terminal_gates=terminal_gates, + phase_reports=tuple( + battery.phase_report(phase) for phase in battery.phases_evaluated + ), + gate_report=gate_report, terminal_gate_path=diagnostic_path, sampling_receipt=sampling_receipt, ) @@ -577,20 +624,27 @@ def _validate_stages(stages: tuple[UKNationalStage, ...]) -> None: def _stage_fit_weight_records( stages: tuple[UKNationalStage, ...], -) -> tuple[tuple[object, ...] | None, bool]: - """Return real fit evidence, requiring it only when HMRC executed.""" +) -> tuple[object, ...] | None: + """The weights-audit evidence artifact: present iff the HMRC stage is. + + ``None`` (no HMRC stage) leaves the artifact unsupplied, so the audit is + a named ``evidence_absent`` gap. A present stage always supplies the + artifact — records that are missing, unreadable, or empty coerce to + ``()``, which the UK audit binding fails: an absent audit is not a + passing audit. + """ hmrc_stage = next( (stage for stage in stages if stage.name == "hmrc_spi_income"), None, ) if hmrc_stage is None: - return (None, False) + return None try: records = getattr(hmrc_stage.transform, "fit_weight_records", None) - return (() if records is None else tuple(records), True) - except Exception: # noqa: BLE001 - the terminal report must name the failure - return ((), True) + return () if records is None else tuple(records) + except Exception: # noqa: BLE001 - the weights audit must name the failure + return () def _write_input_coverage_diagnostic(path: Path, gate: GateResult) -> None: diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/terminal_gates.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/terminal_gates.py index a5fff8dce..5de99d0a4 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/terminal_gates.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/terminal_gates.py @@ -30,6 +30,7 @@ import numpy as np import pandas as pd +from microcosm.build.gate_battery import _evaluate_gate from microcosm.build.gates import ( FitWeightRecord, GateReport, @@ -1378,45 +1379,6 @@ def _missing_fit_weight_evidence_gate() -> GateResult: ) -def _evaluate_gate(name: str, evaluator: Callable[[], GateResult]) -> GateResult: - try: - result = evaluator() - except Exception as exc: # noqa: BLE001 - terminal batch must keep evaluating - return GateResult( - name=name, - passed=False, - failures=( - f"Gate evaluation failed closed with {type(exc).__name__}: {exc}", - ), - details={ - "evaluation_error": { - "type": type(exc).__name__, - "message": str(exc), - } - }, - ) - if not isinstance(result, GateResult): - return GateResult( - name=name, - passed=False, - failures=( - "Gate evaluation failed closed because the evaluator did not " - "return GateResult.", - ), - details={"returned_type": type(result).__name__}, - ) - if result.name != name: - return GateResult( - name=name, - passed=False, - failures=( - f"Gate evaluator returned name {result.name!r}, expected {name!r}.", - ), - details={"returned_gate": result.name}, - ) - return result - - def uk_terminal_gate_report( dataset: Any, coverage_engine: Any, diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index 6e7b09a59..1537565bc 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -43,8 +43,10 @@ UKGateBinding, _uk_gate_surface, ) -from microcosm.build.uk_runtime.national_build import _uk_gate_evidence -from microcosm.build.uk_runtime.national_frame import uk_national_frame +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, +) from microcosm.build.uk_runtime.release_input_coverage import ( UKReleaseInputColumn, UKReleaseInputCoverageManifest, @@ -58,6 +60,7 @@ UKReleaseParityEvidence, uk_terminal_gate_report, ) +from microcosm.frame import engine_tables KEY = base64.b64encode(b"\x07" * 32).decode("ascii") RELEASE_ID = "populace-uk-2023-frs-k535080" @@ -250,20 +253,25 @@ def _assert_identical_verdicts(legacy, battery) -> None: class TestUKSurfaceAdapter: - def test_surface_matches_the_national_build_evidence_adapter(self) -> None: + def test_surface_materializes_the_frame_not_fallbacks(self) -> None: + # The one surviving copy of the legacy duck-attr evidence surface + # (the national build's adapter consolidated into it at the + # orchestration swap). Every attr must resolve to the frame's real + # values — a gate reading household_weight_kind or time_period must + # never see a fallback. person, benunit, household = _tables() frame = uk_national_frame( person=person, benunit=benunit, household=household, time_period="2023" ) surface = _uk_gate_surface(frame) - legacy = _uk_gate_evidence(frame) - - pd.testing.assert_frame_equal(surface.person, legacy.person) - pd.testing.assert_frame_equal(surface.benunit, legacy.benunit) - pd.testing.assert_frame_equal(surface.household, legacy.household) - assert surface.time_period == legacy.time_period - assert surface.household_weight_kind == legacy.household_weight_kind - assert surface.mass_log == legacy.mass_log + tables = engine_tables(frame) + + pd.testing.assert_frame_equal(surface.person, tables["person"]) + pd.testing.assert_frame_equal(surface.benunit, tables["benunit"]) + pd.testing.assert_frame_equal(surface.household, tables["household"]) + assert surface.time_period == "2023" + assert surface.household_weight_kind is uk_household_weight_kind(frame) + assert surface.mass_log == frame.mass_log class TestUKCompatibility: diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index 61daa1aef..305bc3054 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -1,12 +1,21 @@ from __future__ import annotations import json +from datetime import date from pathlib import Path import pandas as pd import pytest -from microcosm.build.gates import FitWeightRecord, GateReport, GateResult +from microcosm.build.gate_battery import ( + GateBatteryBlockedError, + gate_signing_key_env, +) +from microcosm.build.gates import FitWeightRecord, GateResult +from microcosm.build.uk_runtime.battery_bindings import ( + UKGateBinding, + _uk_gate_surface, +) from microcosm.build.uk_runtime.national_build import ( UKNationalStage, build_uk_national_dataset, @@ -17,24 +26,75 @@ uk_national_frame, uk_time_period, ) -from microcosm.build.uk_runtime.terminal_gates import ( - UK_TERMINAL_GATE_SIGNING_KEY_ENV, - UKReleaseParityEvidence, -) -from microcosm.build.uk_runtime.terminal_gates import ( - uk_terminal_gate_report as real_uk_terminal_gate_report, -) -from microcosm.build.uk_runtime.terminal_gates import ( - write_uk_terminal_gate_report as real_write_uk_terminal_gate_report, +from microcosm.build.uk_runtime.release_input_coverage import ( + uk_release_input_coverage_gate, ) +from microcosm.build.uk_runtime.terminal_gates import UKReleaseParityEvidence from microcosm.frame import Frame, MassChangeRecord, WeightKind TEST_UK_RELEASE_ID = "populace-uk-2023-frs-k535080" TEST_UK_CALIBRATION_DIAGNOSTICS_SHA256 = "c" * 64 TEST_UK_TERMINAL_GATE_SIGNING_KEY = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=" +#: A fixed exclusion clock inside the committed register's validity window +#: keeps toy builds deterministic across the suite's lifetime. +TEST_UK_EXCLUSION_CLOCK = date(2026, 9, 1) + + +def _toy_coverage_evaluator(context, parameters): + """Pass the manifest preflight, run the real coverage gate at terminal. + + The manifest-currency check needs the shipped coverage machinery these + toy builds do not carry; the terminal verdict stays the real gate over + the real frame surface, as the legacy seam fixture ran it. + """ + + if parameters.get("check") == "manifest_current": + return GateResult( + name="release_input_coverage", + passed=True, + details={"check": "manifest_current", "toy_preflight": True}, + ) + return uk_release_input_coverage_gate( + _uk_gate_surface(context.frame), context.artifacts["coverage_engine"] + ) + + +def _toy_gate_registry() -> dict[str, UKGateBinding]: + """The seam-test registry: real terminal coverage, pass-through roster. + + These seam tests use toy stages, so the family-roster gate and the + manifest preflight are pass-throughs (both have their own tests) and + every gate without a binding is a named ``evidence_absent`` gap — + non-blocking off the release-candidate posture, exactly the legacy + fixture's effect of reporting only the coverage verdict. + """ + + return { + "release_input_coverage": UKGateBinding( + name="release_input_coverage", + evaluator=_toy_coverage_evaluator, + parameter_keys=frozenset({"check"}), + artifact_keys=frozenset({"coverage_engine"}), + frame_predicate=( + lambda parameters: parameters.get("check") != "manifest_current" + ), + legacy_name="uk_release_input_coverage", + ), + "source_coverage": UKGateBinding( + name="source_coverage", + evaluator=lambda context, parameters: GateResult( + name="source_coverage", + passed=True, + details={"toy_stage_roster": True}, + ), + needs_frame=False, + ), + } def _run_national_build(**kwargs): + kwargs.setdefault("gate_registry", _toy_gate_registry()) + kwargs.setdefault("now", TEST_UK_EXCLUSION_CLOCK) return build_uk_national_dataset( release_id=TEST_UK_RELEASE_ID, calibration_diagnostics_sha256=TEST_UK_CALIBRATION_DIAGNOSTICS_SHA256, @@ -58,50 +118,11 @@ def _replace_person(frame: Frame, person: pd.DataFrame) -> Frame: @pytest.fixture(autouse=True) def _trusted_terminal_gate_signing_key(monkeypatch) -> None: monkeypatch.setenv( - UK_TERMINAL_GATE_SIGNING_KEY_ENV, + gate_signing_key_env("uk"), TEST_UK_TERMINAL_GATE_SIGNING_KEY, ) -@pytest.fixture(autouse=True) -def _isolate_generic_seam_from_shipped_family_contract(monkeypatch) -> None: - """These seam tests use toy stages; family enforcement has its own tests.""" - - from microcosm.build.uk_runtime import national_build - - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_build_stages", - lambda _stage_names: None, - ) - monkeypatch.setattr( - national_build, - "uk_terminal_gate_report", - lambda dataset, engine, **_kwargs: GateReport( - (national_build.uk_release_input_coverage_gate(dataset, engine),) - ), - ) - - def write_generic_seam_report(report, path): - output = Path(path) - output.parent.mkdir(parents=True, exist_ok=True) - output.write_text( - json.dumps( - {"schema_version": 2, "enforced": True, **report.to_manifest()}, - indent=2, - sort_keys=True, - ) - + "\n" - ) - return output - - monkeypatch.setattr( - national_build, - "write_uk_terminal_gate_report", - write_generic_seam_report, - ) - - def _write_toy_h5(path: Path, *, employment_income: float = 0.0) -> None: with pd.HDFStore(path) as store: store.put( @@ -210,6 +231,32 @@ def _failing_gate() -> GateResult: ) +def _registry_with_coverage(gate_result_factory) -> dict[str, UKGateBinding]: + """The toy registry with the terminal coverage verdict stubbed.""" + + def evaluator(context, parameters): + if parameters.get("check") == "manifest_current": + return GateResult( + name="release_input_coverage", + passed=True, + details={"check": "manifest_current", "toy_preflight": True}, + ) + return gate_result_factory() + + registry = _toy_gate_registry() + registry["release_input_coverage"] = UKGateBinding( + name="release_input_coverage", + evaluator=evaluator, + parameter_keys=frozenset({"check"}), + artifact_keys=frozenset({"coverage_engine"}), + frame_predicate=( + lambda parameters: parameters.get("check") != "manifest_current" + ), + legacy_name="uk_release_input_coverage", + ) + return registry + + def test_driver_validates_the_uk_residue_after_each_stage( monkeypatch, tmp_path ) -> None: @@ -225,16 +272,10 @@ def test_driver_validates_the_uk_residue_after_each_stage( """ pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build from microcosm.frame import CONSERVE_MASS, Weights input_h5 = tmp_path / "base.h5" _write_two_row_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) def redistribute_without_refreshing_column(frame: Frame) -> Frame: weights = frame.weights_for("household") @@ -255,55 +296,6 @@ def redistribute_without_refreshing_column(frame: Frame) -> Frame: ) -def test_gate_evidence_reproduces_the_legacy_attr_surface() -> None: - """_uk_gate_evidence exposes exactly what the duck-typed gates read. - - The gate modules stay deliberately duck-typed until #611 types them on - Frame; the evidence adapter must therefore carry the metadata attrs - (kind, period, mass log) the coverage gate's hmrc family getattr-reads, - with the typed weights materialized authoritatively into the tables. - """ - - from microcosm.build.uk_runtime import national_build - - mass_log = ( - MassChangeRecord( - entity="household", - old_total=2.0, - new_total=2.0, - declared_factor=1.0, - reason="Toy reviewed record.", - ), - ) - frame = uk_national_frame( - person=pd.DataFrame( - { - "person_id": [10], - "person_benunit_id": [100], - "person_household_id": [1], - } - ), - benunit=pd.DataFrame({"benunit_id": [100]}), - household=pd.DataFrame({"household_id": [1], "household_weight": [2.0]}), - time_period="2023", - weight_kind=WeightKind.IMPORTANCE, - mass_log=mass_log, - ) - - evidence = national_build._uk_gate_evidence(frame) - - assert evidence.household_weight_kind is WeightKind.IMPORTANCE - assert evidence.time_period == "2023" - assert evidence.mass_log == mass_log - assert evidence.household["household_weight"].tolist() == [2.0] - pd.testing.assert_frame_equal(evidence.person, frame.person) - # The same getattr surface the gates use resolves to real values, never - # the silent fallbacks a plain table mapping produced. - assert getattr(evidence, "household_weight_kind", None) is not None - assert str(getattr(evidence, "time_period", "")) == "2023" - assert tuple(getattr(evidence, "mass_log", ())) == mass_log - - class _RecordedFitStage: fit_weight_records = ( FitWeightRecord("uk_spi_2022_23_income", "design"), @@ -332,37 +324,44 @@ def stage_transform(frame: Frame) -> Frame: person["employment_income"] = 50_000.0 return _replace_person(frame, person) - def assert_current(**_kwargs) -> None: - events.append("manifest_preflight") - - def coverage_gate(evidence, _engine): + def recording_coverage(context, parameters): + if parameters.get("check") == "manifest_current": + events.append("manifest_preflight") + return GateResult( + name="release_input_coverage", + passed=True, + details={"check": "manifest_current"}, + ) events.append("final_coverage_gate") - assert evidence.person["employment_income"].tolist() == [50_000.0] - # The gate battery's evidence carries the frame's metadata surface — - # the coverage gate's hmrc family reads these attrs, and a bare table + surface = _uk_gate_surface(context.frame) + assert surface.person["employment_income"].tolist() == [50_000.0] + # The battery's evidence surface carries the frame's metadata — the + # coverage gate's hmrc family reads these attrs, and a bare table # mapping silently fails them to ''/() (caught by the first # credentialed acceptance build, not by CI's toy stages). - assert evidence.time_period == "2023" - assert evidence.household_weight_kind is WeightKind.DESIGN - assert evidence.mass_log == () + assert surface.time_period == "2023" + assert surface.household_weight_kind is WeightKind.DESIGN + assert surface.mass_log == () return _passing_gate() + registry = _toy_gate_registry() + registry["release_input_coverage"] = UKGateBinding( + name="release_input_coverage", + evaluator=recording_coverage, + parameter_keys=frozenset({"check"}), + artifact_keys=frozenset({"coverage_engine"}), + frame_predicate=( + lambda parameters: parameters.get("check") != "manifest_current" + ), + legacy_name="uk_release_input_coverage", + ) + real_writer = national_build.write_uk_national_frame def recording_writer(frame, path): events.append("staging_write") return real_writer(frame, path) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - assert_current, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - coverage_gate, - ) monkeypatch.setattr( national_build, "write_uk_national_frame", @@ -375,6 +374,7 @@ def recording_writer(frame, path): stages=(UKNationalStage("income", stage_transform),), coverage_engine=object(), input_coverage_path=coverage_json, + gate_registry=registry, ) assert events == [ @@ -386,7 +386,13 @@ def recording_writer(frame, path): assert result.sampling_receipt is None assert result.stage_names == ("income",) assert result.input_coverage.passed is True - assert result.terminal_gates.passed is True + assert result.gate_report["blocked_at_phase"] is None + assert result.gate_report["phases_evaluated"] == ["preflight", "terminal"] + gates = result.gate_report["gates"] + assert gates["uk_release_input_coverage"]["status"] == "passed" + assert result.gate_report["release_evidence"] == { + "calibration_diagnostics_sha256": TEST_UK_CALIBRATION_DIAGNOSTICS_SHA256 + } assert result.terminal_gate_path == coverage_json.resolve() assert result.input_coverage_path == result.terminal_gate_path assert result.provenance.source_h5 == input_h5.resolve() @@ -458,11 +464,8 @@ def _write_clone_family_h5(path: Path) -> None: ) -def test_national_build_samples_the_loaded_frame_before_stages( - monkeypatch, tmp_path -) -> None: +def test_national_build_samples_the_loaded_frame_before_stages(tmp_path) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" staging_h5 = tmp_path / "staging.h5" @@ -474,17 +477,6 @@ def stage_transform(frame: Frame) -> Frame: stage_household_counts.append(len(frame.table("household"))) return frame - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _evidence, _engine: _passing_gate(), - ) - result = _run_national_build( input_h5=input_h5, staging_h5=staging_h5, @@ -493,6 +485,7 @@ def stage_transform(frame: Frame) -> Frame: input_coverage_path=coverage_json, sample_fraction=0.5, sample_seed=3, + gate_registry=_registry_with_coverage(_passing_gate), ) receipt = result.sampling_receipt @@ -504,35 +497,24 @@ def stage_transform(frame: Frame) -> Frame: # Renormalization: the staged artifact carries the full input mass. staged, _staged_provenance = load_uk_national_frame(staging_h5) assert float(staged.weights_for("household").total) == pytest.approx(8 * 2.0) - assert result.terminal_gates.passed is True + assert result.gate_report["blocked_at_phase"] is None def test_legacy_input_coverage_alias_is_byte_compatible_with_origin_main( - monkeypatch, tmp_path, ) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" staging_h5 = tmp_path / "staging.h5" legacy_json = tmp_path / "input_coverage.json" _write_toy_h5(input_h5, employment_income=40_000.0) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) _run_national_build( input_h5=input_h5, staging_h5=staging_h5, coverage_engine=object(), input_coverage_path=legacy_json, + gate_registry=_registry_with_coverage(_passing_gate), ) # Pinned from origin/main's schema-1 serializer for this exact GateResult. @@ -545,88 +527,83 @@ def test_legacy_input_coverage_alias_is_byte_compatible_with_origin_main( assert legacy_json.read_bytes() == expected -def test_legacy_input_coverage_alias_fails_closed_without_signing_key( - monkeypatch, - tmp_path, -) -> None: - """The compatibility output cannot bypass the signed terminal writer.""" +def test_full_scale_build_refuses_to_stage_unsigned(monkeypatch, tmp_path) -> None: + """No full-scale staging artifact without an attested report. + + The battery core records a missing key as ``signing_error`` and carries + on; the national build restores the legacy guarantee for full-scale + builds — the unsigned report is on disk, the H5 is not. + """ pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build, terminal_gates input_h5 = tmp_path / "base.h5" staging_h5 = tmp_path / "staging.h5" - legacy_json = tmp_path / "input_coverage.json" + terminal_json = tmp_path / "terminal_gates.json" _write_two_row_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) - monkeypatch.setattr( - terminal_gates, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) - monkeypatch.setattr( - national_build, - "uk_terminal_gate_report", - real_uk_terminal_gate_report, - ) - monkeypatch.setattr( - national_build, - "write_uk_terminal_gate_report", - real_write_uk_terminal_gate_report, - ) - monkeypatch.delenv(UK_TERMINAL_GATE_SIGNING_KEY_ENV) + monkeypatch.delenv(gate_signing_key_env("uk")) - with pytest.raises(RuntimeError, match="Unsigned failed report was written"): + with pytest.raises(RuntimeError, match="unsigned and this is a full-scale"): _run_national_build( input_h5=input_h5, staging_h5=staging_h5, coverage_engine=object(), - input_coverage_path=legacy_json, + terminal_gate_path=terminal_json, + gate_registry=_registry_with_coverage(_passing_gate), ) assert not staging_h5.exists() - payload = json.loads(legacy_json.read_text(encoding="utf-8")) - assert payload["schema_version"] == 3 - assert payload["passed"] is False + payload = json.loads(terminal_json.read_text(encoding="utf-8")) + assert payload["schema_version"] == 4 + assert payload["shippable"] is False assert payload["attestation"]["signature"] is None assert payload["attestation"]["signing_key_sha256"] is None + assert "signing_error" in payload["attestation"] -def test_national_build_gate_failure_writes_diagnostic_not_h5( +def test_rung_build_proceeds_unsigned_with_an_honest_report( monkeypatch, tmp_path ) -> None: + """A rung is structurally non-releasable, so it may run without the key; + its report says so instead of pretending.""" + + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + staging_h5 = tmp_path / "staging.h5" + terminal_json = tmp_path / "terminal_gates.json" + _write_clone_family_h5(input_h5) + monkeypatch.delenv(gate_signing_key_env("uk")) + + result = _run_national_build( + input_h5=input_h5, + staging_h5=staging_h5, + coverage_engine=object(), + terminal_gate_path=terminal_json, + sample_fraction=0.5, + sample_seed=3, + gate_registry=_registry_with_coverage(_passing_gate), + ) + + assert staging_h5.exists() + assert result.gate_report["shippable"] is False + assert "signing_error" in result.gate_report["attestation"] + + +def test_national_build_gate_failure_writes_diagnostic_not_h5(tmp_path) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" staging_h5 = tmp_path / "staging.h5" coverage_json = tmp_path / "input_coverage.json" _write_toy_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _failing_gate(), - ) - with pytest.raises(RuntimeError, match="Release gates failed"): + with pytest.raises(GateBatteryBlockedError, match="Gate battery blocked"): _run_national_build( input_h5=input_h5, staging_h5=staging_h5, coverage_engine=object(), input_coverage_path=coverage_json, + gate_registry=_registry_with_coverage(_failing_gate), ) assert not staging_h5.exists() @@ -649,51 +626,56 @@ def test_default_terminal_report_write_precedes_gate_failure_raise( _write_toy_h5(input_h5) events: list[str] = [] real_loader = national_build.load_uk_national_frame - real_report_writer = national_build.write_uk_terminal_gate_report - - def preflight(**_kwargs) -> None: - events.append("preflight") - - def stage_contract(_stage_names) -> None: - events.append("stage contract") def load(path): events.append("load") return real_loader(path) - def evaluate(_dataset, _engine): + def recording_coverage(context, parameters): + if parameters.get("check") == "manifest_current": + events.append("preflight") + return GateResult( + name="release_input_coverage", + passed=True, + details={"check": "manifest_current"}, + ) events.append("evaluate") + # The preflight report is already on disk before the frame loads — + # the write-then-block ordering holds per phase, not just at the end. + assert json.loads(default_terminal_json.read_text())["phases_evaluated"] == [ + "preflight" + ] return _failing_gate() - def write_report(report, path): - events.append("write report") - assert Path(path) == default_terminal_json.resolve() - return real_report_writer(report, path) - - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - preflight, + def recording_roster(context, parameters): + events.append("stage contract") + return GateResult(name="source_coverage", passed=True, details={}) + + registry = _toy_gate_registry() + registry["release_input_coverage"] = UKGateBinding( + name="release_input_coverage", + evaluator=recording_coverage, + parameter_keys=frozenset({"check"}), + artifact_keys=frozenset({"coverage_engine"}), + frame_predicate=( + lambda parameters: parameters.get("check") != "manifest_current" + ), + legacy_name="uk_release_input_coverage", ) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_build_stages", - stage_contract, + registry["source_coverage"] = UKGateBinding( + name="source_coverage", + evaluator=recording_roster, + needs_frame=False, ) monkeypatch.setattr(national_build, "load_uk_national_frame", load) - monkeypatch.setattr(national_build, "uk_release_input_coverage_gate", evaluate) - monkeypatch.setattr( - national_build, - "write_uk_terminal_gate_report", - write_report, - ) - with pytest.raises(RuntimeError, match="Release gates failed"): + with pytest.raises(GateBatteryBlockedError, match="Gate battery blocked"): _run_national_build( input_h5=input_h5, staging_h5=staging_h5, coverage_engine=object(), terminal_gate_path=None, + gate_registry=registry, ) events.append("raise") @@ -702,77 +684,97 @@ def write_report(report, path): "stage contract", "load", "evaluate", - "write report", "raise", ] assert default_terminal_json.is_file() - assert json.loads(default_terminal_json.read_text())["passed"] is False + payload = json.loads(default_terminal_json.read_text()) + assert payload["schema_version"] == 4 + assert payload["blocked_at_phase"] == "terminal" + assert payload["gates"]["uk_release_input_coverage"]["status"] == "failed" assert not staging_h5.exists() -def test_national_build_real_terminal_batch_passes_before_staging( - monkeypatch, - tmp_path, -) -> None: - pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build, terminal_gates +def _stub_real_coverage(monkeypatch, gate_result_factory) -> None: + """Point the real registry's coverage binding at a stubbed verdict. + + The bindings resolve the manifest assert and the coverage gate as + module globals at call time, so patching them where the bindings look + them up leaves every other real binding untouched. + """ + + from microcosm.build.uk_runtime import battery_bindings - input_h5 = tmp_path / "healthy.h5" - staging_h5 = tmp_path / "staging.h5" - terminal_json = tmp_path / "terminal_gates.json" - _write_two_row_h5(input_h5) monkeypatch.setattr( - national_build, + battery_bindings, "assert_uk_release_input_coverage_manifest_current", lambda **_kwargs: None, ) monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), + battery_bindings, + "assert_uk_release_input_coverage_build_stages", + lambda _stage_names, manifest=None: None, ) monkeypatch.setattr( - terminal_gates, + battery_bindings, "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) - monkeypatch.setattr( - national_build, - "uk_terminal_gate_report", - real_uk_terminal_gate_report, + lambda _surface, _engine, manifest=None: gate_result_factory(), ) + +def test_national_build_real_terminal_batch_passes_before_staging( + monkeypatch, + tmp_path, +) -> None: + pytest.importorskip("tables") + + input_h5 = tmp_path / "healthy.h5" + staging_h5 = tmp_path / "staging.h5" + terminal_json = tmp_path / "terminal_gates.json" + _write_two_row_h5(input_h5) + _stub_real_coverage(monkeypatch, _passing_gate) + result = _run_national_build( input_h5=input_h5, staging_h5=staging_h5, coverage_engine=object(), terminal_gate_path=terminal_json, + gate_registry=None, # the real UK registry ) - assert result.terminal_gates.passed - assert [gate.name for gate in result.terminal_gates.results] == [ - "uk_release_input_coverage", - "degenerate_release_surface", - "zero_weight_strata", - "weight_ess", - "weight_ratio", - ] - assert result.input_coverage is result.terminal_gates.results[0] + assert result.input_coverage.passed is True assert result.terminal_gate_path == terminal_json.resolve() assert staging_h5.is_file() payload = json.loads(terminal_json.read_text(encoding="utf-8")) - assert payload["passed"] is True - assert set(payload["gates"]) == { - "uk_release_input_coverage", - "degenerate_release_surface", - "zero_weight_strata", - "weight_ess", - "weight_ratio", + assert payload["schema_version"] == 4 + assert payload["blocked_at_phase"] is None + statuses = {entry_id: gate["status"] for entry_id, gate in payload["gates"].items()} + assert statuses == { + "uk_release_input_coverage_manifest_current": "passed", + "uk_release_family_build_stages": "passed", + "uk_release_input_coverage": "passed", + "uk_degenerate_release_surface": "passed", + "uk_zero_weight_strata": "passed", + "uk_weight_ess": "passed", + "uk_weight_ratio": "passed", + # The legacy report omitted unevidenced gates; the battery names + # every gap — non-blocking off the release-candidate posture. + "uk_weights_audit": "evidence_absent", + "uk_export_surface": "evidence_absent", + "uk_target_surface": "evidence_absent", + "uk_target_fit": "evidence_absent", + "uk_input_mass_parity": "evidence_absent", + "uk_qrf_tail_concentration": "evidence_absent", + } + # One exclusion clock: the evaluated exclusion gate stamps the injected + # date, never a per-gate default. + degenerate = payload["gates"]["uk_degenerate_release_surface"] + assert ( + degenerate["details"]["exclusions_evaluated_on"] + == TEST_UK_EXCLUSION_CLOCK.isoformat() + ) + assert payload["release_evidence"] == { + "calibration_diagnostics_sha256": TEST_UK_CALIBRATION_DIAGNOSTICS_SHA256 } - assert "weights_audit" not in payload["gates"] - assert "export_surface" not in payload["gates"] - assert "target_surface" not in payload["gates"] - assert "target_fit" not in payload["gates"] def test_national_build_real_terminal_batch_writes_all_findings_before_raise( @@ -780,92 +782,55 @@ def test_national_build_real_terminal_batch_writes_all_findings_before_raise( tmp_path, ) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build, terminal_gates input_h5 = tmp_path / "defective.h5" staging_h5 = tmp_path / "staging.h5" terminal_json = tmp_path / "terminal_gates.json" _write_two_row_h5(input_h5, employment_income=(0.0, 0.0)) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _failing_gate(), - ) - monkeypatch.setattr( - terminal_gates, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _failing_gate(), - ) - monkeypatch.setattr( - national_build, - "uk_terminal_gate_report", - real_uk_terminal_gate_report, - ) + _stub_real_coverage(monkeypatch, _failing_gate) - with pytest.raises(RuntimeError, match="Release gates failed") as error: + with pytest.raises(GateBatteryBlockedError) as error: _run_national_build( input_h5=input_h5, staging_h5=staging_h5, stages=(UKNationalStage("hmrc_spi_income", lambda dataset: dataset),), coverage_engine=object(), terminal_gate_path=terminal_json, + gate_registry=None, # the real UK registry ) assert "[uk_release_input_coverage]" in str(error.value) - assert "[degenerate_release_surface]" in str(error.value) - assert "[weights_audit]" in str(error.value) + assert "[uk_degenerate_release_surface]" in str(error.value) + assert "[uk_weights_audit]" in str(error.value) + assert error.value.phase == "terminal" assert terminal_json.is_file() payload = json.loads(terminal_json.read_text(encoding="utf-8")) - assert payload["passed"] is False - assert payload["gates"]["uk_release_input_coverage"]["passed"] is False - assert payload["gates"]["degenerate_release_surface"]["passed"] is False - assert payload["gates"]["weights_audit"] == { - "details": {"evidence_missing": True, "fits_checked": 0}, - "failures": [ - "A production fit stage ran but emitted no FitWeightRecord evidence; " - "an absent audit is not a passing audit." - ], - "passed": False, + assert payload["blocked_at_phase"] == "terminal" + assert payload["shippable"] is False + assert payload["gates"]["uk_release_input_coverage"]["status"] == "failed" + assert payload["gates"]["uk_degenerate_release_surface"]["status"] == "failed" + weights_audit = payload["gates"]["uk_weights_audit"] + assert weights_audit["status"] == "failed" + assert weights_audit["details"] == { + "evidence_missing": True, + "fits_checked": 0, } + assert weights_audit["failures"] == [ + "A production fit stage ran but emitted no FitWeightRecord evidence; " + "an absent audit is not a passing audit." + ] assert not staging_h5.exists() -def test_national_build_includes_parity_trio_only_with_real_evidence( +def test_national_build_parity_trio_evaluates_with_evidence_absent_without( monkeypatch, tmp_path, ) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build, terminal_gates input_h5 = tmp_path / "healthy.h5" - staging_h5 = tmp_path / "staging.h5" - terminal_json = tmp_path / "terminal_gates.json" _write_two_row_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) - monkeypatch.setattr( - terminal_gates, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) - monkeypatch.setattr( - national_build, - "uk_terminal_gate_report", - real_uk_terminal_gate_report, - ) + _stub_real_coverage(monkeypatch, _passing_gate) parity = UKReleaseParityEvidence( candidate_columns=("person.employment_income",), reference_columns=("person.employment_income",), @@ -874,28 +839,38 @@ def test_national_build_includes_parity_trio_only_with_real_evidence( target_relative_errors={"population": 0.0}, ) - result = _run_national_build( + with_evidence = _run_national_build( input_h5=input_h5, - staging_h5=staging_h5, + staging_h5=tmp_path / "staging.h5", stages=(UKNationalStage("hmrc_spi_income", _RecordedFitStage()),), coverage_engine=object(), parity_evidence=parity, - terminal_gate_path=terminal_json, + terminal_gate_path=tmp_path / "terminal_gates.json", + gate_registry=None, # the real UK registry ) - assert result.terminal_gates.passed - weights_audit = next( - gate for gate in result.terminal_gates.results if gate.name == "weights_audit" - ) - assert weights_audit.details["resolved_weight_kinds"] == { + gates = with_evidence.gate_report["gates"] + assert gates["uk_weights_audit"]["status"] == "passed" + assert gates["uk_weights_audit"]["details"]["resolved_weight_kinds"] == { "uk_frs_only_spi_fill": "importance", "uk_spi_2022_23_income": "design", } - assert [gate.name for gate in result.terminal_gates.results][-3:] == [ - "export_surface", - "target_surface", - "target_fit", - ] + for entry_id in ("uk_export_surface", "uk_target_surface", "uk_target_fit"): + assert gates[entry_id]["status"] == "passed", entry_id + + without_evidence = _run_national_build( + input_h5=input_h5, + staging_h5=tmp_path / "staging2.h5", + stages=(UKNationalStage("hmrc_spi_income", _RecordedFitStage()),), + coverage_engine=object(), + terminal_gate_path=tmp_path / "terminal_gates2.json", + gate_registry=None, + ) + + gates = without_evidence.gate_report["gates"] + for entry_id in ("uk_export_surface", "uk_target_surface", "uk_target_fit"): + assert gates[entry_id]["status"] == "evidence_absent", entry_id + assert gates[entry_id]["reason"] == "missing evidence: parity_evidence" def test_national_build_rejects_both_gate_path_names_and_h5_collisions( @@ -924,10 +899,9 @@ def test_national_build_rejects_both_gate_path_names_and_h5_collisions( def test_national_build_rejects_duplicate_stage_names_before_running( - monkeypatch, tmp_path + tmp_path, ) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" _write_toy_h5(input_h5) @@ -938,12 +912,6 @@ def transform(frame: Frame) -> Frame: called = True return frame - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - with pytest.raises(ValueError, match="Duplicate UK national stage"): _run_national_build( input_h5=input_h5, @@ -958,11 +926,17 @@ def transform(frame: Frame) -> Frame: assert called is False -def test_national_build_manifest_failure_removes_stale_outputs_before_stages( - monkeypatch, tmp_path +def test_national_build_manifest_failure_blocks_before_stages_with_a_report( + tmp_path, ) -> None: + """Preflight drift blocks before any stage — and now leaves a report. + + The legacy assertions raised bare, deleting the stale outputs and + writing nothing; the battery persists the refusal as a schema-4 report + with the terminal entries honestly ``unreached``. + """ + pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" staging_h5 = tmp_path / "staging.h5" @@ -977,42 +951,52 @@ def stage_transform(frame: Frame) -> Frame: stage_called = True return frame - def reject_manifest(**_kwargs) -> None: - raise ValueError("manifest drift") + def drifting_coverage(context, parameters): + if parameters.get("check") == "manifest_current": + raise ValueError("manifest drift") + return _passing_gate() - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - reject_manifest, + registry = _toy_gate_registry() + registry["release_input_coverage"] = UKGateBinding( + name="release_input_coverage", + evaluator=drifting_coverage, + parameter_keys=frozenset({"check"}), + artifact_keys=frozenset({"coverage_engine"}), + frame_predicate=( + lambda parameters: parameters.get("check") != "manifest_current" + ), + legacy_name="uk_release_input_coverage", ) - with pytest.raises(ValueError, match="manifest drift"): + with pytest.raises(GateBatteryBlockedError, match="manifest drift") as error: _run_national_build( input_h5=input_h5, staging_h5=staging_h5, stages=(UKNationalStage("should_not_run", stage_transform),), coverage_engine=object(), input_coverage_path=coverage_json, + gate_registry=registry, ) + assert error.value.phase == "preflight" assert stage_called is False assert not staging_h5.exists() - assert not coverage_json.exists() + payload = json.loads(coverage_json.read_text()) + assert payload["schema_version"] == 4 + assert payload["blocked_at_phase"] == "preflight" + assert ( + payload["gates"]["uk_release_input_coverage_manifest_current"]["status"] + == "failed" + ) + assert payload["gates"]["uk_release_input_coverage"]["status"] == "unreached" + assert payload["gates"]["uk_weight_ratio"]["status"] == "unreached" -def test_national_build_rejects_stage_that_breaks_entity_links( - monkeypatch, tmp_path -) -> None: +def test_national_build_rejects_stage_that_breaks_entity_links(tmp_path) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" _write_toy_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) def break_links(frame: Frame) -> Frame: person = frame.table("person").copy() @@ -1055,18 +1039,12 @@ def break_links(frame: Frame) -> Frame: ], ) def test_national_build_rejects_invalid_stage_population_metadata( - monkeypatch, tmp_path, stage_name, transform, message + tmp_path, stage_name, transform, message ) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" _write_toy_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) with pytest.raises(ValueError, match=message): _run_national_build( @@ -1077,17 +1055,11 @@ def test_national_build_rejects_invalid_stage_population_metadata( ) -def test_national_build_refuses_to_overwrite_its_input(monkeypatch, tmp_path) -> None: +def test_national_build_refuses_to_overwrite_its_input(tmp_path) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build input_h5 = tmp_path / "base.h5" _write_toy_h5(input_h5) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) with pytest.raises(ValueError, match="must differ"): _run_national_build( @@ -1097,32 +1069,20 @@ def test_national_build_refuses_to_overwrite_its_input(monkeypatch, tmp_path) -> ) -def test_national_build_accepts_hugging_face_style_h5_symlink( - monkeypatch, tmp_path -) -> None: +def test_national_build_accepts_hugging_face_style_h5_symlink(tmp_path) -> None: pytest.importorskip("tables") - from microcosm.build.uk_runtime import national_build cached_blob = tmp_path / "content-addressed-blob" input_h5 = tmp_path / "populace_uk_2023.h5" staging_h5 = tmp_path / "staging.h5" _write_toy_h5(cached_blob, employment_income=40_000.0) input_h5.symlink_to(cached_blob) - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) result = _run_national_build( input_h5=input_h5, staging_h5=staging_h5, coverage_engine=object(), + gate_registry=_registry_with_coverage(_passing_gate), ) assert result.input_h5 == cached_blob.resolve() @@ -1224,18 +1184,8 @@ def test_checkpointed_build_matches_the_monolith(monkeypatch, tmp_path) -> None: pytest.importorskip("tables") pytest.importorskip("h5py") - from microcosm.build.uk_runtime import national_build - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) + registry = _registry_with_coverage(_passing_gate) input_h5 = tmp_path / "base.h5" _write_two_row_h5(input_h5) run_config = {"input_sha256": "a" * 64, "seed": 42} @@ -1245,6 +1195,7 @@ def test_checkpointed_build_matches_the_monolith(monkeypatch, tmp_path) -> None: input_h5=input_h5, staging_h5=tmp_path / "mono.h5", stages=(_counting_stage("one"), _counting_stage("two")), + gate_registry=registry, ) calls: list[str] = [] _run_national_build( @@ -1254,6 +1205,7 @@ def test_checkpointed_build_matches_the_monolith(monkeypatch, tmp_path) -> None: stages=(_counting_stage("one", calls), _counting_stage("two", calls)), checkpoint_dir=tmp_path / "checkpoints", run_config=run_config, + gate_registry=registry, ) assert calls == ["one", "two"] _assert_same_staging_payload(tmp_path / "mono.h5", tmp_path / "staged.h5") @@ -1274,6 +1226,7 @@ def test_checkpointed_build_matches_the_monolith(monkeypatch, tmp_path) -> None: ), checkpoint_dir=tmp_path / "checkpoints", run_config=run_config, + gate_registry=registry, ) assert resumed_calls == [] _assert_same_staging_payload(tmp_path / "mono.h5", tmp_path / "resumed.h5") @@ -1284,18 +1237,8 @@ def test_checkpointed_build_resumes_past_a_crash(monkeypatch, tmp_path) -> None: pytest.importorskip("tables") pytest.importorskip("h5py") - from microcosm.build.uk_runtime import national_build - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) + registry = _registry_with_coverage(_passing_gate) input_h5 = tmp_path / "base.h5" _write_two_row_h5(input_h5) run_config = {"input_sha256": "a" * 64, "seed": 42} @@ -1314,6 +1257,7 @@ def exploding(frame: Frame) -> Frame: ), checkpoint_dir=tmp_path / "checkpoints", run_config=run_config, + gate_registry=registry, ) calls: list[str] = [] @@ -1324,27 +1268,18 @@ def exploding(frame: Frame) -> Frame: stages=(_counting_stage("one", calls), _counting_stage("two", calls)), checkpoint_dir=tmp_path / "checkpoints", run_config=run_config, + gate_registry=registry, ) assert calls == ["two"] -def test_checkpointed_build_pins_the_run_config(monkeypatch, tmp_path) -> None: +def test_checkpointed_build_pins_the_run_config(tmp_path) -> None: """Resuming under a different configuration is refused, never blended.""" pytest.importorskip("tables") pytest.importorskip("h5py") - from microcosm.build.uk_runtime import national_build - monkeypatch.setattr( - national_build, - "assert_uk_release_input_coverage_manifest_current", - lambda **_kwargs: None, - ) - monkeypatch.setattr( - national_build, - "uk_release_input_coverage_gate", - lambda _dataset, _engine: _passing_gate(), - ) + registry = _registry_with_coverage(_passing_gate) input_h5 = tmp_path / "base.h5" _write_two_row_h5(input_h5) @@ -1355,6 +1290,7 @@ def test_checkpointed_build_pins_the_run_config(monkeypatch, tmp_path) -> None: staging_h5=tmp_path / "unpinned.h5", stages=(_counting_stage("one"),), checkpoint_dir=tmp_path / "checkpoints", + gate_registry=registry, ) _run_national_build( @@ -1364,6 +1300,7 @@ def test_checkpointed_build_pins_the_run_config(monkeypatch, tmp_path) -> None: stages=(_counting_stage("one"),), checkpoint_dir=tmp_path / "checkpoints", run_config={"input_sha256": "a" * 64, "seed": 42}, + gate_registry=registry, ) with pytest.raises(ValueError, match="new checkpoint directory"): _run_national_build( @@ -1373,4 +1310,69 @@ def test_checkpointed_build_pins_the_run_config(monkeypatch, tmp_path) -> None: stages=(_counting_stage("one"),), checkpoint_dir=tmp_path / "checkpoints", run_config={"input_sha256": "b" * 64, "seed": 42}, + gate_registry=registry, + ) + + +def test_release_candidate_blocks_on_named_evidence_gaps(tmp_path) -> None: + """The chartered semantics live: a candidate cannot excuse absent + evidence, a dev build records the same gaps and continues.""" + + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + _write_toy_h5(input_h5) + registry = _registry_with_coverage(_passing_gate) + + dev = _run_national_build( + input_h5=input_h5, + staging_h5=tmp_path / "dev.h5", + coverage_engine=object(), + terminal_gate_path=tmp_path / "dev_gates.json", + gate_registry=registry, + ) + assert dev.gate_report["blocked_at_phase"] is None + absent = { + entry_id + for entry_id, gate in dev.gate_report["gates"].items() + if gate["status"] == "evidence_absent" + } + assert "uk_weight_ratio" in absent # unbound in the toy registry + + with pytest.raises(GateBatteryBlockedError) as error: + _run_national_build( + input_h5=input_h5, + staging_h5=tmp_path / "candidate.h5", + coverage_engine=object(), + terminal_gate_path=tmp_path / "candidate_gates.json", + gate_registry=registry, + release_candidate=True, ) + assert error.value.phase == "terminal" + assert "[uk_weight_ratio]" in str(error.value) + assert not (tmp_path / "candidate.h5").exists() + + +def test_release_candidate_is_refused_on_a_rung_before_any_unlink( + tmp_path, +) -> None: + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + _write_toy_h5(input_h5) + terminal_json = tmp_path / "terminal_gates.json" + terminal_json.write_text('{"previous_report": true}\n') + + with pytest.raises(ValueError, match="structurally non-releasable"): + _run_national_build( + input_h5=input_h5, + staging_h5=tmp_path / "staging.h5", + coverage_engine=object(), + terminal_gate_path=terminal_json, + sample_fraction=0.5, + release_candidate=True, + ) + + # Configuration refusals precede the sidecar unlinks: the contradictory + # request must not destroy the previous run's report. + assert terminal_json.read_text() == '{"previous_report": true}\n' From a69aace7551ce1623e46be78c93982f6fa902b28 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:46:04 +0200 Subject: [PATCH 4/7] Drive the battery from the CLI: typed blocks and the candidate posture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The driver catches GateBatteryBlockedError instead of string-matching the "Release gates failed:" prefix (stage reports are written only for a terminal block — a preflight block ran no stage), gains --release-candidate with a parser refusal on any sampled rung, and forwards the flag to the build. The stdout payload (schema 4 -> 5) and the build record (schema 2 -> 3) embed the schema-4 gate report, and the record mirrors calibration_diagnostics_sha256 top-level from the report's signed release_evidence slot. _is_final_release_gate_failure is deleted; the canonical-release-id fence stays as defense in depth over the id namespace. Co-Authored-By: Claude Fable 5 --- .../tests/test_uk_national_build_driver.py | 159 +++++++++++++++--- tools/build_uk_national_dataset.py | 53 ++++-- 2 files changed, 177 insertions(+), 35 deletions(-) diff --git a/packages/microcosm-build/tests/test_uk_national_build_driver.py b/packages/microcosm-build/tests/test_uk_national_build_driver.py index bd5785cc9..97d7508cb 100644 --- a/packages/microcosm-build/tests/test_uk_national_build_driver.py +++ b/packages/microcosm-build/tests/test_uk_national_build_driver.py @@ -10,6 +10,7 @@ import pandas as pd import pytest +from microcosm.build.gate_battery import GateBatteryBlockedError from microcosm.build.uk_runtime.national_frame import ( UKStagingProvenance, _uk_source_file_fingerprint, @@ -78,26 +79,27 @@ def _gate_result(*, passed: bool) -> SimpleNamespace: ) -def _terminal_gates(input_coverage: SimpleNamespace) -> SimpleNamespace: - manifest = { - "passed": bool(input_coverage.passed), +def _fake_gate_report(input_coverage: SimpleNamespace) -> dict: + """A schema-4-shaped payload as the build result now carries it.""" + + return { + "schema_version": 4, + "blocked_at_phase": None, + "shippable": False, + "release_evidence": {"calibration_diagnostics_sha256": "c" * 64}, "gates": { "uk_release_input_coverage": { - "passed": bool(input_coverage.passed), + "status": "passed" if input_coverage.passed else "failed", "failures": list(input_coverage.failures), "details": dict(input_coverage.details), }, - "weight_ess": { - "passed": True, + "uk_weight_ess": { + "status": "passed", "failures": [], "details": {"ess_fraction": 0.5}, }, }, } - return SimpleNamespace( - passed=bool(input_coverage.passed), - to_manifest=lambda: manifest, - ) def _load_builder_module(): @@ -155,7 +157,8 @@ def fake_build(**kwargs): "frs_hmrc_retained_leaves", "hmrc_spi_income", ), - terminal_gates=_terminal_gates(input_coverage), + phase_reports=(), + gate_report=_fake_gate_report(input_coverage), input_coverage=input_coverage, sampling_receipt=None, ) @@ -223,15 +226,18 @@ def fake_build(**kwargs): assert calls[0]["terminal_gate_path"] == staging_h5.with_suffix( ".terminal_gates.json" ) + assert calls[0]["release_candidate"] is False payload = json.loads(capsys.readouterr().out) + assert payload["schema_version"] == 5 assert payload["build_kind"] == "uk_national_staging_dataset" assert payload["stages"] == [ "frs_hmrc_retained_leaves", "hmrc_spi_income", ] assert payload["input_coverage"]["passed"] is True - assert payload["terminal_gates"]["passed"] is True - assert payload["terminal_gates"]["gates"]["weight_ess"]["passed"] is True + assert payload["terminal_gates"]["schema_version"] == 4 + assert payload["terminal_gates"]["blocked_at_phase"] is None + assert payload["terminal_gates"]["gates"]["uk_weight_ess"]["status"] == "passed" assert payload["hmrc_replay"]["summary"] == {"excluded_with_fence": 208} assert payload["artifacts"]["staging_h5"]["sha256"] evidence_path = staging_h5.with_suffix(".hmrc_income.json") @@ -249,8 +255,10 @@ def fake_build(**kwargs): assert payload["artifacts"]["terminal_gates"]["sha256"] assert payload["artifacts"]["build_record"]["sha256"] record = json.loads(build_record_path.read_text(encoding="utf-8")) + assert record["schema_version"] == 3 assert record["status"] == "passed" - assert record["terminal_gates"]["gates"]["weight_ess"]["passed"] is True + assert record["calibration_diagnostics_sha256"] == "c" * 64 + assert record["terminal_gates"]["gates"]["uk_weight_ess"]["status"] == "passed" assert record["dataset"] == { "entity_rows": {"benunit": 1, "household": 1, "person": 2}, "household_weight_kind": "importance", @@ -300,10 +308,14 @@ def fake_build(**kwargs): evidence=lambda: {"stage": "hmrc_spi_income"}, replay_report=replay_report, ) - kwargs["terminal_gate_path"].write_text('{"passed": false}\n') - raise RuntimeError( - "Release gates failed: [uk_release_input_coverage] gift_aid remains " - "a reviewed exclusion with positive effective-mass signal" + kwargs["terminal_gate_path"].write_text('{"blocked_at_phase": "terminal"}\n') + raise GateBatteryBlockedError( + "terminal", + [ + "[uk_release_input_coverage] gift_aid remains a reviewed " + "exclusion with positive effective-mass signal" + ], + kwargs["terminal_gate_path"], ) monkeypatch.setattr(builder, "build_uk_national_dataset", fake_build) @@ -346,7 +358,7 @@ def fake_write_replay(report, path): ], ) - with pytest.raises(RuntimeError, match="Release gates failed"): + with pytest.raises(GateBatteryBlockedError, match="Gate battery blocked"): builder.main() evidence = json.loads( @@ -362,6 +374,81 @@ def fake_write_replay(report, path): assert not staging_h5.with_suffix(".build.json").exists() +def test_national_driver_writes_no_stage_reports_for_a_preflight_block( + monkeypatch, + tmp_path, +) -> None: + """A preflight block ran no stage; aggregate reports would be fiction.""" + + builder = _load_builder_module() + input_h5 = tmp_path / "base.h5" + staging_h5 = tmp_path / "staging.h5" + spi_tab = tmp_path / "put2223uk.tab" + hmrc_ods = tmp_path / "hmrc.ods" + frs_raw_dir = tmp_path / "frs_2023_24" + frs_raw_dir.mkdir() + for path in ( + input_h5, + spi_tab, + hmrc_ods, + frs_raw_dir / "adult.tab", + frs_raw_dir / "benefits.tab", + ): + path.write_bytes(b"source") + + def fake_build(**kwargs): + kwargs["terminal_gate_path"].write_text('{"blocked_at_phase": "preflight"}\n') + raise GateBatteryBlockedError( + "preflight", + ["[uk_release_input_coverage_manifest_current] manifest drift"], + kwargs["terminal_gate_path"], + ) + + monkeypatch.setattr(builder, "build_uk_national_dataset", fake_build) + monkeypatch.setattr( + builder, + "verify_certified_uk_candidate", + lambda path: SimpleNamespace( + path=Path(path).resolve(), + filename="populace_uk_2023.h5", + tier="frs", + revision="test-revision", + sha256="a" * 64, + size_bytes=6, + ), + ) + monkeypatch.setattr( + builder, + "write_hmrc_replay_report", + lambda *_args: pytest.fail("preflight blocks must not emit replay reports"), + ) + monkeypatch.setattr( + sys, + "argv", + [ + "build_uk_national_dataset.py", + *_IDENTITY_CLI_ARGUMENTS, + "--input-h5", + str(input_h5), + "--staging-h5", + str(staging_h5), + "--frs-raw-dir", + str(frs_raw_dir), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(hmrc_ods), + ], + ) + + with pytest.raises(GateBatteryBlockedError): + builder.main() + + assert not staging_h5.with_suffix(".hmrc_income.json").exists() + assert not staging_h5.with_suffix(".hmrc_replay.json").exists() + assert not staging_h5.with_suffix(".build.json").exists() + + def test_national_driver_does_not_write_reports_for_stage_failure( monkeypatch, tmp_path, @@ -840,6 +927,40 @@ def test_national_driver_refuses_canonical_release_ids_for_rung_builds( assert "non-releasable" in capsys.readouterr().err +def test_national_driver_refuses_release_candidate_on_a_rung( + monkeypatch, capsys +) -> None: + builder = _load_builder_module() + monkeypatch.setattr( + sys, + "argv", + [ + "build_uk_national_dataset.py", + "--release-id", + "uk-dev-rung-check", + "--calibration-diagnostics-sha256", + "c" * 64, + "--input-h5", + "base.h5", + "--staging-h5", + "staging.h5", + "--frs-raw-dir", + "frs_2023_24", + "--spi-tab", + "put2223uk.tab", + "--hmrc-ods", + "hmrc.ods", + "--sample-fraction", + "0.10", + "--release-candidate", + ], + ) + + with pytest.raises(SystemExit): + builder._parse_args() + assert "non-releasable" in capsys.readouterr().err + + def test_staging_run_config_pins_the_sampling_identity(monkeypatch, tmp_path) -> None: builder = _load_builder_module() for name in ("adult.tab", "benefits.tab", "put2223uk.tab", "hmrc.ods"): diff --git a/tools/build_uk_national_dataset.py b/tools/build_uk_national_dataset.py index e7d9e4a4d..0dfa97062 100644 --- a/tools/build_uk_national_dataset.py +++ b/tools/build_uk_national_dataset.py @@ -12,6 +12,7 @@ import pandas as pd +from microcosm.build.gate_battery import GateBatteryBlockedError from microcosm.build.uk_runtime.frs_hmrc_leaves import ( UKFRSHMRCRetainedLeavesStageTransform, ) @@ -50,8 +51,9 @@ #: artifacts; a sampled rung build must never carry one. Mirrors the #: microcosm-data contract's release-identity check without importing the #: data shard into the build tool. The durable coupling is the gate -#: battery's ``release_candidate`` flag; this fence holds until the #611 -#: consumer half wires it. +#: battery's ``release_candidate`` flag (wired below: ``--release-candidate`` +#: is refused on a rung); this fence stays as defense in depth over the id +#: namespace itself. # Year and count widths mirror the microcosm-data contract's release-identity # regex ([1-9][0-9]*), and the tier alternation is built from the build # shard's ratified UK_RELEASE_TIERS so a newly ratified tier is fenced @@ -292,11 +294,28 @@ def _parse_args() -> argparse.Namespace: "Reviewed degenerate-release-surface exclusion register " f"overriding the committed {UK_DEGENERATE_EXCLUSION_REGISTER_RESOURCE} " "(#630). Stale entries fail the gate; dormant entries are " - "reported. The gate is always armed; the override changes the " - "run's policy digest away from the certified pin." + "reported. The gate is always armed; the override is digested " + "into the report's evidence_sha256, so an overridden run " + "self-describes against the committed register." + ), + ) + parser.add_argument( + "--release-candidate", + action="store_true", + help=( + "Arm the battery's release-candidate posture: every " + "evidence_absent gap blocks instead of being recorded. Refused " + "on a sampled rung — a rung is structurally non-releasable " + "(#627). Default off: the staging build records its gaps " + "honestly and continues." ), ) args = parser.parse_args() + if args.release_candidate and args.sample_fraction != 1.0: + parser.error( + "--release-candidate is refused on a sampled rung; a rung build " + "is structurally non-releasable (#627)." + ) if args.sample_seed < 0: parser.error("sample seed must be a non-negative integer.") if args.sample_fraction != 1.0 and ( @@ -503,6 +522,7 @@ def main() -> int: **checkpoint_arguments, sample_fraction=args.sample_fraction, sample_seed=args.sample_seed, + release_candidate=args.release_candidate, ) except ValueError as error: if args.sample_fraction != 1.0 and _RUNG_NAMED_EDGE_SIGNATURE in str(error): @@ -531,9 +551,12 @@ def main() -> int: print(json.dumps(receipt, indent=2, sort_keys=True)) return _RUNG_ABORT_EXIT_CODE raise - except RuntimeError as error: + except GateBatteryBlockedError as error: + # Only the terminal block leaves completed stage evidence behind; a + # preflight block ran no stage, and any other RuntimeError is a + # stage failure that must not be dressed in aggregate reports. if ( - _is_final_release_gate_failure(error) + error.phase == "terminal" and retained_leaves_transform.last_result is not None and hmrc_transform.last_result is not None ): @@ -582,7 +605,7 @@ def main() -> int: ) _write_json(build_record_path, build_record) payload = { - "schema_version": 4, + "schema_version": 5, "build_kind": "uk_national_staging_dataset", "sampling": { "sample_fraction": float(args.sample_fraction), @@ -590,7 +613,7 @@ def main() -> int: "rung_token": UK_SAMPLE_RUNG_TOKENS[args.sample_fraction], }, "stages": list(result.stage_names), - "terminal_gates": result.terminal_gates.to_manifest(), + "terminal_gates": dict(result.gate_report), "input_coverage": { "passed": result.input_coverage.passed, "failures": list(result.input_coverage.failures), @@ -727,10 +750,14 @@ def _aggregate_build_record( household_weights = pd.to_numeric( result.frame.table("household")["household_weight"], errors="raise" ) + release_evidence = dict(result.gate_report["release_evidence"]) return { - "schema_version": 2, + "schema_version": 3, "build_kind": "uk_national_staging_dataset", "status": "passed", + "calibration_diagnostics_sha256": release_evidence[ + "calibration_diagnostics_sha256" + ], "stages": list(result.stage_names), "parameters": { "seed": int(seed), @@ -770,7 +797,7 @@ def _aggregate_build_record( ), }, "source_vintages": dict(family_evidence.get("source_vintages", {})), - "terminal_gates": result.terminal_gates.to_manifest(), + "terminal_gates": dict(result.gate_report), "input_coverage": { "passed": bool(result.input_coverage.passed), "failures": list(result.input_coverage.failures), @@ -873,12 +900,6 @@ def _replay_summary(hmrc_result: object) -> dict[str, object]: return dict(summary) -def _is_final_release_gate_failure(error: RuntimeError) -> bool: - """Match only the national seam's post-stage, pre-staging hard gate.""" - - return str(error).startswith("Release gates failed:") - - def _validate_distinct_paths( *, evidence_path: Path, From a09e3af18c020b5f87f6a62d56c35999ebaada0c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:46:04 +0200 Subject: [PATCH 5/7] Record the consumer flip: contract doc line and changelog Co-Authored-By: Claude Fable 5 --- changelog.d/611-uk-battery-consumer.changed.md | 1 + docs/gate-battery-contract.md | 6 +++++- 2 files changed, 6 insertions(+), 1 deletion(-) create mode 100644 changelog.d/611-uk-battery-consumer.changed.md diff --git a/changelog.d/611-uk-battery-consumer.changed.md b/changelog.d/611-uk-battery-consumer.changed.md new file mode 100644 index 000000000..8a1e5014e --- /dev/null +++ b/changelog.d/611-uk-battery-consumer.changed.md @@ -0,0 +1 @@ +The UK national build runs its gates through the shared `GateBatteryRun` executor: preflight before the frame loads, terminal immediately before the staging writer, one schema-4 report at the existing `terminal_gates.json` path with every declared entry present — evidence the build cannot supply is a named `evidence_absent` gap that blocks only under the new `--release-candidate` posture (refused on a sampled rung, the #627 coupling). Gate failures raise the typed `GateBatteryBlockedError` in place of the `"Release gates failed:"` string, preflight refusals now leave a persisted report, the signing key moves to `MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY` (unsigned full-scale builds still refuse to stage; rungs proceed with an honest `shippable: false`), and `calibration_diagnostics_sha256` rides the new signed `release_evidence` slot plus the build record. One exclusion-expiry clock is threaded to every exclusion-consuming gate, and a `--degenerate-exclusions` override is digested into the report's `evidence_sha256` so an overridden run self-describes. The legacy `uk_terminal_gate_report` remains solely as the differential-test oracle; `_UKGateEvidence`, the duplicated `_evaluate_gate`, and the driver's error string-match are deleted. The schema-3 release verifier cannot read schema-4 reports — the schema-4 verification path must land before any exact-k release assembly. diff --git a/docs/gate-battery-contract.md b/docs/gate-battery-contract.md index fabc8b935..8eede2182 100644 --- a/docs/gate-battery-contract.md +++ b/docs/gate-battery-contract.md @@ -224,7 +224,11 @@ stretches on declaration alone: | criticality mix | 8 blocking + 1 diagnostic | all blocking | The differences are entirely in the two country inputs — the spec file -and the registry — which is the point. +and the registry — which is the point. The UK national build is the +executor's production consumer: it constructs one `GateBatteryRun` per +build and runs both phases under `BLOCKS_ARTIFACT` (preflight before the +frame loads, terminal immediately before the staging writer), while BE +remains spec-only. ## The reference rule From a3355a7d12abf6adc27d19199e3eabb3b1e693a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:14:00 +0200 Subject: [PATCH 6/7] Adversarial-review fixes: honest override label, no-destruction fence, alias posture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Findings from the review round (adversarial pass + Vahid on #662): The override label follows content, not artifact presence. The driver materialized the committed register even without --degenerate-exclusions and passed it through the artifact channel, so every default run's signed evidence self-described as an override — the loud-override signal inverted, and the build record contradicted the report. The driver now preflights the committed register without passing it, and the binding labels by comparing the resolved records' policy payloads to the committed register: byte-identical content is the committed policy whichever route delivered it. The no-destruction-before-validation fence now covers every validation: the exclusion clock and the battery construction (release identity, spec parameters, release_evidence values) moved above the sidecar unlinks, so an empty release id, an empty diagnostics digest, or a datetime clock can no longer destroy a previous build's artifacts before refusing. --release-candidate with the schema-1 alias is refused at both layers: in alias mode the alias is last-written over the report path, and a release candidate must keep its signed schema-4 report. The alias writer is now atomic like every other writer on this surface, and a failing alias write during a terminal block chains under the typed GateBatteryBlockedError instead of displacing it. From Vahid's review: both operator docs (release.env.example, README) now name both signing variables and which verification path reads each; the changelog flags the silent string-match break for downstream catchers; the ordering invariant is named in code. Also: the release_evidence signature test asserts baseline equality before tamper inequality, the clock's arm-time (build-start) semantics are documented, the stale consolidation note in battery_bindings is updated, and a dead shippable conjunct is removed. Co-Authored-By: Claude Fable 5 --- README.md | 24 +++++--- .../611-uk-battery-consumer.changed.md | 2 +- .../src/microcosm/build/gate_battery.py | 1 - .../build/uk_runtime/battery_bindings.py | 32 ++++++---- .../build/uk_runtime/national_build.py | 59 +++++++++++++++---- .../tests/test_gate_battery.py | 22 ++++--- .../tests/test_uk_battery_bindings.py | 27 +++++++++ .../tests/test_uk_national_build.py | 50 +++++++++++++++- .../tests/test_uk_national_build_driver.py | 35 +++++++++++ tools/build_uk_national_dataset.py | 19 ++++-- tools/release.env.example | 5 ++ 11 files changed, 232 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index bdef84bd1..ca829be3f 100644 --- a/README.md +++ b/README.md @@ -141,11 +141,19 @@ warns if neither is set. After that, every release publishes with an automatic Slack alert. Canonical UK exact-k builds also require a stable, base64-encoded 32-byte -`POPULACE_UK_TERMINAL_GATE_SIGNING_KEY`. Source `tools/release.env` before the -national build as well as publication. The terminal-gate aggregator authenticates -the complete report, canonical release id, and exact calibration-diagnostics -digest with HMAC-SHA256; the persistence seam -cannot sign caller-composed gate results, and publication independently verifies -the report from the same out-of-band key. If the key is missing or malformed, -the writer first persists an unsigned failed report and then raises, and -publication rejects it. +release key. Source `tools/release.env` before the national build as well as +publication. Two variables carry it during the report-format migration — +export both from the same key material: + +- `MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY` — what the national build signs + with (the gate-battery executor) and what schema-4 report verification + reads. +- `POPULACE_UK_TERMINAL_GATE_SIGNING_KEY` — what schema-3 (legacy-format) + report verification reads; retires with the legacy format. + +The gate battery authenticates the complete report, canonical release id, and +exact calibration-diagnostics digest with HMAC-SHA256; the persistence seam +cannot sign caller-composed gate results, and publication independently +verifies the report from the same out-of-band key. If the key is missing or +malformed, a full-scale build persists the unsigned report and then refuses +to stage, and publication rejects unsigned reports. diff --git a/changelog.d/611-uk-battery-consumer.changed.md b/changelog.d/611-uk-battery-consumer.changed.md index 8a1e5014e..6df9461e8 100644 --- a/changelog.d/611-uk-battery-consumer.changed.md +++ b/changelog.d/611-uk-battery-consumer.changed.md @@ -1 +1 @@ -The UK national build runs its gates through the shared `GateBatteryRun` executor: preflight before the frame loads, terminal immediately before the staging writer, one schema-4 report at the existing `terminal_gates.json` path with every declared entry present — evidence the build cannot supply is a named `evidence_absent` gap that blocks only under the new `--release-candidate` posture (refused on a sampled rung, the #627 coupling). Gate failures raise the typed `GateBatteryBlockedError` in place of the `"Release gates failed:"` string, preflight refusals now leave a persisted report, the signing key moves to `MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY` (unsigned full-scale builds still refuse to stage; rungs proceed with an honest `shippable: false`), and `calibration_diagnostics_sha256` rides the new signed `release_evidence` slot plus the build record. One exclusion-expiry clock is threaded to every exclusion-consuming gate, and a `--degenerate-exclusions` override is digested into the report's `evidence_sha256` so an overridden run self-describes. The legacy `uk_terminal_gate_report` remains solely as the differential-test oracle; `_UKGateEvidence`, the duplicated `_evaluate_gate`, and the driver's error string-match are deleted. The schema-3 release verifier cannot read schema-4 reports — the schema-4 verification path must land before any exact-k release assembly. +The UK national build runs its gates through the shared `GateBatteryRun` executor: preflight before the frame loads, terminal immediately before the staging writer, one schema-4 report at the existing `terminal_gates.json` path with every declared entry present — evidence the build cannot supply is a named `evidence_absent` gap that blocks only under the new `--release-candidate` posture (refused on a sampled rung, the #627 coupling). Gate failures raise the typed `GateBatteryBlockedError` in place of the `"Release gates failed:"` string — any downstream consumer matching that prefix breaks silently and must switch to catching the typed error — preflight refusals now leave a persisted report, the signing key moves to `MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY` (unsigned full-scale builds still refuse to stage; rungs proceed with an honest `shippable: false`), and `calibration_diagnostics_sha256` rides the new signed `release_evidence` slot plus the build record. One exclusion-expiry clock is threaded to every exclusion-consuming gate — resolved when the battery is armed, before the stages, where the legacy aggregator resolved it after them — and a `--degenerate-exclusions` override is digested into the report's `evidence_sha256` so an overridden run self-describes; the override label follows the records' content, so re-supplying the committed register is not a deviation. The legacy `uk_terminal_gate_report` remains solely as the differential-test oracle; `_UKGateEvidence`, the duplicated `_evaluate_gate`, and the driver's error string-match are deleted. The schema-3 release verifier cannot read schema-4 reports — the schema-4 verification path must land before any exact-k release assembly. diff --git a/packages/microcosm-build/src/microcosm/build/gate_battery.py b/packages/microcosm-build/src/microcosm/build/gate_battery.py index 26ad06834..79a7513eb 100644 --- a/packages/microcosm-build/src/microcosm/build/gate_battery.py +++ b/packages/microcosm-build/src/microcosm/build/gate_battery.py @@ -948,7 +948,6 @@ def report_payload(self) -> dict[str, object]: and self._blocked_at_phase is None and blocking_ok and signing_key is not None - and self._gates_manifest_sha256 is not None ) attestation: dict[str, object] = { "schema_version": GATE_BATTERY_ATTESTATION_SCHEMA_VERSION, 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 944c94a0b..8e7a57c51 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 @@ -19,11 +19,10 @@ with the declared one. Any *other* unexpected name passes through untouched so that check keeps biting. -The evidence surface handed to the legacy gate modules mirrors the -national build's adapter (the three entity tables plus period, -weight-kind, and mass-log metadata); the two copies consolidate when the -national build swaps onto the battery executor and the legacy -orchestration path retires. +The evidence surface handed to the legacy gate modules (the three entity +tables plus period, weight-kind, and mass-log metadata) lives here as the +single copy: the national build's adapter consolidated into it when the +orchestration swapped onto the battery executor. """ from __future__ import annotations @@ -260,16 +259,29 @@ def _resolve_degenerate_exclusions( The committed register is the reviewed policy of record (#630/#610); a supplied ``reviewed_degenerate_exclusions`` artifact is the loud review-time override — the evidence hook digests whichever resolved, - so an overridden run self-describes in the signed report. + so an overridden run self-describes in the signed report. The label + follows the *content*, not the artifact's presence: records identical + to the committed register are the committed policy whichever route + delivered them, so an override cannot masquerade as a deviation (or a + caller re-supplying the register as a false one). """ + committed = uk_default_degenerate_reviewed_exclusions() override = context.artifacts.get("reviewed_degenerate_exclusions") if override is None: - return uk_default_degenerate_reviewed_exclusions(), "committed" - return ( - coerce_reviewed_exclusions(override, label="UK degenerate-surface policy"), - "override", + return committed, "committed" + resolved = coerce_reviewed_exclusions( + override, label="UK degenerate-surface policy" ) + committed_payload = { + name: record.policy_payload() for name, record in committed.items() + } + resolved_payload = { + name: record.policy_payload() for name, record in resolved.items() + } + if resolved_payload == committed_payload: + return committed, "committed" + return resolved, "override" def _evaluate_degenerate_release_surface( 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 682a3ff74..62b5cf931 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 @@ -37,6 +37,7 @@ from microcosm.build.gate_battery import ( BlockingMode, EvidenceContext, + GateBatteryBlockedError, GateBatteryRun, GateBinding, GatePhaseReport, @@ -128,7 +129,9 @@ class UKNationalBuildResult: stage_names: tuple[str, ...] #: The in-memory phase reports, declared order (preflight, terminal). phase_reports: tuple[GatePhaseReport, ...] - #: The exact schema-4 payload persisted at ``terminal_gate_path``. + #: The schema-4 payload the battery persisted at ``terminal_gate_path`` + #: (in compatibility-alias mode the file is last-written as the schema-1 + #: alias; this field always carries the full battery payload). gate_report: Mapping[str, object] terminal_gate_path: Path #: The #627 rung receipt; ``None`` on a full-scale (fraction 1.0) build. @@ -349,7 +352,10 @@ def build_uk_national_dataset( records the gap and continues. A sampled rung is structurally non-releasable, so requesting both is refused. ``now`` is the shared exclusion-expiry clock (default: today, UTC), threaded to every - exclusion-consuming gate so one report carries one evaluation date. + exclusion-consuming gate so one report carries one evaluation date; it + is resolved once when the battery is armed — before the stages — where + the legacy aggregator resolved it after them, so a receipt expiring + mid-build is judged by the date the build started. ``gate_registry`` overrides the binding registry (tests only). """ @@ -384,8 +390,10 @@ def build_uk_national_dataset( materialized_stages = tuple(stages) _validate_stages(materialized_stages) - # Configuration refusals precede the battery and the sidecar unlinks: a - # misconfigured run must not delete a previous report or write a new one. + # Invariant: no destructive step precedes argument validation. Every + # configuration refusal sits above the sidecar unlinks and the battery, + # so a misconfigured run can neither delete a previous report nor write + # a new one (the #658 --degenerate-exclusions ordering bug, generalized). if checkpoint_dir is not None and run_config is None: raise ValueError( "a checkpointed UK national build requires run_config: the " @@ -408,19 +416,28 @@ def build_uk_national_dataset( "a sampled rung build is structurally non-releasable (#627); " "release_candidate requires sample_fraction == 1.0." ) + if release_candidate and legacy_input_coverage_output: + raise ValueError( + "input_coverage_path is a compatibility alias whose schema-1 " + "payload is last-written over the report path; a release " + "candidate must keep its signed schema-4 report, so the two " + "are mutually exclusive." + ) if (input_mass_reference is None) != (input_mass_policy is None): raise ValueError( "input_mass_parity arms with a frozen reference and reviewed " "thresholds together; supply both or neither." ) - staging_path.unlink(missing_ok=True) - diagnostic_path.unlink(missing_ok=True) engine = ( coverage_engine if coverage_engine is not None else PolicyEngineUKCoverageEngine() ) + # The clock and the battery construction validate their inputs (the + # date's type; release identity, spec parameters, release_evidence + # values), so they sit inside the no-destruction-before-validation + # fence too: the unlinks come strictly last. evaluation_date = exclusion_evaluation_date(now) battery = GateBatteryRun( load_country_spec("uk").gates, @@ -432,6 +449,8 @@ def build_uk_national_dataset( "calibration_diagnostics_sha256": calibration_diagnostics_sha256 }, ) + staging_path.unlink(missing_ok=True) + diagnostic_path.unlink(missing_ok=True) # Mirrors the US cheap preflight: graph or reference drift blocks before # source stages — now with the refusal persisted as a schema-4 report. battery.run_phase( @@ -507,11 +526,21 @@ def build_uk_national_dataset( ) try: battery.enforce("terminal", mode=BlockingMode.BLOCKS_ARTIFACT) - finally: + except GateBatteryBlockedError as blocked: # The alias consumer reads the schema-1 shape at this exact path, in # the blocked case too — same last-write order as the legacy flow. + # A failing alias write must not displace the typed block: the block + # is the build's outcome, the write failure rides along as its cause. if legacy_input_coverage_output and coverage_outcome.result is not None: - _write_input_coverage_diagnostic(diagnostic_path, coverage_outcome.result) + try: + _write_input_coverage_diagnostic( + diagnostic_path, coverage_outcome.result + ) + except Exception as write_error: # noqa: BLE001 - keep the block typed + raise blocked from write_error + raise + if legacy_input_coverage_output: + _write_input_coverage_diagnostic(diagnostic_path, coverage_outcome.result) gate_report = battery.report_payload() attestation = gate_report["attestation"] signing_error = ( @@ -643,12 +672,18 @@ def _stage_fit_weight_records( try: records = getattr(hmrc_stage.transform, "fit_weight_records", None) return () if records is None else tuple(records) - except Exception: # noqa: BLE001 - the weights audit must name the failure + except Exception: # noqa: BLE001 - unreadable records coerce to () and + # fail the audit as missing evidence rather than crashing the batch. return () def _write_input_coverage_diagnostic(path: Path, gate: GateResult) -> None: - """Write the byte-compatible origin/main schema for the legacy alias.""" + """Write the byte-compatible origin/main schema for the legacy alias. + + Atomic like every other writer on this surface: the alias last-writes + over the gate-report path, and a crash mid-write must not leave + truncated JSON where a consumer expects a report. + """ path.parent.mkdir(parents=True, exist_ok=True) payload = { @@ -660,10 +695,12 @@ def _write_input_coverage_diagnostic(path: Path, gate: GateResult) -> None: "details": dict(gate.details), }, } - path.write_text( + temporary_path = path.with_name(path.name + ".tmp") + temporary_path.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) + temporary_path.replace(path) def _weight_kind_from_stored(value: object) -> WeightKind: diff --git a/packages/microcosm-build/tests/test_gate_battery.py b/packages/microcosm-build/tests/test_gate_battery.py index 8b6920533..6272442ea 100644 --- a/packages/microcosm-build/tests/test_gate_battery.py +++ b/packages/microcosm-build/tests/test_gate_battery.py @@ -578,16 +578,22 @@ def test_release_evidence_rides_in_the_report_and_is_signed( assert report["attestation"]["release_evidence"] == expected signature = report["attestation"]["signature"] report["attestation"]["signature"] = None + + def recompute() -> str: + return hmac.new( + base64.b64decode(KEY), + json.dumps( + report, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + + # Baseline first: the untampered reconstruction must reproduce the + # signature, or the tamper inequality below would pass vacuously. + assert recompute() == signature report["release_evidence"]["calibration_diagnostics_sha256"] = "cd" * 32 report["attestation"]["release_evidence"] = report["release_evidence"] - tampered = hmac.new( - base64.b64decode(KEY), - json.dumps( - report, sort_keys=True, separators=(",", ":"), allow_nan=False - ).encode("utf-8"), - hashlib.sha256, - ).hexdigest() - assert tampered != signature, "release_evidence sits outside the signature" + assert recompute() != signature, "release_evidence sits outside the signature" def test_release_evidence_defaults_to_an_empty_mapping(self, tmp_path, signing_env): manifest = _manifest([_entry("t", gate="support")], ["terminal"]) diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index 1537565bc..b2f9da552 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -504,6 +504,33 @@ def test_review_override_is_loud_in_the_evidence_payload(self) -> None: assert overridden["reviewed_exclusions"] == {} assert overridden != committed, "an override must move the evidence digest" + def test_resupplying_the_committed_register_is_not_an_override(self) -> None: + # The label follows content, not the artifact's presence: a caller + # routing the committed register through the artifact (as a driver + # preflight might) runs the committed policy and must say so — and + # a review file byte-identical to the register is no deviation. + from microcosm.build.uk_runtime.terminal_gates import ( + uk_default_degenerate_reviewed_exclusions, + ) + + binding = UK_GATE_REGISTRY["degenerate_release_surface"] + committed = binding.evidence_payload( + EvidenceContext(artifacts={"exclusions_evaluated_on": CLOCK}), {} + ) + resupplied = binding.evidence_payload( + EvidenceContext( + artifacts={ + "exclusions_evaluated_on": CLOCK, + "reviewed_degenerate_exclusions": dict( + uk_default_degenerate_reviewed_exclusions() + ), + } + ), + {}, + ) + assert resupplied == committed + assert resupplied["exclusions_register"] == "committed" + def test_a_datetime_clock_is_refused(self, uk_gates) -> None: person, benunit, household = _tables() frame = uk_national_frame( diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index 305bc3054..f0bbb8130 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from datetime import date +from datetime import date, datetime from pathlib import Path import pandas as pd @@ -1376,3 +1376,51 @@ def test_release_candidate_is_refused_on_a_rung_before_any_unlink( # Configuration refusals precede the sidecar unlinks: the contradictory # request must not destroy the previous run's report. assert terminal_json.read_text() == '{"previous_report": true}\n' + + +@pytest.mark.parametrize( + ("bad_arguments", "match"), + [ + ({"release_id": ""}, "release_id"), + ({"calibration_diagnostics_sha256": ""}, "release_evidence"), + ({"now": datetime(2026, 9, 1, 12, 0)}, "date"), + ( + {"release_candidate": True, "use_alias_path": True}, + "mutually exclusive", + ), + ], + ids=["empty-release-id", "empty-diagnostics-sha", "datetime-clock", "alias"], +) +def test_every_identity_refusal_precedes_the_sidecar_unlinks( + tmp_path, bad_arguments, match +) -> None: + """No destructive step precedes argument validation — for every + validation, including the ones the battery construction owns.""" + + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + _write_toy_h5(input_h5) + staging_h5 = tmp_path / "staging.h5" + terminal_json = tmp_path / "terminal_gates.json" + staging_h5.write_bytes(b"previous-artifact") + terminal_json.write_text('{"previous_report": true}\n') + arguments: dict = { + "input_h5": input_h5, + "staging_h5": staging_h5, + "release_id": TEST_UK_RELEASE_ID, + "calibration_diagnostics_sha256": TEST_UK_CALIBRATION_DIAGNOSTICS_SHA256, + "coverage_engine": object(), + "now": TEST_UK_EXCLUSION_CLOCK, + "gate_registry": _toy_gate_registry(), + "terminal_gate_path": terminal_json, + } + arguments.update(bad_arguments) + if arguments.pop("use_alias_path", False): + arguments["input_coverage_path"] = arguments.pop("terminal_gate_path") + + with pytest.raises((ValueError, TypeError), match=match): + build_uk_national_dataset(**arguments) + + assert staging_h5.read_bytes() == b"previous-artifact" + assert terminal_json.read_text() == '{"previous_report": true}\n' diff --git a/packages/microcosm-build/tests/test_uk_national_build_driver.py b/packages/microcosm-build/tests/test_uk_national_build_driver.py index 97d7508cb..50b2ab137 100644 --- a/packages/microcosm-build/tests/test_uk_national_build_driver.py +++ b/packages/microcosm-build/tests/test_uk_national_build_driver.py @@ -227,6 +227,10 @@ def fake_build(**kwargs): ".terminal_gates.json" ) assert calls[0]["release_candidate"] is False + # No --degenerate-exclusions: the artifact channel stays empty so the + # binding resolves the committed register itself and the run never + # self-describes as an override (the register is still preflighted). + assert calls[0]["reviewed_degenerate_exclusions"] is None payload = json.loads(capsys.readouterr().out) assert payload["schema_version"] == 5 assert payload["build_kind"] == "uk_national_staging_dataset" @@ -961,6 +965,37 @@ def test_national_driver_refuses_release_candidate_on_a_rung( assert "non-releasable" in capsys.readouterr().err +def test_national_driver_refuses_release_candidate_with_the_legacy_alias( + monkeypatch, capsys +) -> None: + builder = _load_builder_module() + monkeypatch.setattr( + sys, + "argv", + [ + "build_uk_national_dataset.py", + *_IDENTITY_CLI_ARGUMENTS, + "--input-h5", + "base.h5", + "--staging-h5", + "staging.h5", + "--frs-raw-dir", + "frs_2023_24", + "--spi-tab", + "put2223uk.tab", + "--hmrc-ods", + "hmrc.ods", + "--input-coverage-json", + "coverage.json", + "--release-candidate", + ], + ) + + with pytest.raises(SystemExit): + builder._parse_args() + assert "signed schema-4 report" in capsys.readouterr().err + + def test_staging_run_config_pins_the_sampling_identity(monkeypatch, tmp_path) -> None: builder = _load_builder_module() for name in ("adult.tab", "benefits.tab", "put2223uk.tab", "hmrc.ods"): diff --git a/tools/build_uk_national_dataset.py b/tools/build_uk_national_dataset.py index 0dfa97062..7be5707dc 100644 --- a/tools/build_uk_national_dataset.py +++ b/tools/build_uk_national_dataset.py @@ -316,6 +316,12 @@ def _parse_args() -> argparse.Namespace: "--release-candidate is refused on a sampled rung; a rung build " "is structurally non-releasable (#627)." ) + if args.release_candidate and args.input_coverage_json is not None: + parser.error( + "--release-candidate is refused with --input-coverage-json; the " + "schema-1 alias is last-written over the report path and a " + "candidate must keep its signed schema-4 report." + ) if args.sample_seed < 0: parser.error("sample seed must be a non-negative integer.") if args.sample_fraction != 1.0 and ( @@ -453,14 +459,19 @@ def main() -> int: # corrupted committed register must not surface hours later at # terminal-gate time. weighted_integrity_arguments = _weighted_integrity_arguments(args) - reviewed_degenerate_exclusions = ( + if args.degenerate_exclusions is None: + # Preflight the committed register without passing it: a corrupted + # register dies here, while the absent artifact leaves the binding + # resolving the same policy of record itself — the artifact stays + # the review-time override channel, so a default run never + # self-describes as an override. uk_default_degenerate_reviewed_exclusions() - if args.degenerate_exclusions is None - else load_uk_reviewed_exclusion_register( + reviewed_degenerate_exclusions = None + else: + reviewed_degenerate_exclusions = load_uk_reviewed_exclusion_register( args.degenerate_exclusions, resource=UK_DEGENERATE_EXCLUSION_REGISTER_RESOURCE, ) - ) candidate = verify_certified_uk_candidate(args.input_h5) evidence_path.unlink(missing_ok=True) replay_path.unlink(missing_ok=True) diff --git a/tools/release.env.example b/tools/release.env.example index f500d1f48..02c5e5b1a 100644 --- a/tools/release.env.example +++ b/tools/release.env.example @@ -15,4 +15,9 @@ export SLACK_WEBHOOK_POPULACE_UK= # Canonical UK exact-k terminal reports are authenticated with a stable # 32-byte release key encoded as base64. Generate and store this once on the # build machine; source this file before both the UK build and publish steps. +# During the report-format migration both variables carry the same key: +# MICROCOSM_* is what the national build signs with (the gate-battery +# executor) and what schema-4 verification reads; POPULACE_* is what +# schema-3 (legacy-format) verification reads and retires with that format. +export MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY= export POPULACE_UK_TERMINAL_GATE_SIGNING_KEY= From 93735e2ab609e5703fd6ab340b76335726e668be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:25:13 +0200 Subject: [PATCH 7/7] Two audit questions, two names: exclusions_policy vs override_supplied MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Vahid's follow-up on the override-label fix: the signed report's evidence labeled by content ("committed" for a register-identical review file) while the #658 build record labeled by presence ("override" whenever the flag was passed) — opposite answers under near-identical names in two artifacts of the same build. Both questions are real, so each now carries its own name: the evidence payload's exclusions_policy answers "which register content governed this run", and the build record's boolean degenerate_exclusions_override_supplied answers "did the operator invoke the override path". The two can honestly disagree, and both artifacts now say so in their comments. Co-Authored-By: Claude Fable 5 --- .../microcosm/build/uk_runtime/battery_bindings.py | 8 +++++++- .../tests/test_uk_battery_bindings.py | 6 +++--- tools/build_uk_national_dataset.py | 14 +++++++------- 3 files changed, 17 insertions(+), 11 deletions(-) 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 8e7a57c51..b066b2e35 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 @@ -307,8 +307,14 @@ def _degenerate_exclusions_evidence( context: EvidenceContext, parameters: Mapping[str, Any] ) -> object: resolved, source = _resolve_degenerate_exclusions(context) + # ``exclusions_policy`` answers "which register content governed this + # run" — deliberately distinct from the build record's + # ``degenerate_exclusions_override_supplied``, which answers "did the + # operator invoke the override path". The two can honestly disagree + # (a review file byte-identical to the committed register), so they + # carry different names. return { - "exclusions_register": source, + "exclusions_policy": source, "reviewed_exclusions": { name: record.policy_payload() for name, record in sorted(resolved.items()) }, diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index b2f9da552..af7db2ed5 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -488,7 +488,7 @@ def test_review_override_is_loud_in_the_evidence_payload(self) -> None: committed = binding.evidence_payload( EvidenceContext(artifacts={"exclusions_evaluated_on": CLOCK}), {} ) - assert committed["exclusions_register"] == "committed" + assert committed["exclusions_policy"] == "committed" assert "household.source_year" in committed["reviewed_exclusions"] overridden = binding.evidence_payload( @@ -500,7 +500,7 @@ def test_review_override_is_loud_in_the_evidence_payload(self) -> None: ), {}, ) - assert overridden["exclusions_register"] == "override" + assert overridden["exclusions_policy"] == "override" assert overridden["reviewed_exclusions"] == {} assert overridden != committed, "an override must move the evidence digest" @@ -529,7 +529,7 @@ def test_resupplying_the_committed_register_is_not_an_override(self) -> None: {}, ) assert resupplied == committed - assert resupplied["exclusions_register"] == "committed" + assert resupplied["exclusions_policy"] == "committed" def test_a_datetime_clock_is_refused(self, uk_gates) -> None: person, benunit, household = _tables() diff --git a/tools/build_uk_national_dataset.py b/tools/build_uk_national_dataset.py index 7be5707dc..9024e5688 100644 --- a/tools/build_uk_national_dataset.py +++ b/tools/build_uk_national_dataset.py @@ -776,13 +776,13 @@ def _aggregate_build_record( "sample_fraction": float(sample_fraction), "sample_seed": int(sample_seed), "rung_token": UK_SAMPLE_RUNG_TOKENS[sample_fraction], - # The attested policy digest is content-addressed, so a - # content-identical --degenerate-exclusions override would be - # invisible there; the record keeps the provenance honest - # without a path (this record is path-free by contract). - "degenerate_exclusions_register": ( - "override" if degenerate_exclusions_override else "committed" - ), + # Answers "did the operator invoke the override path" — the + # operator-action record, kept path-free by contract. The signed + # report's evidence answers the different question "which + # register content governed" (``exclusions_policy``); a review + # file byte-identical to the committed register makes the two + # honestly disagree, which is why they carry distinct names. + "degenerate_exclusions_override_supplied": (degenerate_exclusions_override), }, "sampling": ( None if result.sampling_receipt is None else dict(result.sampling_receipt)