Enforce the UK local binding fences at execution time (#760)

changelog.d/760-uk-local-binding-enforcement.added.md

@@ -0,0 +1 @@
+UK rowwise local binding now enforces the 7 reviewed census binding fences at execution time: the doctrine solve derives bound families structurally from the target surface, refuses declared-vs-derived mismatches, checks those families against the committed schema-2 `local_binding_adjudications.json` register with correct-or-renew approval windows, and echoes the adjudications in both the dry-run plan and candidate manifest; the first shipped adjudication accepts census disclosure-control noise for household counts and names constituency as the winning grain for the rowwise arm. Both expiring adjudication registers — the local binding register and the national calibration measure exclusions — now warn when an in-force entry is within one week of its expiry date, so renewal is signalled before the correct-or-renew failure lands.

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

@@ -227,6 +227,11 @@
       "kind": "legacy_json",
       "schema_id": "legacy_json"
     },
+    {
+      "path": "local_binding_adjudications.json",
+      "kind": "legacy_json",
+      "schema_id": "legacy_json"
+    },
     {
       "path": "uk_local_target_census.json",
       "kind": "legacy_json",

packages/microcosm-build/src/microcosm/build/uk/local_binding_adjudications.json

@@ -0,0 +1,13 @@
+{
+  "schema_version": 2,
+  "description": "Reviewed adjudications of the UK local binding fences (uk_local_target_census.json binding_fences). Binding a target family whose fence has no in-force entry here fails in the rowwise doctrine solve before any solve runs. Keys are fence ids; the reason states the ruling (accept the stated basis, or replace it with the named alternative) and cites the evidence.",
+  "exclusions": {
+    "census_disclosure_control_noise": {
+      "reason": "Accept the stated basis: constituency household counts bind to the published disclosure-controlled census values as-is — ONS/NRS cell-key perturbation and NISRA flexible-table-builder noise is documented, the counts are target values, never exact controls; the measured cross-grain deltas at review (constituency sums vs national totals: E&W +105, Scotland -554, NI +3) are accepted; constituency is the winning grain for the rowwise arm, whose targets are the sha-pinned OA ladder's occupied-household counts summed at constituency grain (ladder_targets.constituency_household_targets, provenance recorded per build) with no cross-grain reconciliation. Evidence: uk_local_target_census.json#/binding_fences/census_disclosure_control_noise.",
+      "approved_by": "juaristi22",
+      "adjudication": "microcosm#760",
+      "approved_on": "2026-08-27",
+      "expires_on": "2026-11-27"
+    }
+  }
+}

packages/microcosm-build/src/microcosm/build/uk/uk_local_target_census.json

@@ -542,6 +542,6 @@
     "bound_in_code": "The household-side metric is computed by local_targets.compute_household_metrics today.",
     "documented_unpinned": "The official product exists and was verified at the recorded URL on verified_on, but no exact table, vintage, or hash is pinned; binding work pins it per family.",
     "pinned_in_ladder": "The product is downloaded and sha-pinned per build by the UK OA ladder artifact tool, which records every source hash in the artifact metadata; target values derive from the artifact's own sums.",
-    "review_required_before_binding": "The fence is a reviewed adjudication requirement recorded for binding work; execution-time enforcement in the local solve configuration lands with the solve-doctrine increment of microcosm#495."
+    "review_required_before_binding": "The fence is a reviewed adjudication requirement enforced by the rowwise doctrine solve (local_rowwise.require_adjudicated_uk_local_binding): binding a family whose fences lack an in-force entry in local_binding_adjudications.json is refused before any solve runs."
   }
 }

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

