review: guard-only band classifier, backfill count, docstring (PR #1713)

Addresses reviewer feedback:
- #3 (Khalim + kimjunte): the boiler_efficiency_band classifier column is now
  GUARD-ONLY — LLM fallback removed. The band format is deterministic, so an LLM
  guess could only fabricate a band the landlord never stated (moves SAP +
  eligibility). New LoggingUnknownColumnClassifier is the non-fabricating fallback:
  maps guard-misses to UNKNOWN (never stored) and logs them for review — kimjunte's
  'warn on None', placed in the classifier path not the shared guard (which sees
  legitimate None constantly on the modelling/backfill paths).
- #2 (Khalim): backfill --apply now reports rows ACTUALLY written (upsert
  rowcount), not the candidate count, so an idempotent re-run reports 0.
- #1 (Khalim): overlay module docstring scrubbed of stale 'slot/pending' wording
  to match the cert-native anchor mechanism.

kimjunte's finaliser optional-skip question (declared-vs-shared-column flow) left
for reviewer alignment, not changed. 529 tests green; pyright clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

applications/landlord_description_overrides/handler.py

@@ -26,6 +26,9 @@ from domain.epc.property_overrides.roof_guard import roof_guard
 from domain.data_transformation.guarded_column_classifier import (
     GuardedColumnClassifier,
 )
+from domain.data_transformation.logging_unknown_column_classifier import (
+    LoggingUnknownColumnClassifier,
+)
 from domain.epc.property_overrides.water_heating_type import WaterHeatingType
 from domain.epc.property_overrides.water_heating_guard import water_heating_guard
 from domain.epc.property_overrides.wall_type import WallType
