Merge branch 'main' into trigger-e2e-locally

.claude/skills/expand-sap-accuracy-corpus/SKILL.md

@@ -151,6 +151,14 @@ Pattern: `with E.session() as (ctx,page): E.goto(...); E.set_text/set_select(...
   2022" ≈ U1.4/g0.72) — the "...known data" option demands per-row Frame-Factor/
   Data-Source/Location/Building-Part and fails validation. Trust the grid only
   after Save & Close + reload (the live grid mis-shows older rows as 0.00).
+- **Window-width `.fill()` can DOUBLE the value** → "Openings: Width has to be less
+  or equal to 50" validation error. `_add_window`'s `page.locator(TextBoxExtWidth)
+  .fill(str(area))` went through Elmhurst's keystroke/numeric formatter and rendered
+  `11.1` as `111.10` (uprn_10010215568) — the Recommendations gate then blocks the
+  download. After building Openings, VERIFY the grid width (reload + read the row);
+  if wrong, JS-set it: `el.value='11.1'; dispatch input+change+blur`, then Save&Close
+  (the entry boxes mirror the single row = edit-in-place). JS-set bypasses the
+  formatter and sticks. Belt-and-braces: set the width by JS in `_add_window` too.
 - **U-value-known overrides** (walls/roof/floor) don't toggle via automation →
   accept the natural RdSAP age-band U-value; note the divergence for reconciliation.
 - **Recommendations page = validation gate.** It must show ZERO errors before the

.claude/skills/expand-sap-accuracy-corpus/worklist.md

@@ -78,7 +78,7 @@ Skip the 🚩 MVHR / 🚩 heat-pump-fuel and ⛔ sparse certs.
 - [⚠] 10093388044 — SAP-17.1 · eng 87 / lodged 93 · 🚩 heat-pump fuel-39 (flagged)
 - [x] 10090944225 — SAP-17.0 (full-SAP GROUND-FLOOR FLAT, band L 2015, combi PCDB 17045 Ideal Logic Combi ES35, control 2110 CBI time+temp zone, NATURAL vent + 2 extract fans, AP50 3.48, party wall 3.41m² U0, measured U walls 0.27/floor 0.14/window 1.41, TFA 49.97, no cylinder/no secondary) · eng 81 / elm 80 (lodged 82) · PINNED engine 81. Built in Elmhurst (boiler 17045 via search, control CBI/2110 under Boilers→Zone control, water from primary, AP50 Blower Door 3.48+cert, Single meter). engine-on-Elmhurst-inputs 80 = worksheet 80 (exact) → faithful; eng 81 ≈ lodged 82 (−1). Remaining +1 = documented full-SAP→RdSAP residual localised to the FLOOR (cert measured U 0.14 vs Elmhurst RdSAP solid default 0.23; walls/party match) — engine correctly uses measured 0.14, NOT a mapper bug. ⚠ Initial worksheet read 79 due to a spurious leftover REA/691 electric secondary (Table-11 frac 0.1, ~1.5 SAP) the shared assessment retained — REMOVED via present=Yes→clear secondary code to empty→present=No (79→80). 🔧 Build notes: storage→combi reset (clear leftover SEB MainHeatingCode pass-1, reset meter Single); control 2110 is Boilers→**Zone control**→CBI (not Standard). No mapper change.
 - [⚠] 10090341811 — SAP-17.0 · eng 80 / lodged 89 · 🚩 MVHR idx 500352 not credited (flagged)
-- [ ] 10010215568 — RdSAP-17.1 · eng 75 / lodged 74
+- [x] 10010215568 — RdSAP-17.1 (NATIVE; END-TERRACE BUNGALOW single-storey, band J 2003-2006, cavity as-built insulated, pitched 200mm loft, solid insulated floor, mains-gas REGULAR boiler PCDB 15100 Vaillant Ecotec plus 415 + cylinder size2/Normal-110L/foam 25mm/thermostat, control 2106 CBE, water from primary, party wall 9.8m, double glazed, natural vent, 86% LED, TFA 75.46) · eng 75 / elm 73 (lodged 74) · PINNED engine 75. Built in Elmhurst (House+1 storey bungalow, boiler 15100 via search, CBE/2106 control, cylinder Normal/foam/25mm/thermostat, water from primary, window 11.1m² double-unknown-date). engine-on-Elmhurst-inputs 74 ≈ worksheet 73 → calculator faithful; eng 75 ≈ lodged 74 (+1). +2 vs Elmhurst = documented engine-uses-lodged vs Elmhurst-RdSAP-default (wall thickness 250 default vs lodged 310, baths 2 vs lodged 1). Cylinder thickness 25mm correctly carried by RdSAP-17.1 mapper (no cylinder-gap). ⚠ window width fill bug: `.fill("11.1")` rendered 111.10 (keystroke formatter doubled it) → validation "Width ≤ 50"; fixed by JS-setting width. No mapper change. build_10010215568.py.
 - [⚠] 10093117227 — SAP-17.1 · eng 90 / lodged 80 · 🚩 heat-pump fuel-39 (flagged)
 - [⚠] 10023444170 — SAP-17.0 · eng 80 / lodged 83 · 🚩 MVHR idx 500167 not credited (flagged)
 - [ ] 100020980961 — SAP-16.3 · eng 66 / lodged 65

applications/modelling_e2e/handler.py

@@ -12,9 +12,10 @@ When no lodged EPC is found, EPC Prediction (Path 3, ADR-0031) synthesises one
 from the postcode cohort. ``_cohort_cache`` is module-level so warm Lambda
 containers re-processing the same postcode avoid redundant fetches.
 
-``secondary_heating_removal`` is excluded unconditionally: the live ``material``
-catalogue does not yet carry this measure type, causing a crash during catalogue
-reads for properties with a lodged secondary heater.
+All Measure Types are considered: pricing goes through
+``catalogue_with_off_catalogue_overrides`` so the measures the live ``material``
+catalogue cannot supply (``secondary_heating_removal``, the glazing and heating
+gaps) are priced from the committed off-catalogue overlay instead of crashing.
 
 DB engine is module-scoped so the connection pool is reused across warm
 invocations (ADR-0012).
@@ -45,7 +46,6 @@ from domain.epc_prediction.prediction_target import build_prediction_target
 from domain.geospatial.coordinates import Coordinates
 from domain.geospatial.planning_restrictions import PlanningRestrictions
 from domain.geospatial.spatial_reference import SpatialReference
-from domain.modelling.measure_type import MeasureType
 from domain.property.property import Property, PropertyIdentity
 from domain.tasks.tasks import Source
 from harness.console import run_modelling
@@ -67,7 +67,9 @@ from repositories.geospatial.geospatial_s3_repository import (
     ParquetReader,
 )
 from repositories.postgres_unit_of_work import PostgresUnitOfWork
-from repositories.product.product_postgres_repository import ProductPostgresRepository
+from repositories.product.composite_product_repository import (
+    catalogue_with_off_catalogue_overrides,
+)
 from repositories.property.landlord_override_overlays import overlays_from
 from repositories.property.override_backed_prediction_attributes_reader import (
     OverrideBackedPredictionAttributesReader,
@@ -212,7 +214,7 @@ def handler(body: dict[str, Any], context: Any) -> None:
     read_session = Session(engine)
     try:
         scenario = ScenarioPostgresRepository(read_session).get_many([scenario_id])[0]
-        products = ProductPostgresRepository(read_session)
+        products = catalogue_with_off_catalogue_overrides(read_session)
 
         errors: list[int] = []
 
@@ -282,19 +284,14 @@ def handler(body: dict[str, Any], context: Any) -> None:
                     None if no_solar else _solar_insights_for(solar_client, spatial)
                 )
 
-                # secondary_heating_removal is absent from the live material.type
-                # enum; exclude unconditionally until the catalogue gap is resolved.
-                considered: Optional[frozenset[MeasureType]] = (
-                    frozenset(MeasureType)
-                    - {MeasureType.SECONDARY_HEATING_REMOVAL}
-                    - {MeasureType.SYSTEM_TUNE_UP_ZONED}
-                )
-
+                # All Measure Types are considered: the off-catalogue overlay
+                # (catalogue_with_off_catalogue_overrides) prices the measures the
+                # live material catalogue cannot supply, so none need excluding.
                 plan = run_modelling(
                     effective_epc,
                     planning_restrictions=restrictions,
                     solar_insights=solar_insights,
-                    considered_measures=considered,
+                    considered_measures=None,
                     products=products,
                     scenario=scenario,
                     print_table=False,

datatypes/epc/domain/mapper.py

@@ -3296,6 +3296,16 @@ def _normalize_sap_schema_16_x(data: Dict[str, Any]) -> Dict[str, Any]:
     # still fail loud at from_dict — that is correct, the data is insufficient.
     d.setdefault("tenure", 0)
 
+    # Some 16.x certs (notably 16.3, cert 2628-…-2970) lodge `door_count` but
+    # omit `insulated_door_count` — RdSapSchema17_1 requires it. It is a door
+    # *refinement* (which of the lodged doors are the better-insulated type), not
+    # a core fabric field: an absent value means "no insulated doors recorded", so
+    # default 0 (the conservative RdSAP assumption — all doors take the standard
+    # door U-value; the engine's `insulated_share = insulated/door_count` is then
+    # 0). Mirrors the `tenure` default above. Certs that omit `door_count` itself
+    # still fail loud — that is correct, the fabric data is insufficient.
+    d.setdefault("insulated_door_count", 0)
+
     # 16.2 lodges glazing in BOTH `multiple_glazing_type` (frequently the "ND"
     # not-defined sentinel) AND the windows[].description. When the numeric field
     # is undefined, honour an explicit "Single glazed" description so it is not

datatypes/epc/domain/tests/test_from_sap_schema.py

@@ -519,6 +519,25 @@ class TestFromSapSchema16_2:
         assert epc.uprn == 100020933894
         assert Sap10Calculator().calculate(epc).sap_score == 61  # lodged 56
 
+    def test_16_x_missing_insulated_door_count_defaults_to_zero(self) -> None:
+        # Some 16.x certs lodge `door_count` but omit `insulated_door_count`,
+        # which RdSapSchema17_1 requires — previously raised "missing required
+        # field 'insulated_door_count'", aborting the whole prediction cohort
+        # (EpcComparablePropertiesRepository.candidates_for maps every cohort
+        # cert in one list comprehension, so one unmappable cert killed the
+        # predict_epc path). Regression for cert 2628-3046-6233-4584-2970
+        # (SAP-Schema-16.3, uprn 100061905751) seen via modelling_e2e property
+        # 730259. The normaliser defaults the missing count to 0 — no insulated
+        # doors recorded, the conservative RdSAP assumption.
+        data = load("sap_16_3.json")
+        del data["insulated_door_count"]
+        assert "insulated_door_count" not in data
+
+        epc = EpcPropertyDataMapper.from_api_response(data)
+
+        assert isinstance(epc, EpcPropertyData)
+        assert epc.insulated_door_count == 0
+
     def test_16_2_normalizer_does_not_mutate_caller_dict(self) -> None:
         # Mirror _normalize_shower_outlets' contract: the caller's dict is
         # untouched (deep copy), so a re-dispatch sees the original shape.

infrastructure/epc_client/epc_client_service.py

@@ -18,6 +18,8 @@ from datatypes.epc.search import EpcSearchResult
 
 class EpcClientService:
     BASE_URL = "https://api.get-energy-performance-data.communities.gov.uk"
+    # The gov API's per-UPRN search latency is variable: usually ~0.2s but with
+    # intermittent slow spells.
     REQUEST_TIMEOUT = 10.0
 
     def __init__(self, auth_token: str) -> None:

repositories/postgres_unit_of_work.py

@@ -11,8 +11,8 @@ from repositories.property_baseline.property_baseline_postgres_repository import
 )
 from repositories.epc.epc_postgres_repository import EpcPostgresRepository
 from repositories.plan.plan_postgres_repository import PlanPostgresRepository
-from repositories.product.product_postgres_repository import (
-    ProductPostgresRepository,
+from repositories.product.composite_product_repository import (
+    catalogue_with_off_catalogue_overrides,
 )
 from repositories.property.property_postgres_repository import (
     PropertyPostgresRepository,
@@ -49,7 +49,7 @@ class PostgresUnitOfWork(UnitOfWork):
         self.spatial = spatial_repo
         self.property_baseline = PropertyBaselinePostgresRepository(self._session)
         self.scenario = ScenarioPostgresRepository(self._session)
-        self.product = ProductPostgresRepository(self._session)
+        self.product = catalogue_with_off_catalogue_overrides(self._session)
         self.plan = PlanPostgresRepository(self._session)
         return self
 

repositories/product/composite_product_repository.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from sqlmodel import Session
+
+from domain.modelling.product import Product
+from repositories.product.product_json_repository import ProductJsonRepository
+from repositories.product.product_postgres_repository import (
+    ProductPostgresRepository,
+)
+from repositories.product.product_repository import ProductRepository
+
+# The committed off-catalogue cost overlay — Measure Types the live ``material``
+# catalogue cannot supply, priced from a static JSON until the ETL stocks them.
+OFF_CATALOGUE_COSTS_PATH: Path = Path(__file__).resolve().parent / "off_catalogue_costs.json"
+
+
+class CompositeProductRepository(ProductRepository):
+    """Resolves a Product from an ``override`` source first, then a ``fallback``.
+
+    The override holds off-catalogue costs the live ``material`` catalogue cannot
+    supply — today ``secondary_heating_removal``, whose Measure Type the FE-owned
+    ``material.type`` pgEnum does not carry. Querying that value against the DB
+    raises a ``DataError`` that poisons the session's transaction (it is not a
+    benign ``ProductNotFound``), so checking the override **first** keeps that
+    Measure Type away from the catalogue entirely. Every other Measure Type
+    misses the override and falls through to the catalogue unchanged.
+    """
+
+    def __init__(
+        self, override: ProductRepository, fallback: ProductRepository
+    ) -> None:
+        self._override = override
+        self._fallback = fallback
+
+    def get(self, measure_type: str) -> Product:
+        product = self._override.get_optional(measure_type)
+        if product is not None:
+            return product
+        return self._fallback.get(measure_type)
+
+
+def catalogue_with_off_catalogue_overrides(session: Session) -> ProductRepository:
+    """The live ``material`` catalogue with the committed off-catalogue JSON costs
+    layered on top — the single composition point both the pipeline's Unit of
+    Work and the local e2e runner price through."""
+    return CompositeProductRepository(
+        override=ProductJsonRepository(OFF_CATALOGUE_COSTS_PATH),
+        fallback=ProductPostgresRepository(session),
+    )

repositories/product/off_catalogue_costs.json

@@ -0,0 +1,8 @@
+{
+  "secondary_heating_removal": { "unit_cost_per_m2": 270.0 },
+  "double_glazing": { "unit_cost_per_m2": 550.0 },
+  "secondary_glazing": { "unit_cost_per_m2": 400.0 },
+  "gas_boiler_upgrade": { "unit_cost_per_m2": 3000.0 },
+  "system_tune_up": { "unit_cost_per_m2": 500.0 },
+  "system_tune_up_zoned": { "unit_cost_per_m2": 900.0 }
+}

scripts/hyde/elmhurst_download.py

@@ -28,7 +28,7 @@ SESSION_DIR = HERE / ".elmhurst-session"
 SAMPLE_DIR = (
     HERE.parent.parent
     / "backend/epc_api/json_samples/real_life_examples"
-    / "SAP-Schema-17.0/uprn_10090944225"
+    / "RdSAP-Schema-17.1/uprn_10010215568"
 )
 
 ASSESSMENT_GUID = "B44A0DB4-4C08-4241-B818-86F060172105"

scripts/run_modelling_e2e.py

@@ -9,7 +9,12 @@ from S3, and fetches Google Solar — then runs the Modelling stage (every
 Recommendation Generator → the Optimiser → a costed, attributed Plan). The same
 local computation runs whether or not you store the result: by default it
 persists **nothing** (the run is for inspecting recommendations); pass
-`--persist` to write the inputs + the Plan to the DB.
+`--persist` to write the inputs + the Plan to the DB. With `--persist`, a
+lodged-EPC Property **also** gets its Baseline Performance row written (lodged
+vs calculator-effective SAP + the bill block) via the same orchestrator the
+first-run pipeline uses — run per Property so one bad cert does not abort the
+batch. A predicted (EPC-less) Property has no lodged figures, so it gets a Plan
+but no baseline row.
 
 To keep the inspected recommendations identical to what gets stored, **both
 modes price against the live ``material`` catalogue (read-only)** and model
@@ -21,12 +26,9 @@ the run synthesises an Increasing-EPC-to-``--goal`` Scenario with no exclusions.
 `--measures` / `--exclude-measures` are optional overlays layered on top of the
 Scenario's own exclusions.
 
-KNOWN GOTCHA: the live ``material.type`` enum does not yet carry
-``secondary_heating_removal``, so a Property with a lodged secondary heater
-crashes the catalogue read for that measure. Until the catalogue stocks it, pass
-``--exclude-measures secondary_heating_removal`` (an ASHP row is also absent, but
-ASHP is priced off the rate sheet so it degrades to ``material_id=None`` rather
-than crashing — no flag needed).
+``secondary_heating_removal`` is priced from the committed off-catalogue JSON
+overlay (the live ``material.type`` enum cannot carry it), so no exclusion flag
+is needed. ASHP is priced off the rate sheet (``material_id=None``), also fine.
 
 Config: loads `backend/.env` for the DB creds (`DB_*`), the EPC API token
 (`OPEN_EPC_API_TOKEN` — the Bearer token for the new gov API), the Google Solar
@@ -104,13 +106,24 @@ from repositories.comparable_properties.epc_comparable_properties_repository imp
 from repositories.geospatial.geospatial_s3_repository import (  # noqa: E402
     GeospatialS3Repository,
 )
+from repositories.product.composite_product_repository import (  # noqa: E402
+    catalogue_with_off_catalogue_overrides,
+)
+from repositories.product.product_repository import ProductRepository  # noqa: E402
 from repositories.property.override_backed_prediction_attributes_reader import (  # noqa: E402
     OverrideBackedPredictionAttributesReader,
 )
-from repositories.product.product_postgres_repository import (  # noqa: E402
-    ProductPostgresRepository,
-)
 from repositories.postgres_unit_of_work import PostgresUnitOfWork  # noqa: E402
+from orchestration.property_baseline_orchestrator import (  # noqa: E402
+    PropertyBaselineOrchestrator,
+)
+from domain.property_baseline.calculator_rebaseliner import (  # noqa: E402
+    CalculatorRebaseliner,
+)
+from domain.sap10_calculator.calculator import Sap10Calculator  # noqa: E402
+from repositories.fuel_rates.fuel_rates_static_file_repository import (  # noqa: E402
+    FuelRatesStaticFileRepository,
+)
 from repositories.scenario.scenario_postgres_repository import (  # noqa: E402
     ScenarioPostgresRepository,
 )
@@ -418,6 +431,161 @@ def _predict_epc(
     return predicted
 
 
+def _run_from_db(
+    args: argparse.Namespace,
+    *,
+    engine: Engine,
+    products: ProductRepository,
+    scenario: Optional[Scenario],
+    considered: Optional[frozenset[MeasureType]],
+    baseline_orchestrator: Optional[PropertyBaselineOrchestrator],
+    md_path: Path,
+    csv_path: Path,
+    candidates_path: Path,
+    target: str,
+    measures_note: str,
+) -> None:
+    """Re-model from already-persisted inputs — **zero gov-API calls**.
+
+    Reads each Property's Effective EPC (lodged-or-predicted, overrides folded),
+    planning protections and solar straight from the DB (a prior ``--persist``
+    ingestion must have stored them), runs the same modelling, and — with
+    ``--persist`` — re-writes the Plan and, for lodged-EPC Properties, the
+    Baseline. A predicted Property has no lodged figures, so it gets no baseline
+    row (same rule as the live path). One bad property is logged and skipped.
+    """
+    md_lines: list[str] = [f"# Modelling recommendations ({target}, {measures_note})\n"]
+    csv_rows: list[str] = [
+        "property_id,uprn,api_sap,baseline_sap,sap_delta,post_sap,measures,"
+        "measure_types,cost_of_works"
+    ]
+    candidate_csv_rows: list[str] = [
+        "property_id,uprn,surface,measure_type,cost_total,contingency_rate,"
+        "selected,description"
+    ]
+    total = len(args.property_ids)
+    run_start = time.monotonic()
+    errors = 0
+    for index, property_id in enumerate(args.property_ids, start=1):
+        elapsed = time.monotonic() - run_start
+        eta = (elapsed / (index - 1)) * (total - index + 1) if index > 1 else 0.0
+        print(
+            f"[{index}/{total}] · {errors} err · elapsed {elapsed / 60:.1f}m "
+            f"· ETA {eta / 60:.1f}m · property {property_id} (from DB)",
+            flush=True,
+        )
+        try:
+            with PostgresUnitOfWork(lambda: Session(engine)) as uow:
+                prop = uow.property.get(property_id)
+                effective_epc: EpcPropertyData = prop.effective_epc
+                restrictions: PlanningRestrictions = prop.planning_restrictions
+                uprn: Optional[int] = prop.identity.uprn
+                epc: Optional[EpcPropertyData] = prop.epc
+                solar_insights: Optional[dict[str, Any]] = (
+                    uow.solar.get(uprn) if uprn is not None else None
+                )
+            predicted = epc is None
+            plan: Plan = run_modelling(
+                effective_epc,
+                goal_band=args.goal,
+                planning_restrictions=restrictions,
+                solar_insights=solar_insights,
+                considered_measures=considered,
+                products=products,
+                scenario=scenario,
+                print_table=False,
+            )
+            candidates: list[Recommendation] = candidate_recommendations(
+                epc if epc is not None else effective_epc,
+                planning_restrictions=restrictions,
+                solar_insights=solar_insights,
+                considered_measures=considered,
+                products=products,
+            )
+            if args.persist:
+                assert scenario is not None  # guaranteed by the --persist guard
+                with PostgresUnitOfWork(lambda: Session(engine)) as uow:
+                    uow.plan.save(
+                        plan,
+                        property_id=property_id,
+                        scenario_id=scenario.id,
+                        portfolio_id=args.portfolio_id,
+                        is_default=scenario.is_default,
+                    )
+                    uow.property.mark_modelled(
+                        property_id, has_recommendations=bool(plan.measures)
+                    )
+                    uow.commit()
+                # Lodged EPC also gets its Baseline Performance re-established from
+                # the persisted EPC; predicted Properties have no lodged figures.
+                if epc is not None:
+                    assert baseline_orchestrator is not None
+                    baseline_orchestrator.run([property_id])
+        except Exception as error:  # noqa: BLE001 — one bad property must not stop the run
+            errors += 1
+            line = f"property {property_id}: ERROR — {type(error).__name__}: {error}"
+            print(line + "\n")
+            md_lines.append(f"## Property {property_id}\n\n`{line}`\n")
+            csv_rows.append(f"{property_id},,,,,,,ERROR,")
+            continue
+
+        measure_types = [m.measure_type for m in plan.measures]
+        selected: set[MeasureType] = {m.measure_type for m in plan.measures}
+        flags = [
+            name
+            for name, on in (
+                ("conservation", restrictions.in_conservation_area),
+                ("listed", restrictions.is_listed),
+                ("heritage", restrictions.is_heritage),
+            )
+            if on
+        ]
+        context = (
+            f"{', '.join(flags) if flags else 'unrestricted'}; "
+            f"{'solar ✓' if solar_insights is not None else 'no solar'}"
+        )
+        source_tag = " · ⚠ PREDICTED (no lodged EPC)" if predicted else ""
+        candidate_lines = _candidate_lines(candidates, selected)
+        print(
+            f"=== Property {property_id} (uprn {uprn}) === "
+            f"SAP {plan.baseline.sap_continuous:.1f} -> {plan.post_sap_continuous:.1f} "
+            f"· {len(plan.measures)} measure(s) · £{plan.cost_of_works:,.0f} "
+            f"· {context}{source_tag}"
+        )
+        print(format_plan_table(plan))
+        md_lines.append(f"## Property {property_id} (uprn {uprn}){source_tag}\n")
+        md_lines.append(
+            f"SAP {plan.baseline.sap_continuous:.1f} → {plan.post_sap_continuous:.1f} "
+            f"· {len(plan.measures)} measure(s) · cost £{plan.cost_of_works:,.0f} "
+            f"· {context}\n"
+        )
+        md_lines.append("**Selected Plan**\n")
+        md_lines.extend(_measure_summary(m) for m in plan.measures)
+        md_lines.append("")
+        md_lines.append("**All candidate measures (cost per measure)**\n")
+        md_lines.extend(candidate_lines)
+        md_lines.append("")
+        api_sap: Optional[int] = epc.energy_rating_current if epc is not None else None
+        calc_sap: float = plan.baseline.sap_continuous
+        api_cell = "" if api_sap is None else str(api_sap)
+        delta_cell = "" if api_sap is None else f"{calc_sap - api_sap:.2f}"
+        csv_rows.append(
+            f"{property_id},{uprn},{api_cell},{calc_sap:.2f},{delta_cell},"
+            f"{plan.post_sap_continuous:.2f},{len(plan.measures)},"
+            f"{'|'.join(measure_types)},{plan.cost_of_works:.0f}"
+        )
+        candidate_csv_rows.extend(
+            _candidate_csv_rows(property_id, uprn, candidates, selected)
+        )
+
+    md_path.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
+    csv_path.write_text("\n".join(csv_rows) + "\n", encoding="utf-8")
+    candidates_path.write_text("\n".join(candidate_csv_rows) + "\n", encoding="utf-8")
+    print(f"wrote {md_path.resolve()}")
+    print(f"wrote {csv_path.resolve()}")
+    print(f"wrote {candidates_path.resolve()}")
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -466,6 +634,16 @@ def main() -> None:
         "(parent dirs created) instead of ./modelling_e2e.*; lets batched runs "
         "keep separate, durable output files",
     )
+    parser.add_argument(
+        "--from-db",
+        action="store_true",
+        default=False,
+        help="re-model from already-persisted inputs: read each Property's "
+        "Effective EPC + planning protections + solar from the DB and skip the "
+        "live EPC/spatial/solar fetch entirely (zero gov-API calls). Requires a "
+        "prior --persist ingestion run; with --persist it re-writes the Plan "
+        "(and Baseline for lodged-EPC Properties) without re-fetching.",
+    )
     args = parser.parse_args()
 
     if args.persist and (args.scenario_id is None or args.portfolio_id is None):
@@ -519,7 +697,19 @@ def main() -> None:
     # One read-only session for the live `material` catalogue, reused across the
     # batch so both store and no-store runs price against the same DB rows.
     catalogue_session = Session(engine)
-    products = ProductPostgresRepository(catalogue_session)
+    products = catalogue_with_off_catalogue_overrides(catalogue_session)
+    # When persisting, a lodged-EPC Property also gets a Baseline Performance row
+    # via the production orchestrator (lodged-vs-calculator SAP, the bill block) —
+    # the same establish-and-persist the first-run pipeline runs, here per Property
+    # so one bad cert doesn't abort the batch. Predicted Properties have no lodged
+    # figures, so they are skipped below.
+    baseline_orchestrator: Optional[PropertyBaselineOrchestrator] = None
+    if args.persist:
+        baseline_orchestrator = PropertyBaselineOrchestrator(
+            unit_of_work=lambda: PostgresUnitOfWork(lambda: Session(engine)),
+            rebaseliner=CalculatorRebaseliner(Sap10Calculator()),
+            fuel_rates=FuelRatesStaticFileRepository(),
+        )
     scenario: Optional[Scenario] = (
         _scenario_for(catalogue_session, args.scenario_id)
         if args.scenario_id is not None
@@ -539,14 +729,35 @@ def main() -> None:
     )
     measures_note = ",".join(sorted(considered)) if considered else "all measures"
     mode = "PERSISTING to DB" if args.persist else "no DB writes"
+    source = "persisted DB inputs" if args.from_db else "live EPC/solar"
     print(
         f"modelling {len(args.property_ids)} propertie(s) · {target} · {measures_note} · "
-        f"{mode} (DB material catalogue, live EPC/solar)...\n"
+        f"{mode} (DB material catalogue, {source})...\n"
     )
 
+    if args.from_db:
+        # Read inputs from the DB and skip every live fetcher (no gov-API calls).
+        # Self-contained loop + file writing; the live path below is left as-is.
+        _run_from_db(
+            args,
+            engine=engine,
+            products=products,
+            scenario=scenario,
+            considered=considered,
+            baseline_orchestrator=baseline_orchestrator,
+            md_path=md_path,
+            csv_path=csv_path,
+            candidates_path=candidates_path,
+            target=target,
+            measures_note=measures_note,
+        )
+        catalogue_session.close()
+        return
+
     md_lines: list[str] = [f"# Modelling recommendations ({target}, {measures_note})\n"]
     csv_rows: list[str] = [
-        "property_id,uprn,baseline_sap,post_sap,measures,measure_types,cost_of_works"
+        "property_id,uprn,api_sap,baseline_sap,sap_delta,post_sap,measures,"
+        "measure_types,cost_of_works"
     ]
     candidate_csv_rows: list[str] = [
         "property_id,uprn,surface,measure_type,cost_total,contingency_rate,"
@@ -690,6 +901,12 @@ def main() -> None:
                     solar_insights=solar_insights,
                     plan=plan,
                 )
+                # A lodged EPC also gets its Baseline Performance persisted
+                # (reads the EPC just saved above). Predicted Properties have no
+                # lodged figures to baseline, so they are skipped.
+                if epc is not None:
+                    assert baseline_orchestrator is not None
+                    baseline_orchestrator.run([property_id])
         except (
             Exception
         ) as error:  # noqa: BLE001 — one bad property must not stop the run
@@ -702,7 +919,7 @@ def main() -> None:
             line = f"property {property_id} (uprn {uprn}): ERROR — {type(error).__name__}: {error}"
             print(line + "\n")
             md_lines.append(f"## Property {property_id}\n\n`{line}`\n")
-            csv_rows.append(f"{property_id},{uprn or ''},,,,ERROR,")
+            csv_rows.append(f"{property_id},{uprn or ''},,,,,,ERROR,")
             continue
 
         measure_types = [m.measure_type for m in plan.measures]
@@ -737,8 +954,16 @@ def main() -> None:
         md_lines.append("**All candidate measures (cost per measure)**\n")
         md_lines.extend(candidate_lines)
         md_lines.append("")
+        # api_sap is the lodged/register SAP (off the cert); a predicted Property
+        # has none, so it and the delta are left blank. baseline_sap is the
+        # calculator's score on the Effective EPC — the two whose divergence the
+        # run is for reviewing (mirrors lodged vs effective in the baseline table).
+        api_sap: Optional[int] = epc.energy_rating_current if epc is not None else None
+        calc_sap: float = plan.baseline.sap_continuous
+        api_sap_cell = "" if api_sap is None else str(api_sap)
+        sap_delta_cell = "" if api_sap is None else f"{calc_sap - api_sap:.2f}"
         csv_rows.append(
-            f"{property_id},{uprn},{plan.baseline.sap_continuous:.2f},"
+            f"{property_id},{uprn},{api_sap_cell},{calc_sap:.2f},{sap_delta_cell},"
             f"{plan.post_sap_continuous:.2f},{len(plan.measures)},"
             f"{'|'.join(measure_types)},{plan.cost_of_works:.0f}"
         )

tests/applications/modelling_e2e/test_handler.py

@@ -167,7 +167,7 @@ def test_lodged_epc_path_saves_epc_plan_and_marks_modelled() -> None:
             patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
         ).return_value.get_many.return_value = [MagicMock()]
         stack.enter_context(
-            patch("applications.modelling_e2e.handler.ProductPostgresRepository")
+            patch("applications.modelling_e2e.handler.catalogue_with_off_catalogue_overrides")
         )
         stack.enter_context(
             patch("applications.modelling_e2e.handler.Session")
@@ -285,7 +285,7 @@ def test_prediction_path_saves_plan_without_epc_save() -> None:
             patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
         ).return_value.get_many.return_value = [MagicMock()]
         stack.enter_context(
-            patch("applications.modelling_e2e.handler.ProductPostgresRepository")
+            patch("applications.modelling_e2e.handler.catalogue_with_off_catalogue_overrides")
         )
         stack.enter_context(
             patch("applications.modelling_e2e.handler.Session")
@@ -381,7 +381,7 @@ def test_empty_cohort_gates_property_out_and_raises() -> None:
             patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
         ).return_value.get_many.return_value = [MagicMock()]
         stack.enter_context(
-            patch("applications.modelling_e2e.handler.ProductPostgresRepository")
+            patch("applications.modelling_e2e.handler.catalogue_with_off_catalogue_overrides")
         )
         stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
         MockUoW = stack.enter_context(
@@ -462,7 +462,7 @@ def test_partial_batch_failure_raises_runtime_error_listing_failed_ids() -> None
             patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
         ).return_value.get_many.return_value = [MagicMock()]
         stack.enter_context(
-            patch("applications.modelling_e2e.handler.ProductPostgresRepository")
+            patch("applications.modelling_e2e.handler.catalogue_with_off_catalogue_overrides")
         )
         stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
         stack.enter_context(
@@ -576,7 +576,7 @@ def test_cohort_cache_prevents_duplicate_candidates_for_calls() -> None:
             patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
         ).return_value.get_many.return_value = [MagicMock()]
         stack.enter_context(
-            patch("applications.modelling_e2e.handler.ProductPostgresRepository")
+            patch("applications.modelling_e2e.handler.catalogue_with_off_catalogue_overrides")
         )
         stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
         stack.enter_context(
@@ -659,7 +659,7 @@ def test_dry_run_skips_all_db_writes() -> None:
             patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
         ).return_value.get_many.return_value = [MagicMock()]
         stack.enter_context(
-            patch("applications.modelling_e2e.handler.ProductPostgresRepository")
+            patch("applications.modelling_e2e.handler.catalogue_with_off_catalogue_overrides")
         )
         stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
         stack.enter_context(

tests/domain/sap10_calculator/test_real_cert_sap_accuracy.py

@@ -557,6 +557,30 @@ _EXPECTATIONS: Final[tuple[RealCertExpectation, ...]] = (
         cert_num="2298-4036-7306-3186-4930",
         sap_score=81,
     ),
+    # UPRN 10010215568 → cert 9978-7098-7226-2633-3994. RdSAP-Schema-17.1 — NATIVE
+    # RdSAP, END-TERRACE BUNGALOW (single-storey), band J (2003-2006), cavity wall
+    # as-built insulated (assumed), pitched 200 mm loft, solid insulated floor,
+    # mains-gas REGULAR boiler (PCDB 15100 Vaillant Ecotec plus 415) + cylinder
+    # (size 2 = Normal/110 L, factory foam 25 mm, thermostat), control 2106 (CBE
+    # programmer + room thermostat + TRVs), water from primary, party wall 9.8 m,
+    # double glazed, natural ventilation, 86% LED, TFA 75.46. Lodged 74; engine 75.
+    # Built in Elmhurst RdSAP10 on the lodged inputs (evidence saved:
+    # elmhurst_summary.pdf / elmhurst_worksheet.pdf): worksheet SAP 73. Engine on
+    # Elmhurst's own parsed inputs = 74 ≈ worksheet 73 → calculator faithful (~1).
+    # The cylinder insulation thickness (25 mm) IS carried by the RdSAP-17.1 mapper
+    # (cylinder_insulation_thickness_mm=25) — no mapper gap here (cf. the RdSAP-17.0
+    # cylinder-thickness drop on uprn_68151071). The residual engine 75 vs Elmhurst
+    # 73 (+2) is the documented engine-uses-lodged vs Elmhurst-RdSAP-default gap:
+    # Elmhurst defaulted the cavity wall thickness to 250 mm (lodged 310 mm) and
+    # number of baths to 2 (lodged 1, lifting Elmhurst's HW demand). Engine 75 ≈
+    # lodged 74 (+1) is the validated anchor. PINNED to the observed engine 75 —
+    # mapping deliberately untuned.
+    RealCertExpectation(
+        schema="RdSAP-Schema-17.1",
+        sample="uprn_10010215568",
+        cert_num="9978-7098-7226-2633-3994",
+        sap_score=75,
+    ),
 )
 
 

tests/orchestration/test_ara_first_run_pipeline_integration.py

@@ -192,14 +192,8 @@ def test_first_run_baselines_through_repos_and_is_idempotent_on_rerun(
                     is_active=True,
                     description="Zoned heating controls + cylinder tune-up",
                 ),
-                MaterialRow(
-                    id=9,
-                    type="secondary_heating_removal",
-                    total_cost=250.0,
-                    cost_unit="gbp_per_unit",
-                    is_active=True,
-                    description="Secondary heating removal",
-                ),
+                # secondary_heating_removal is off-catalogue (priced from the
+                # JSON overlay, not the pgEnum-constrained material table).
             ]
         )
         session.commit()
@@ -324,14 +318,10 @@ def test_modelling_optimises_and_persists_a_multi_measure_plan(
                     is_active=True,
                     description="LED bulb",
                 ),
-                MaterialRow(
-                    id=9,
-                    type="secondary_heating_removal",
-                    total_cost=250.0,
-                    cost_unit="gbp_per_unit",
-                    is_active=True,
-                    description="Secondary heating removal",
-                ),
+                # No secondary_heating_removal row: the FE-owned ``material.type``
+                # pgEnum cannot carry that Measure Type, so it is priced from the
+                # committed off-catalogue JSON overlay (£270, no material_id) that
+                # the Unit of Work layers over the catalogue, not from the DB.
             ]
         )
         session.commit()
@@ -402,13 +392,17 @@ def test_modelling_optimises_and_persists_a_multi_measure_plan(
         "air_source_heat_pump",
         "secondary_heating_removal",
     }
-    # Each persisted measure carries the catalogue id of the Product it installs
+    # Each catalogue-sourced measure carries the id of the Product it installs
     # (the MaterialRow ids seeded above), replacing the retired
     # recommendation_materials BOM with a single material_id on the row.
     assert by_type["air_source_heat_pump"].material_id == 5
     assert by_type["suspended_floor_insulation"].material_id == 2
     assert by_type["low_energy_lighting"].material_id == 4
-    assert by_type["secondary_heating_removal"].material_id == 9
+    # Secondary heating removal is priced from the off-catalogue JSON overlay
+    # (£270 flat per-dwelling, ADR-0028), so it carries no catalogue material_id.
+    assert by_type["secondary_heating_removal"].material_id is None
+    assert by_type["secondary_heating_removal"].estimated_cost is not None
+    assert abs(by_type["secondary_heating_removal"].estimated_cost - 270.0) <= 1e-6
     for rec in rec_rows:
         assert rec.default is True
         assert rec.already_installed is False
@@ -498,14 +492,8 @@ def test_modelling_recommends_nothing_when_already_at_the_target_band(
                     is_active=True,
                     description="LED bulb",
                 ),
-                MaterialRow(
-                    id=9,
-                    type="secondary_heating_removal",
-                    total_cost=250.0,
-                    cost_unit="gbp_per_unit",
-                    is_active=True,
-                    description="Secondary heating removal",
-                ),
+                # secondary_heating_removal is off-catalogue (priced from the
+                # JSON overlay, not the pgEnum-constrained material table).
             ]
         )
         session.commit()

tests/repositories/product/test_composite_product_repository.py

@@ -0,0 +1,109 @@
+"""Behaviour of the composite ProductRepository: an off-catalogue override
+source checked first, the live catalogue as fallback. The override holds costs
+the ``material`` catalogue cannot supply (today ``secondary_heating_removal``,
+whose Measure Type the ``material.type`` pgEnum does not carry). Resolving it
+from the override keeps that Measure Type away from the DB entirely, while every
+other Measure Type falls through to the catalogue unchanged."""
+
+import pytest
+
+from domain.modelling.product import Product
+from repositories.product.composite_product_repository import (
+    OFF_CATALOGUE_COSTS_PATH,
+    CompositeProductRepository,
+)
+from repositories.product.product_json_repository import ProductJsonRepository
+from repositories.product.product_repository import ProductNotFound, ProductRepository
+
+
+class _StaticRepo(ProductRepository):
+    """A ProductRepository over a fixed measure-type → Product map."""
+
+    def __init__(self, products: dict[str, Product]) -> None:
+        self._products = products
+
+    def get(self, measure_type: str) -> Product:
+        product = self._products.get(measure_type)
+        if product is None:
+            raise ProductNotFound(f"no product for measure type {measure_type!r}")
+        return product
+
+
+class _ExplodingRepo(ProductRepository):
+    """A fallback that must never be reached — any read fails the test loudly,
+    standing in for the catalogue whose pgEnum query would poison the session."""
+
+    def get(self, measure_type: str) -> Product:
+        raise AssertionError(
+            f"fallback was queried for {measure_type!r}; the override should have "
+            "resolved it without touching the catalogue"
+        )
+
+
+def _product(measure_type: str, unit_cost: float) -> Product:
+    return Product(
+        measure_type=measure_type, unit_cost_per_m2=unit_cost, contingency_rate=0.25
+    )
+
+
+def test_get_returns_the_override_without_touching_the_fallback() -> None:
+    # Arrange — the override supplies the off-catalogue cost; the fallback would
+    # raise if consulted (mirroring the live pgEnum DataError on this type).
+    override = _StaticRepo(
+        {"secondary_heating_removal": _product("secondary_heating_removal", 270.0)}
+    )
+    repo = CompositeProductRepository(override=override, fallback=_ExplodingRepo())
+
+    # Act
+    product: Product = repo.get("secondary_heating_removal")
+
+    # Assert
+    assert product.measure_type == "secondary_heating_removal"
+    assert abs(product.unit_cost_per_m2 - 270.0) <= 1e-9
+
+
+def test_get_falls_through_to_the_fallback_when_the_override_misses() -> None:
+    # Arrange — the override has nothing for this type; the catalogue does.
+    override = _StaticRepo({})
+    fallback = _StaticRepo(
+        {"cavity_wall_insulation": _product("cavity_wall_insulation", 18.5)}
+    )
+    repo = CompositeProductRepository(override=override, fallback=fallback)
+
+    # Act
+    product: Product = repo.get("cavity_wall_insulation")
+
+    # Assert
+    assert product.measure_type == "cavity_wall_insulation"
+    assert abs(product.unit_cost_per_m2 - 18.5) <= 1e-9
+
+
+def test_get_raises_product_not_found_when_neither_source_has_it() -> None:
+    # Arrange
+    repo = CompositeProductRepository(
+        override=_StaticRepo({}), fallback=_StaticRepo({})
+    )
+
+    # Act / Assert
+    with pytest.raises(ProductNotFound):
+        repo.get("loft_insulation")
+
+
+def test_committed_override_file_prices_secondary_heating_removal() -> None:
+    # The committed overlay carries the flat per-dwelling decommission cost
+    # (ADR-0028, ported from the legacy heater_removal: (£25 + £200) × 1.2 VAT);
+    # contingency 0.25 is joined from config.
+    product: Product = ProductJsonRepository(OFF_CATALOGUE_COSTS_PATH).get(
+        "secondary_heating_removal"
+    )
+
+    assert abs(product.unit_cost_per_m2 - 270.0) <= 1e-9
+    assert abs(product.contingency_rate - 0.25) <= 1e-9
+
+
+def test_committed_override_file_prices_the_glazing_gaps() -> None:
+    # Glazing is absent from the live catalogue (no active rows), so the overlay
+    # carries per-window costs the generator uses directly (count x unit cost).
+    repo = ProductJsonRepository(OFF_CATALOGUE_COSTS_PATH)
+    assert abs(repo.get("double_glazing").unit_cost_per_m2 - 550.0) <= 1e-9
+    assert abs(repo.get("secondary_glazing").unit_cost_per_m2 - 400.0) <= 1e-9

utilities/aws_lambda/task_handler.py

@@ -42,7 +42,7 @@ def task_handler(
         @wraps(func)
         def wrapper(event: dict[str, Any], context: Any) -> Any:
             with factory() as orchestrator:
-                results: list[Any] = []
+                task_ids: list[dict[str, str]] = []
                 failures: list[dict[str, Any]] = []
 
                 for record in _records(event):
@@ -52,19 +52,21 @@ def task_handler(
                         str(raw_source_id) if raw_source_id is not None else None
                     )
 
-                    _, subtask = orchestrator.create_task_with_subtask(
+                    task, subtask = orchestrator.create_task_with_subtask(
                         task_source=task_source,
                         inputs=body,
                         source=source,
                         source_id=source_id,
                     )
+                    task_ids.append(
+                        {"task_id": str(task.id), "subtask_id": str(subtask.id)}
+                    )
 
                     try:
-                        result = orchestrator.run_subtask(
+                        orchestrator.run_subtask(
                             subtask.id,
                             work=lambda body=body: func(body, context),
                         )
-                        results.append(result)
                     except Exception:
                         logger.exception(
                             "subtask failed (task_source=%s source_id=%s)",
@@ -78,8 +80,8 @@ def task_handler(
                             raise
 
                 if "Records" in event:
-                    return {"batchItemFailures": failures}
-                return results
+                    return {"batchItemFailures": failures, "tasks": task_ids}
+                return task_ids
 
         return wrapper