@@ -264,10 +264,12 @@
     align_area_targets,
 )
 from microcosm.build.uk_runtime.local_rowwise import (
+    UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE,
     UKRowwiseDoctrineSolve,
     UKRowwiseLocalMatrix,
     build_uk_rowwise_local_matrix,
     past_cap_census,
+    require_adjudicated_uk_local_binding,
     rowwise_area_support_summary,
     rowwise_calibration_mass_reason,
     solve_uk_rowwise_weights_under_doctrine,
@@ -496,6 +498,7 @@
     "UK_GATE_REGISTRY",
     "UK_FRAME_METADATA_KEY",
     "UK_LOCAL_MAX_WEIGHT_RATIO",
+    "UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE",
     "UK_LOCAL_SOLVE_DOCTRINE",
     "UK_LOCAL_TARGET_LOSS_CAP",
     "UK_CGT_GAINS_AMOUNT_COLUMN",
@@ -506,6 +509,7 @@
     "load_uk_stage_checkpoint",
     "load_uk_stage_predecessor",
     "materialize_uk_cgt_calibration_frame",
+    "require_adjudicated_uk_local_binding",
     "rowwise_calibration_mass_reason",
     "uk_cgt_annual_exempt_amount",
     "UK_CGT_REQUIRED_COLUMNS",

packages/microcosm-build/src/microcosm/build/uk_runtime/local_rowwise.py

@@ -20,22 +20,31 @@
 
 from __future__ import annotations
 
+import warnings
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
+from datetime import date
 from typing import Any
 
 import numpy as np
 import pandas as pd
 from scipy import sparse as sp
 
+from microcosm.build.uk_runtime import local_target_census
 from microcosm.build.uk_runtime.local_doctrine import (
     UK_LOCAL_SOLVE_DOCTRINE,
 )
+from microcosm.build.uk_runtime.local_targets import AREA_TYPES
 from microcosm.build.uk_runtime.national_frame import (
     uk_national_frame,
     uk_time_period,
     validate_uk_national_frame,
 )
+from microcosm.build.uk_runtime.weighted_integrity import (
+    coerce_reviewed_exclusions,
+    exclusion_evaluation_date,
+    load_uk_reviewed_exclusion_register,
+)
 from microcosm.calibrate.solve import (
     CONSERVE_MASS,
     FREE_MASS,
@@ -47,15 +56,19 @@
 from microcosm.frame import Frame, WeightKind
 
 __all__ = [
+    "UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE",
     "UKRowwiseDoctrineSolve",
     "UKRowwiseLocalMatrix",
     "build_uk_rowwise_local_matrix",
     "past_cap_census",
+    "require_adjudicated_uk_local_binding",
     "rowwise_calibration_mass_reason",
     "rowwise_area_support_summary",
     "solve_uk_rowwise_weights_under_doctrine",
 ]
 
+UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE = "local_binding_adjudications.json"
+
 
 @dataclass(frozen=True)
 class UKRowwiseLocalMatrix:
@@ -106,6 +119,7 @@ class UKRowwiseDoctrineSolve:
     final_loss: float
     n_nonzero: int
     past_cap_census: Mapping[str, Any]
+    binding_adjudications: Mapping[str, Any]
 
 
 def past_cap_census(
@@ -403,6 +417,234 @@ def _require_uniform_target_surface(problem: UKRowwiseLocalMatrix) -> None:
         )
 
 
+def require_adjudicated_uk_local_binding(
+    bound_families: Sequence[str],
+    target_frame: pd.DataFrame,
+    *,
+    census: Mapping[str, Any] | None = None,
+    register: Mapping[str, Any] | None = None,
+    now: Any = None,
+) -> dict[str, Any]:
+    """Require in-force review records before binding fenced UK local families."""
+
+    census_payload = (
+        local_target_census.load_uk_local_target_census()
+        if census is None
+        else census
+    )
+    family_rows = _uk_local_census_family_rows(census_payload)
+    declared, parsed = _normalise_uk_local_bound_families(
+        bound_families,
+        family_rows=family_rows,
+    )
+    derived = _derive_uk_local_bound_families_from_target_frame(
+        target_frame,
+        family_rows=family_rows,
+    )
+    if sorted(declared) != derived:
+        missing = sorted(set(derived) - set(declared))
+        extra = sorted(set(declared) - set(derived))
+        raise ValueError(
+            "UK local binding declarations: declared bound families "
+            f"{sorted(declared)} disagree with the families derived from "
+            f"the target surface {derived}; missing {missing}, extra "
+            f"{extra}; the declaration must name exactly what the matrix "
+            "binds."
+        )
+
+    records = (
+        load_uk_reviewed_exclusion_register(
+            None,
+            resource=UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE,
+        )
+        if register is None
+        else coerce_reviewed_exclusions(
+            register,
+            label=UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE,
+        )
+    )
+    evaluated_on = exclusion_evaluation_date(now)
+    stood_on: dict[str, dict[str, dict[str, str]]] = {}
+    expiring_soon: dict[str, str] = {}
+    for declared_family in declared:
+        census_family, _area_type = parsed[declared_family]
+        stood_on[declared_family] = {}
+        for fence_id in family_rows[census_family].get("adjudications", ()):
+            fence = str(fence_id)
+            record = records.get(fence)
+            if record is None:
+                raise ValueError(
+                    f"{UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE}: "
+                    f"bound family {declared_family!r} requires an in-force "
+                    f"adjudication of fence {fence!r} and the committed "
+                    "register records none. Record a reviewed adjudication "
+                    "(reason stating the ruling and evidence, approver, "
+                    "window) before binding — the fence is a reviewed "
+                    "requirement, never a runtime flag."
+                )
+            if record.expired(evaluated_on) or record.premature(evaluated_on):
+                raise ValueError(
+                    f"{UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE}: "
+                    f"bound family {declared_family!r} requires adjudication "
+                    f"of fence {fence!r}, but the entry is outside its "
+                    f"reviewed window ({record.approved_on}.."
+                    f"{record.expires_on}, evaluated "
+                    f"{evaluated_on.isoformat()}): correct the underlying "
+                    "gap or renew the adjudication with a new approval and "
+                    "expiry."
+                )
+            if (date.fromisoformat(record.expires_on) - evaluated_on).days <= 7:
+                expiring_soon[fence] = record.expires_on
+            stood_on[declared_family][fence] = record.policy_payload()
+
+    for fence, expires_on in sorted(expiring_soon.items()):
+        warnings.warn(
+            f"{UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE}: adjudication "
+            f"of fence {fence!r} expires {expires_on} — within one week of "
+            f"the evaluation date {evaluated_on.isoformat()}; renew the "
+            "adjudication or the fence closes the rowwise solve on expiry.",
+            UserWarning,
+            stacklevel=2,
+        )
+
+    fence_families: dict[str, set[str]] = {}
+    for family, row in family_rows.items():
+        for fence_id in row.get("adjudications", ()):
+            fence_families.setdefault(str(fence_id), set()).add(family)
+    bound_census_families = {parsed[name][0] for name in declared}
+    dormant = sorted(
+        fence
+        for fence in records
+        if not (fence_families.get(fence, set()) & bound_census_families)
+    )
+    return {
+        "register_resource": UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE,
+        "evaluated_on": evaluated_on.isoformat(),
+        "bound_families": list(declared),
+        "stood_on": stood_on,
+        "dormant": dormant,
+    }
+
+
+def _uk_local_census_family_rows(
+    census: Mapping[str, Any],
+) -> dict[str, Mapping[str, Any]]:
+    rows = census.get("families")
+    if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)):
+        raise ValueError("UK local target census must carry a families list.")
+    families: dict[str, Mapping[str, Any]] = {}
+    for row in rows:
+        if not isinstance(row, Mapping):
+            raise ValueError("UK local target census family rows must be objects.")
+        family = str(row.get("family", ""))
+        if not family:
+            raise ValueError("UK local target census family row has no family id.")
+        if family in families:
+            raise ValueError(
+                f"UK local target census declares duplicate family {family!r}."
+            )
+        families[family] = row
+    return families
+
+
+def _normalise_uk_local_bound_families(
+    bound_families: Sequence[str],
+    *,
+    family_rows: Mapping[str, Mapping[str, Any]],
+) -> tuple[tuple[str, ...], dict[str, tuple[str, str]]]:
+    if isinstance(bound_families, str):
+        raise ValueError(
+            "UK local binding declarations: bound_families must be a "
+            "sequence of family/area_type strings, not one string."
+        )
+    declared = tuple(str(name) for name in bound_families)
+    if not declared:
+        raise ValueError(
+            "UK local binding declarations: bound_families must name at "
+            "least one family/area_type pair."
+        )
+    blanks = [name for name in declared if not name.strip()]
+    if blanks:
+        raise ValueError(
+            "UK local binding declarations: bound_families must not contain "
+            "blank family names."
+        )
+    padded = [name for name in declared if name != name.strip()]
+    if padded:
+        raise ValueError(
+            "UK local binding declarations: bound_families must not carry "
+            f"surrounding whitespace: {padded}."
+        )
+    duplicates = sorted({name for name in declared if declared.count(name) > 1})
+    if duplicates:
+        raise ValueError(
+            "UK local binding declarations: duplicate bound family(ies) "
+            f"{duplicates}."
+        )
+
+    parsed: dict[str, tuple[str, str]] = {}
+    for name in declared:
+        if "/" not in name:
+            raise ValueError(
+                "UK local binding declarations: bound family "
+                f"{name!r} must have form '<census_family>/<area_type>'."
+            )
+        family, area_type = name.rsplit("/", 1)
+        if not family or not area_type:
+            raise ValueError(
+                "UK local binding declarations: bound family "
+                f"{name!r} must have form '<census_family>/<area_type>'."
+            )
+        if area_type not in AREA_TYPES:
+            raise ValueError(
+                "UK local binding declarations: bound family "
+                f"{name!r} names unsupported area_type {area_type!r}; "
+                f"expected one of {list(AREA_TYPES)}."
+            )
+        if family not in family_rows:
+            raise ValueError(
+                "UK local binding declarations: bound family "
+                f"{name!r} names unknown census family {family!r}; "
+                f"expected one of {sorted(family_rows)}."
+            )
+        parsed[name] = (family, area_type)
+    return declared, parsed
+
+
+def _derive_uk_local_bound_families_from_target_frame(
+    target_frame: pd.DataFrame,
+    *,
+    family_rows: Mapping[str, Mapping[str, Any]],
+) -> list[str]:
+    required = {"area_type", "metric"}
+    if not required <= set(target_frame.columns):
+        missing = sorted(required - set(target_frame.columns))
+        raise ValueError(
+            "UK local binding declarations: target_frame column(s) "
+            f"{missing} are required to derive bound families."
+        )
+    derived: set[str] = set()
+    cells = target_frame[["area_type", "metric"]].drop_duplicates()
+    for row in cells.itertuples(index=False):
+        area_type = str(row.area_type)
+        if area_type not in AREA_TYPES:
+            raise ValueError(
+                "UK local binding declarations: target surface names "
+                f"unsupported area_type {area_type!r}; expected one of "
+                f"{list(AREA_TYPES)}."
+            )
+        metric = str(row.metric)
+        family = local_target_census.family_for_metric(metric)
+        if family not in family_rows:
+            raise ValueError(
+                "UK local binding declarations: target surface metric "
+                f"{metric!r} classified into unknown census family "
+                f"{family!r}."
+            )
+        derived.add(f"{family}/{area_type}")
+    return sorted(derived)
+
+
 def rowwise_calibration_mass_reason(bound_families: Sequence[str]) -> str:
     """The mass-record reason a rowwise doctrine calibration declares.
 
@@ -497,6 +739,10 @@ def solve_uk_rowwise_weights_under_doctrine(
 
     doctrine = UK_LOCAL_SOLVE_DOCTRINE
     _require_uniform_target_surface(problem)
+    binding_adjudications = require_adjudicated_uk_local_binding(
+        bound_families,
+        problem.target_frame,
+    )
     mass_reason = rowwise_calibration_mass_reason(bound_families)
 
     household = frame.table("household")
@@ -682,6 +928,7 @@ def solve_uk_rowwise_weights_under_doctrine(
         final_loss=float(result.final_loss),
         n_nonzero=int(result.n_nonzero),
         past_cap_census=census,
+        binding_adjudications=binding_adjudications,
     )