@@ -234,14 +237,17 @@ def _build_columns(
             source_column=src,
             # The SEDBUK band rides the SAME "Heating" source column as
             # main_heating_system (like Property Type feeds property_type +
-            # built_form_type). The deterministic guard extracts the structured
-            # `Boiler: <A-G> rated ...` band — authoritative for the format — and
-            # the LLM is the fallback for oddities, returning UNKNOWN (never
-            # stored) for a plain boiler or a non-boiler heating system (ADR-0068).
+            # built_form_type). The band's `Boiler: <A-G> rated ...` format is
+            # fully deterministic, so the guard is authoritative and the column is
+            # **guard-only** — NO LLM fallback. An LLM guess could fabricate a band
+            # the landlord never stated, which here moves SAP + funding eligibility
+            # (reviewer feedback). Unrecognised descriptions (a plain boiler, a
+            # non-boiler heating system) become UNKNOWN — never stored — and are
+            # logged for review (ADR-0068).
             classifier=GuardedColumnClassifier(
                 guard=boiler_efficiency_band_guard,
-                fallback=ChatGptColumnClassifier(
-                    chat_gpt, BoilerEfficiencyBand, BoilerEfficiencyBand.UNKNOWN
+                fallback=LoggingUnknownColumnClassifier(
+                    BoilerEfficiencyBand.UNKNOWN, "boiler_efficiency_band"
                 ),
             ),
             repo=LandlordOverridesRepository[BoilerEfficiencyBand](

domain/data_transformation/logging_unknown_column_classifier.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import logging
+from enum import Enum
+from typing import TypeVar
+
+from domain.data_transformation.column_classifier import ColumnClassifier
+
+logger = logging.getLogger(__name__)
+
+E = TypeVar("E", bound=Enum)
+
+# Cap the sample of unresolved descriptions logged, so a large batch surfaces the
+# surprise without flooding the log.
+_SAMPLE_SIZE = 10
+
+
+class LoggingUnknownColumnClassifier(ColumnClassifier[E]):
+    """A ``ColumnClassifier`` that resolves nothing — it maps every description to
+    the enum's ``unknown`` member and logs a warning naming the column and a sample
+    of what it saw.
+
+    Its purpose is to be a **deterministic, non-fabricating fallback** for a
+    ``GuardedColumnClassifier`` whose guard is authoritative for the column's
+    format. Where the LLM fallback would *guess* a category (and could invent a
+    value the landlord never stated — high-stakes for the Boiler Efficiency Band,
+    which moves SAP + funding eligibility), this fallback records ``unknown`` and
+    surfaces the misses for review instead. Guard hits never reach it, so only the
+    genuinely unrecognised descriptions are logged.
+    """
+
+    def __init__(self, unknown: E, column_name: str) -> None:
+        self._unknown = unknown
+        self._column_name = column_name
+
+    def classify(self, descriptions: set[str]) -> dict[str, E]:
+        if not descriptions:
+            return {}
+        sample = sorted(descriptions)[:_SAMPLE_SIZE]
+        more = "" if len(descriptions) <= _SAMPLE_SIZE else f" (+{len(descriptions) - _SAMPLE_SIZE} more)"
+        logger.warning(
+            "%s: %d description(s) not recognised by the guard -> recorded %r "
+            "(never fabricated, not persisted); sample: %s%s",
+            self._column_name,
+            len(descriptions),
+            self._unknown.value,
+            sample,
+            more,
+        )
+        return {description: self._unknown for description in descriptions}

domain/epc/property_overlays/main_heating_system_overlay.py

@@ -19,14 +19,18 @@ adding its code — coherent companions fall out. Synthesis owns coherence; the
 calculator never normalises a lodged cert.
 
 The SEDBUK A-G efficiency band the "Heating" column carries is honoured as the
-fifth Heating Companion (ADR-0068): a gas/oil boiler archetype still maps to its
-modern/condensing Table 4b code (which drives combi-vs-regular hot-water
-behaviour), but the `band` argument sets a `(winter, summer)` efficiency slot the
-calculator consumes ahead of that code's Table 4b default — so an old low-rated
-boiler is modelled at its band efficiency, not the condensing 84 %. C-and-below
-reuse accredited Table 4b values; A/B are pending Elmhurst validation (no slot
-until pinned). Solid-fuel / electric / CPSU boilers are intrinsic-efficiency and
-take no band. Heat pumps and community heating (which resolve via
+fifth Heating Companion (ADR-0068), via the boiler's **cert-native efficiency
+anchor** — no synthesised efficiency field. The `band` argument (see
+`band_efficiency_anchor`) picks:
+  - **C-and-below** → the band's Table 4b `sap_main_heating_code` (the code is
+    itself combi/regular-specific, so hot-water behaviour is preserved), so an
+    old low-rated boiler is modelled at its band efficiency, not the condensing
+    84 %;
+  - **A/B** (exceed Table 4b's 84 % ceiling) → a representative PCDB
+    `main_heating_index_number`; `_fold_heating` clears the base code, so the
+    effective cert reads as a real PCDB-lodged cert.
+Solid-fuel / electric / CPSU boilers are intrinsic-efficiency and take no band
+(the anchor is a no-op). Heat pumps and community heating (which resolve via
 main_heating_index_number / community codes, not a Table 4b code) are left
 UNKNOWN until modelled. Unresolvable values produce no overlay.
 """

scripts/backfill_boiler_efficiency_band.py

@@ -117,8 +117,13 @@ _UPSERT_BAND = text(
 
 def backfill(conn: Connection, *, apply: bool) -> int:
     """Backfill the ``boiler_efficiency_band`` rows off the existing
-    ``main_heating_system`` descriptions. Returns the number of band rows the run
-    writes (or would write, in dry-run)."""
+    ``main_heating_system`` descriptions.
+
+    In ``--apply`` returns the number of rows **actually written** — the upsert's
+    ``WHERE override_value <> EXCLUDED.override_value`` no-ops rows already at the
+    target band, so an idempotent re-run reports 0, not the candidate count. In
+    dry-run returns the number of **candidate** rows (it cannot know how many
+    differ without writing)."""
     rows = [
         MainHeatingRow(
             property_id=r.property_id,
@@ -129,21 +134,24 @@ def backfill(conn: Connection, *, apply: bool) -> int:
         for r in conn.execute(_SELECT_MAIN_HEATING)
     ]
     to_write = band_backfill_rows(rows)
-    if apply:
-        for band_row in to_write:
-            conn.execute(
-                _UPSERT_BAND,
-                {
-                    "property_id": band_row.property_id,
-                    "portfolio_id": band_row.portfolio_id,
-                    "building_part": band_row.building_part,
-                    "override_value": band_row.override_value,
-                    "original_spreadsheet_description": (
-                        band_row.original_spreadsheet_description
-                    ),
-                },
-            )
-    return len(to_write)
+    if not apply:
+        return len(to_write)
+    written = 0
+    for band_row in to_write:
+        result = conn.execute(
+            _UPSERT_BAND,
+            {
+                "property_id": band_row.property_id,
+                "portfolio_id": band_row.portfolio_id,
+                "building_part": band_row.building_part,
+                "override_value": band_row.override_value,
+                "original_spreadsheet_description": (
+                    band_row.original_spreadsheet_description
+                ),
+            },
+        )
+        written += result.rowcount or 0
+    return written
 
 
 def main() -> None:
@@ -161,13 +169,17 @@ def main() -> None:
         conn.execute(text("SET statement_timeout = 120000"))
         total = backfill(conn, apply=args.apply)
 
-    verb = "backfilled" if args.apply else "would backfill"
+    noun = "row(s) written" if args.apply else "candidate row(s)"
     print(
-        f"{verb} {total} boiler_efficiency_band override row(s) from the existing "
+        f"{total} boiler_efficiency_band {noun} from the existing "
         "main_heating_system descriptions (Boiler: <A-G> rated ...). The modelling "
         "gate (gas/oil boilers only) is applied at overlay time (ADR-0068)."
     )
-    if not args.apply:
+    if args.apply:
+        print(
+            "(Rows already at their target band are no-ops, so a re-run reports 0.)"
+        )
+    else:
         print("\nDRY-RUN — nothing written. Re-run with --apply to execute.")
 
 

tests/domain/data_transformation/test_logging_unknown_column_classifier.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import logging
+from enum import Enum
+from typing import Optional
+
+import pytest
+
+from domain.data_transformation.guarded_column_classifier import (
+    GuardedColumnClassifier,
+)
+from domain.data_transformation.logging_unknown_column_classifier import (
+    LoggingUnknownColumnClassifier,
+)
+
+
+class _Band(Enum):
+    A = "A"
+    D = "D"
+    UNKNOWN = "Unknown"
+
+
+def test_maps_every_description_to_unknown() -> None:
+    classifier = LoggingUnknownColumnClassifier(_Band.UNKNOWN, "boiler_efficiency_band")
+
+    assert classifier.classify({"x", "y"}) == {"x": _Band.UNKNOWN, "y": _Band.UNKNOWN}
+
+
+def test_logs_a_warning_naming_the_column_and_a_sample(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    classifier = LoggingUnknownColumnClassifier(_Band.UNKNOWN, "boiler_efficiency_band")
+
+    with caplog.at_level(logging.WARNING):
+        classifier.classify({"Community heating"})
+
+    assert "boiler_efficiency_band" in caplog.text
+    assert "Community heating" in caplog.text
+
+
+def test_empty_input_neither_logs_nor_returns_rows(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    classifier = LoggingUnknownColumnClassifier(_Band.UNKNOWN, "boiler_efficiency_band")
+
+    with caplog.at_level(logging.WARNING):
+        result = classifier.classify(set())
+
+    assert result == {}
+    assert caplog.records == []
+
+
+def _band_guard(description: str) -> Optional[_Band]:
+    # A tiny stand-in guard: recognises "A rated" / "D rated", else None.
+    for member in (_Band.A, _Band.D):
+        if f"{member.value} rated" in description:
+            return member
+    return None
+
+
+def test_guard_only_composition_never_reaches_an_llm(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # The band column is guard-only: recognised bands resolve deterministically,
+    # and unrecognised descriptions become UNKNOWN (logged) — never an LLM guess
+    # that could fabricate a band (reviewer feedback).
+    classifier = GuardedColumnClassifier(
+        guard=_band_guard,
+        fallback=LoggingUnknownColumnClassifier(_Band.UNKNOWN, "boiler_efficiency_band"),
+    )
+
+    with caplog.at_level(logging.WARNING):
+        result = classifier.classify({"Boiler: D rated Combi", "Gas boiler"})
+
+    assert result == {"Boiler: D rated Combi": _Band.D, "Gas boiler": _Band.UNKNOWN}
+    # Only the miss is logged; the guarded hit is not.
+    assert "Gas boiler" in caplog.text
+    assert "D rated" not in caplog.text