UK parity increment 4: weighted integrity gates (input-mass parity + QRF tail concentration) in the terminal battery (#609)

changelog.d/609-uk-weighted-integrity-gates.added.md

@@ -0,0 +1,78 @@
+Ported the two incident-purchased US weighted-integrity gates into the UK
+terminal battery (#609, increment 4 of the #578 parity plan). The shared
+`input_mass_totals` helper was promoted from `populace.build.us_runtime` to
+`populace.build.input_mass` (the US name remains a re-export), and a new
+`uk_runtime.weighted_integrity` module supplies the UK evidence plumbing:
+`uk_dataset_input_mass_totals` broadcasts household weights through the
+national person/benunit/household tables, `uk_input_mass_parity_gate` keeps
+the #278 semantics verbatim (zero candidate mass fails at any tolerance,
+candidate-only columns are reported and never fail, near-zero reference
+columns are skipped) and records the frozen reference's filename, revision,
+sha256, and vintage, and `uk_qrf_tail_concentration_gate` derives its column
+surface from the `fit_weighted_qrf_stage*` outputs declared in the HMRC
+source manifest with no sparsity filter — every declared output is represented,
+absent or nonnumeric outputs fail by name, and `min_nonzero_records` classifies
+thin present outputs without allowing an armed zero-column gate to pass.
+
+Both gates join `uk_terminal_gate_report` under the optional-evidence rule: a
+path with no frozen reference or reviewed thresholds omits them instead of
+inventing passes, and an armed gate missing either fails closed by name.
+Thresholds carry no committed defaults — arming requires explicit
+`UKInputMassParityPolicy` / `UKQRFTailConcentrationPolicy` values, which are
+sealed into `policy_sha256`, alongside new `input_mass_parity` and
+`qrf_tail_concentration` evidence digests; the terminal-gate report schema
+moved to 3 and the attestation to 5, with the populace-data publication
+contract updated in lockstep. Empty reviewed-exclusion registers are
+committed under the universal discipline (mandatory reason, dormant entries
+reported, stale entries fail — added on top of the shared input-mass gate,
+which lacked staleness detection). The staging launcher grew the matching
+`--input-mass-*` and `--qrf-tail-*` flags, and the #609 measurement pass is
+now runnable: `tools/measure_uk_weighted_integrity_baselines.py` records
+weighted totals and top-k concentration for any national artifact, and
+`tools/build_uk_efrs_parity_reference.py --emit-weighted-totals` extracts the
+pinned eFRS incumbent's totals to a file outside the repository.
+
+The first measurement pass ran against the pinned enhanced-FRS incumbent and
+the certified compact, and its result is why both gates ship unarmed rather
+than with inherited constants. The US 0.75 top-share threshold fails the
+incumbent on 16 of 28 checked columns, and no single global threshold is both
+incumbent-compatible and able to catch a #462-scale incident. The US 1e9 mass
+floor would stop checking 50 of 131 columns, including the two the release
+input coverage manifest requires distributional effective mass for, so the
+adjudicated floor is 0.0. And the certified compact turns out not to be a
+valid candidate against the incumbent — 8.1% less household mass, a 22.3%
+median per-column drift, and no `hmrc_spi_*` columns, because it is the input
+to the stage that creates them — which confirms the issue's reading that both
+thresholds must come from a staged candidate. The findings are recorded beside
+the policy dataclasses so a later reader cannot reintroduce the US numbers by
+default.
+
+The measurement recorder is disclosure-controlled at the source, because its
+output exists to be posted: UKDS End User Licence CD137 v16.00 clause 8 binds
+published outputs to the standards in CD171-ResearchDataHandling §5.2.1, which
+requires that no output refer to unit records (naming maxima and minima) and
+that nothing be reported from one or two cases. So the recorder emits no
+maximum or minimum, suppresses concentration statistics and carrier counts for
+columns with fewer carriers than `--sdc-minimum-count` (default 10, the
+guide's secondary-disclosure advice; raise to 30 where a study's Special
+Conditions require it), refuses a `--top-k` narrower than that count, and
+records the rules it applied alongside the citation obligations under clauses
+11 and 12. Per-column weighted totals aggregate every carrier and are reported
+unconditionally.
+
+The #610 review hardened every output and publication seam around those gates.
+The legacy `--input-coverage-json` alias now runs through the signed terminal
+writer before its compatibility projection, so missing credentials persist an
+unsigned failed receipt and stop staging. Input-mass references must match the
+committed enhanced-FRS source identity and the reviewed canonical digest of
+all 131 weighted totals at load, gate, and publication time.
+Exclusion JSON now rejects duplicate keys, malformed shapes, nulls, and
+non-string names or reasons; thin exclusions are always classified as dormant.
+Publication independently rejects absent, nonnumeric, vacuous, partially
+omitted, or impossible threshold/share/carrier/thin-count QRF evidence and
+reconciles every declared output across the checked and thin maps. The launcher
+collision-checks the reference and both exclusion inputs against every source
+and output path before unlinking anything.
+Approval identity, receipt metadata, and expiry semantics for exclusions
+remain an author-owned follow-up rather than an invented schema change in this
+review.

packages/populace-build/src/populace/build/gates.py

@@ -2409,7 +2409,7 @@ def tail_concentration_gate(
             concentrated (a documented, tracked defect or a genuinely
             concentrated instrument). A column now below the threshold is a
             stale entry and fails; an entry for a column absent from this
-            surface is dormant and only reported.
+            surface or too thin to check is dormant and only reported.
 
     Returns:
         Pass iff every checked, non-excluded column's top-``top_k`` weighted
@@ -2486,7 +2486,9 @@ def tail_concentration_gate(
             f"concentration threshold now, remove the exclusion: "
             f"{sorted(stale_exclusions)}."
         )
-    dormant_exclusions = sorted(set(exclusions) - set(column_values))
+    dormant_exclusions = sorted(
+        set(exclusions) - set(used_exclusions) - set(stale_exclusions)
+    )
 
     return GateResult(
         name="tail_concentration",

packages/populace-build/src/populace/build/input_mass.py

@@ -0,0 +1,78 @@
+"""Weighted per-column input mass for populace frames.
+
+These totals feed :func:`populace.build.gates.input_mass_parity_gate`:
+comparing a derived artifact's persisted-input totals against its dense
+parent (or a certified reference release) catches input bases that a sparse
+selection or a rebuilt base pipeline silently zeroes — the failure mode of
+populace issue #278, where the sparse default release carried ~$0 in
+IRA-contribution, HSA, pension-contribution, and childcare inputs while
+hitting its own calibration target surface.
+
+The computation is schema-driven and country-agnostic: it reads only the
+frame's entity layout (id and membership columns) and each entity's resolved
+weights, so US and UK callers share one implementation instead of forking it.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import numpy as np
+import pandas as pd
+
+from populace.frame import Frame
+
+__all__ = ["input_mass_totals"]
+
+
+def input_mass_totals(
+    frame: Frame,
+    *,
+    columns: Iterable[str] | None = None,
+) -> dict[str, float]:
+    """Weighted totals of the frame's numeric and boolean value columns.
+
+    Every non-structural numeric column is summed under the owning entity's
+    effective weights (household weights broadcast through membership for
+    entities without their own vector); boolean columns total their weighted
+    ``True`` mass. String/enum columns and structural columns (entity ids and
+    person membership ids) are skipped.
+
+    Args:
+        frame: A populace frame in any country schema.
+        columns: Optional restriction — when given, only these columns are
+            totalled. Pass the engine's input-variable list on raw build
+            frames so source-survey scratch columns that never persist do not
+            enter the comparison.
+
+    Returns:
+        Column name -> weighted total. The mapping is flat because the frame
+        already enforces globally unique column names across entity tables.
+    """
+
+    schema = frame.schema
+    structural = {schema.person_id_column}
+    for group in schema.group_entities:
+        structural.add(schema.id_column(group))
+        structural.add(schema.membership_column(group))
+    requested = None if columns is None else {str(name) for name in columns}
+
+    totals: dict[str, float] = {}
+    for entity in frame.entities:
+        table = frame.table(entity)
+        weights = np.asarray(frame.resolve_weights(entity).values, dtype=np.float64)
+        for column in table.columns:
+            if column in structural:
+                continue
+            if requested is not None and column not in requested:
+                continue
+            values = table[column]
+            if pd.api.types.is_bool_dtype(values):
+                numeric = values.fillna(False).to_numpy(dtype=np.float64)
+            elif pd.api.types.is_numeric_dtype(values):
+                numeric = pd.to_numeric(values, errors="coerce")
+                numeric = numeric.fillna(0.0).to_numpy(dtype=np.float64)
+            else:
+                continue
+            totals[column] = float(numeric @ weights)
+    return totals

packages/populace-build/src/populace/build/uk/country_package.json

@@ -8,7 +8,9 @@
     "hmrc_income_release_gate_report.json",
     "hmrc_income_replay_report.json",
     "hmrc_income_source_stages.json",
+    "input_mass_reviewed_exclusions.json",
     "national_staging_build_record.json",
+    "qrf_tail_reviewed_exclusions.json",
     "release_input_coverage_manifest.json",
     "uk_local_target_census.json"
   ]

packages/populace-build/src/populace/build/uk/input_mass_reviewed_exclusions.json

@@ -0,0 +1,5 @@
+{
+ "schema_version": 1,
+ "description": "Reviewed input-mass parity exclusions for the UK terminal battery (#609). Each entry maps an 'entity.column' name to a non-empty reason documenting why that column is allowed to drift or disappear against the frozen reference. The gate reports dormant entries (columns outside the audited surface) and FAILS stale ones (columns now within tolerance), so this register cannot rot. Empty until an incident or adjudication earns an entry.",
+ "exclusions": {}
+}

packages/populace-build/src/populace/build/uk/qrf_tail_reviewed_exclusions.json

@@ -0,0 +1,5 @@
+{
+ "schema_version": 1,
+ "description": "Reviewed QRF tail-concentration exclusions for the UK terminal battery (#609). Each entry maps a declared fit_weighted_qrf_stage* output column to a non-empty reason naming the tracked defect or the genuinely concentrated instrument. The shared gate reports dormant entries (columns absent from the declared surface) and FAILS stale ones (columns now below the concentration threshold), so this register cannot rot. Empty until an incident or adjudication earns an entry.",
+ "exclusions": {}
+}

packages/populace-build/src/populace/build/uk_runtime/__init__.py

@@ -398,6 +398,17 @@
     uk_zero_weight_strata_gate,
     write_uk_terminal_gate_report,
 )
+from populace.build.uk_runtime.weighted_integrity import (
+    UKInputMassParityPolicy,
+    UKInputMassReference,
+    UKQRFTailConcentrationPolicy,
+    load_uk_input_mass_reference,
+    load_uk_reviewed_exclusion_register,
+    uk_dataset_input_mass_totals,
+    uk_input_mass_parity_gate,
+    uk_qrf_tail_concentration_columns,
+    uk_qrf_tail_concentration_gate,
+)
 
 __all__ = [
     "UK_CGT_ANNUAL_EXEMPT_AMOUNTS",
@@ -727,10 +738,19 @@
     "UK_MAX_TO_MEDIAN_WEIGHT_RATIO",
     "UK_MIN_ESS_FRACTION",
     "UK_TERMINAL_GATE_SCHEMA_VERSION",
+    "UKInputMassParityPolicy",
+    "UKInputMassReference",
+    "UKQRFTailConcentrationPolicy",
     "UKReleaseParityEvidence",
     "UKZeroWeightStratumDeclaration",
+    "load_uk_input_mass_reference",
+    "load_uk_reviewed_exclusion_register",
+    "uk_dataset_input_mass_totals",
     "uk_degenerate_release_surface_gate",
     "uk_export_surface_gate",
+    "uk_input_mass_parity_gate",
+    "uk_qrf_tail_concentration_columns",
+    "uk_qrf_tail_concentration_gate",
     "uk_target_fit_gate",
     "uk_target_surface_gate",
     "uk_terminal_gate_report",

packages/populace-build/src/populace/build/uk_runtime/hmrc_source_contract.py

@@ -65,6 +65,7 @@
     "HMRC_DISTRIBUTIONAL_INPUTS",
     "UK_HMRC_INCOME_SOURCE_STAGES_RESOURCE",
     "assert_uk_hmrc_income_source_contract_current",
+    "uk_hmrc_weighted_qrf_output_columns",
 ]
 
 UK_HMRC_INCOME_SOURCE_STAGES_RESOURCE = "hmrc_income_source_stages.json"
@@ -913,6 +914,62 @@ def assert_uk_hmrc_income_source_contract_current(
     _raise_failures(failures)
 
 
+def uk_hmrc_weighted_qrf_output_columns(
+    resource: Any | None = None,
+) -> tuple[str, ...]:
+    """Columns produced by declared ``fit_weighted_qrf_stage*`` operations.
+
+    This is the UK tail-concentration gate's column surface (#609). It is
+    derived from the declarative source manifest rather than a hand list —
+    mirroring the US derivation from ``us/source_stages.json`` — so a new
+    weighted-QRF output is covered by the gate the day the manifest declares
+    it. Declaration order is preserved (stage 1 before stage 2).
+    """
+
+    payload = _load_payload(resource)
+    stages = payload.get("stages")
+    if not isinstance(stages, Sequence) or isinstance(stages, (str, bytes)):
+        raise ValueError("UK HMRC source manifest must declare a stages list.")
+    outputs: dict[str, None] = {}
+    weighted_qrf_operations = 0
+    for stage in stages:
+        if not isinstance(stage, Mapping):
+            raise ValueError("UK HMRC source manifest stages must be objects.")
+        operations = stage.get("operations", ())
+        if not isinstance(operations, Sequence) or isinstance(operations, (str, bytes)):
+            raise ValueError("UK HMRC source manifest stage operations must be a list.")
+        for operation in operations:
+            if not isinstance(operation, Mapping):
+                raise ValueError("UK HMRC source manifest operations must be objects.")
+            kind = operation.get("kind")
+            if not isinstance(kind, str) or not kind.startswith("fit_weighted_qrf"):
+                continue
+            weighted_qrf_operations += 1
+            declared = operation.get("outputs")
+            if (
+                not isinstance(declared, Sequence)
+                or isinstance(declared, (str, bytes))
+                or not declared
+            ):
+                raise ValueError(
+                    f"UK HMRC source manifest operation {kind!r} must declare a "
+                    "non-empty outputs list."
+                )
+            for output in declared:
+                if not isinstance(output, str) or not output:
+                    raise ValueError(
+                        f"UK HMRC source manifest operation {kind!r} declares a "
+                        "non-string or empty output."
+                    )
+                outputs[output] = None
+    if not weighted_qrf_operations:
+        raise ValueError(
+            "UK HMRC source manifest declares no fit_weighted_qrf operations; "
+            "an empty tail-concentration surface would make the gate vacuous."
+        )
+    return tuple(outputs)
+
+
 def _load_payload(resource: Any | None) -> Mapping[str, Any]:
     target = (
         files("populace.build.uk").joinpath(UK_HMRC_INCOME_SOURCE_STAGES_RESOURCE)

packages/populace-build/src/populace/build/uk_runtime/national_build.py

@@ -26,6 +26,9 @@
     assert_uk_release_input_coverage_manifest_current,
 )
 from populace.build.uk_runtime.terminal_gates import (
+    UKInputMassParityPolicy,
+    UKInputMassReference,
+    UKQRFTailConcentrationPolicy,
     UKReleaseParityEvidence,
     uk_terminal_gate_report,
     write_uk_terminal_gate_report,
@@ -362,6 +365,9 @@ def build_uk_national_dataset(
     stages: Sequence[UKNationalStage] = (),
     coverage_engine: Any | None = None,
     parity_evidence: UKReleaseParityEvidence | None = None,
+    input_mass_reference: UKInputMassReference | None = None,
+    input_mass_policy: UKInputMassParityPolicy | None = None,
+    qrf_tail_policy: UKQRFTailConcentrationPolicy | None = None,
     terminal_gate_path: str | Path | None = None,
     input_coverage_path: str | Path | None = None,
 ) -> UKNationalBuildResult:
@@ -430,16 +436,18 @@ def build_uk_national_dataset(
         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,
     )
+    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"
         )
         _write_input_coverage_diagnostic(diagnostic_path, input_coverage)
-    else:
-        write_uk_terminal_gate_report(terminal_gates, diagnostic_path)
     if not terminal_gates.passed:
         raise RuntimeError(
             "Release gates failed: " + "; ".join(terminal_gates.failures)