Fix #792: derive band edges from the compiled register, never the pruned roster

changelog.d/792-band-edges-compiled-register.fixed.md

@@ -0,0 +1 @@
+Banded calibration targets now derive upper edges from the compiled pre-exclusion register, fixing microcosm#792 so excluding a banded target no longer widens its surviving lower neighbour. The band-edge register is a required argument of the UK calibration stage and run seam — never defaulted, always reconciled against the materialized roster and the exclusion receipt (an empty receipt demands name-identical rosters) — and a register that cannot bound a spec refuses the run rather than letting the target drop out of the solve.

packages/microcosm-build/src/microcosm/build/target_materialization.py

@@ -21,9 +21,11 @@
 # encodings are in use — a numeric lower bound (HMRC SPI income bands) and a
 # published range label (DWP award bands, in monthly units, hence
 # ``band_period_factor``) — and both reduce to one lower edge, because no
-# reference anywhere declares an upper bound. A band's upper edge is
-# therefore its sibling's lower edge within the same contract target, and the
-# top band runs to infinity.
+# reference anywhere declares an upper bound. A band's upper edge is its
+# sibling's lower edge within the same compiled contract target. Edges must
+# never derive from an exclusion-pruned roster, or excluding a band silently
+# widens its lower neighbour; only the compiled register's top band runs to
+# infinity.
 # Entity-count indicators: a value of one per record of the owning entity.
 _COUNT_VALUE_VARIABLES = frozenset(
     {"household_count", "person_count", "benunit_count"}
@@ -73,6 +75,17 @@ class MeasureResolution:
     receipt: Mapping[str, Any]
 
 
+class BandEdgeCoverageError(RuntimeError):
+    """A supplied band-edge register does not cover a materialized spec.
+
+    Deliberately not a :class:`ValueError`: the per-spec materialization loop
+    converts ValueErrors into :class:`MaterializationSkip` entries, and a
+    register that cannot bound a spec is a wrong-register problem for the
+    whole run, not a per-spec data defect — it must refuse, never let the
+    target quietly drop out of the solve (#792 review finding 2).
+    """
+
+
 class MeasureResolutionError(RuntimeError):
     """Raised when simulated measure resolution cannot bind the registry."""
 
@@ -126,6 +139,7 @@ def resolve_target_measures(
     period: int | str,
     max_rounds: int = 8,
     contract_targets: Mapping[str, Mapping[str, Any]] | None = None,
+    band_edge_registry: TargetRegistry | None = None,
 ) -> MeasureResolution:
     """Resolve provider-computed inputs until target materialization binds.
 
@@ -159,6 +173,7 @@ def resolve_target_measures(
             registry,
             contract,
             period=period,
+            band_edge_registry=band_edge_registry,
         )
         skipped = tuple(result.skipped)
         round_receipt = {
@@ -414,6 +429,13 @@ def _band_bounds(
             f"groupby_variable {binding['groupby_variable']!r} but its spec "
             "carries no readable band edge"
         )
+    if lower not in band_edges:
+        raise BandEdgeCoverageError(
+            f"banded measure {getattr(spec, 'measure', '?')!r} carries band lower edge "
+            f"{lower!r} that is absent from its contract target's band-edge set "
+            f"{list(band_edges)!r}; the band-edge register does not cover this spec's "
+            "band — pass the compiled (pre-exclusion) register this spec was pruned from."
+        )
     upper = math.inf
     for edge in band_edges:
         if edge > lower:
@@ -429,11 +451,15 @@ def materialize_target_bindings(
     *,
     period: int | str,
     providers: Mapping[str, Provider] | None = None,
+    band_edge_registry: TargetRegistry | None = None,
 ) -> TargetMaterializationResult:
     """Prepare measure columns declared by compiled Ledger target specs."""
 
     provider_registry = {**default_provider_registry(), **(providers or {})}
-    band_edges = _band_edges_by_group(registry, contract_targets)
+    band_edges = _band_edges_by_group(
+        registry if band_edge_registry is None else band_edge_registry,
+        contract_targets,
+    )
     skipped: list[MaterializationSkip] = []
     for spec in registry.specs:
         if hasattr(adapter, "has_column") and adapter.has_column(

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

@@ -197,6 +197,7 @@ def run_uk_calibration(
     input_sha256: str,
     ledger_artifact: Any,
     register_registry: TargetRegistry,
+    band_edge_registry: TargetRegistry,
     calibration_year: int,
     exclusion_receipt: Mapping[str, Mapping[str, str]],
     doctrine: Any,
@@ -211,6 +212,15 @@ def run_uk_calibration(
 
     started_at = time.perf_counter()
     started_ts = datetime.now(UTC)
+    # Pure-argument validation precedes every environment probe: an
+    # incoherent register/receipt/band-edge triple must refuse identically
+    # whether or not a git checkout or Logbook chain is reachable.
+    _validate_band_edge_registry(
+        register_registry=register_registry,
+        band_edge_registry=band_edge_registry,
+        exclusion_receipt=exclusion_receipt,
+    )
+    edge_registry = band_edge_registry
     code_pin = git_code_pin(_REPOSITORY)
     # Predecessor configuration is validated before anything is written: a
     # disagreeing chain must refuse with no artifact on disk, not after a
@@ -230,6 +240,7 @@ def run_uk_calibration(
         "ledger": _ledger_provenance(ledger_artifact),
         **dict(run_config_extra),
     }
+    run_config["band_edge_register_sha256"] = edge_registry.version
     state = AttemptState(
         # Attempts are distinct rows even when they re-run one release: both
         # the local chain and the store refuse a repeated build id.
@@ -246,6 +257,7 @@ def run_uk_calibration(
             input_sha256=input_sha256,
             ledger_artifact=ledger_artifact,
             register_registry=register_registry,
+            band_edge_registry=edge_registry,
             calibration_year=calibration_year,
             exclusion_receipt=exclusion_receipt,
             doctrine=doctrine,
@@ -287,6 +299,45 @@ def _new_calibration_attempt_id(*, timestamp: datetime) -> str:
     )
 
 
+def _validate_band_edge_registry(
+    *,
+    register_registry: TargetRegistry,
+    band_edge_registry: TargetRegistry,
+    exclusion_receipt: Mapping[str, Mapping[str, str]],
+) -> None:
+    """Require the band-edge register to reconstitute the compiled roster.
+
+    The reconciliation always runs — an empty receipt is a claim that nothing
+    was pruned, so the two rosters must then be name-identical; it is never
+    permission to skip the check. Receipt keys are spec names by the
+    applier's construction: ``apply_uk_calibration_measure_exclusions``
+    builds the receipt from the matched ``spec.name`` set and raises on any
+    exclusion matching zero registry specs, so ``pruned + receipt keys ==
+    compiled`` is an exact identity, not a heuristic.
+    """
+
+    register_names = _registry_spec_names(register_registry)
+    edge_names = _registry_spec_names(band_edge_registry)
+    excluded_names = {str(name) for name in exclusion_receipt}
+    if not register_names <= edge_names:
+        missing = sorted(register_names - edge_names)
+        raise ValueError(
+            "band-edge register must include every materialized registry spec; "
+            f"missing={missing}."
+        )
+    extra = edge_names - register_names
+    if extra != excluded_names:
+        raise ValueError(
+            "band-edge register extra spec names must match the measure "
+            f"exclusion receipt; extra={sorted(extra)}, "
+            f"receipt={sorted(excluded_names)}."
+        )
+
+
+def _registry_spec_names(registry: TargetRegistry) -> set[str]:
+    return {str(spec.name) for spec in registry.specs}
+
+
 def _record_failed_attempt(
     *,
     error: BaseException,
@@ -336,6 +387,7 @@ def _run_uk_calibration_attempt(
     input_sha256: str,
     ledger_artifact: Any,
     register_registry: TargetRegistry,
+    band_edge_registry: TargetRegistry,
     calibration_year: int,
     exclusion_receipt: Mapping[str, Mapping[str, str]],
     doctrine: Any,
@@ -374,6 +426,7 @@ def _run_uk_calibration_attempt(
         period=calibration_year,
         doctrine=doctrine,
         measure_resolver=measure_resolver,
+        band_edge_registry=band_edge_registry,
     )
     calibrated = stage(frame)
     append_phase(state, "national_calibration_solved")

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

@@ -88,6 +88,7 @@ def materialize_uk_ledger_targets(
     registry: TargetRegistry,
     *,
     period: int | str,
+    band_edge_registry: TargetRegistry | None = None,
 ) -> TargetMaterializationResult:
     """Materialize compiled UK Ledger target bindings on an adapter."""
 
@@ -102,6 +103,7 @@ def materialize_uk_ledger_targets(
             "baseline_flag_crosstab": _uk_baseline_flag_crosstab,
             "input_substitution_counterfactual": _uk_input_substitution,
         },
+        band_edge_registry=band_edge_registry,
     )
 
 

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

@@ -48,13 +48,19 @@ def __init__(
         period: int,
         doctrine: UKNationalSolveDoctrine = UK_NATIONAL_SOLVE_DOCTRINE,
         measure_resolver: object | None = None,
+        band_edge_registry: TargetRegistry,
     ) -> None:
         self.compilation = (
             registry
             if isinstance(registry, UKLedgerTargetCompilation)
             else UKLedgerTargetCompilation(registry=registry, unsupported=())
         )
         self.registry = self.compilation.registry
+        # Required, never defaulted: the stage cannot tell a pruned registry
+        # from a full one, so a fallback to self.registry would quietly
+        # restate #792 for any caller holding an exclusion-pruned roster.
+        # A caller whose registry is unpruned passes it explicitly.
+        self.band_edge_registry = band_edge_registry
         # The materialization period is the declared calibration year the
         # registry was compiled at — never the input frame's base-year
         # time_period, which lags it (survey 2024, calibration 2025).
@@ -90,6 +96,7 @@ def __call__(self, frame: Frame) -> Frame:
             adapter,
             self.registry,
             period=self.period,
+            band_edge_registry=self.band_edge_registry,
         )
         if materialized.skipped:
             skipped = [skip.__dict__ for skip in materialized.skipped]
@@ -209,12 +216,14 @@ def _resolve_measures(self, frame: Frame) -> MeasureResolution | None:
                 lambda: _CalibrationFrameAdapter(frame),
                 self.registry,
                 period=self.period,
+                band_edge_registry=self.band_edge_registry,
             )
         return resolve_target_measures(
             lambda: _CalibrationFrameAdapter(frame),
             self.registry,
             self.measure_resolver,
             period=self.period,
+            band_edge_registry=self.band_edge_registry,
         )
 
     def checkpoint_metadata(self) -> Mapping[str, object]: