Merge pull request #1343 from Hestia-Homes/fix/effective-baseline-overrides-and-epc-fidelity

fix(baseline): honour EPC + overrides as the authoritative Effective baseline

CONTEXT.md

@@ -128,7 +128,7 @@ The SAP / EPC Band / carbon emissions / Primary Energy Intensity the modelling p
 _Avoid_: modelled performance, rebaselined performance (only correct when rebaselining ran), scored values
 
 **Calculated SAP10 Performance**:
-The SAP score, EPC Band, CO2 emissions, Primary Energy Intensity, space heating kWh, and hot water kWh produced by **SAP10 Calculation** from a Property's EpcPropertyData. It is **not** a separately-persisted third value-set beside Lodged and Effective: in every baselining scenario the calculator's output *is* the **Effective Performance** (real lodged SAP10 EPC with no overrides ⇒ Calculated = Lodged = Effective; overrides or an estimated / pre-SAP10 EPC ⇒ Calculated = Effective, there being no lodged SAP10 figure to compare against). The calculator is therefore the mechanism that produces Effective Performance, having superseded the old ML-API rebaseliner. The calculator is **load-bearing**: for `sap_version < 10.2` (lodged under a superseded methodology) its output *is* the Effective Performance; for `≥ 10.2` the API's lodged figures are kept and the calculator runs **alongside, logging any divergence** (SAP > 0.5, PEUI/CO2 beyond tolerance) as a validation signal (see [[sap-spec-version]]). It is load-bearing for **Bill Derivation regardless of version** (the EPC lodges no per-end-use kWh), so a calculator strict-raise **aborts the batch** and the un-mapped cert is fixed immediately. ADR-0009 introduced the term, amended by ADR-0010, realized by ADR-0013 (whose shadow stepping-stone is superseded) and ADR-0014.
+The SAP score, EPC Band, CO2 emissions, Primary Energy Intensity, space heating kWh, and hot water kWh produced by **SAP10 Calculation** from a Property's EpcPropertyData. It is **not** a separately-persisted third value-set beside Lodged and Effective: in every baselining scenario the calculator's output *is* the **Effective Performance** (real lodged SAP10 EPC with no overrides ⇒ Calculated = Lodged = Effective; overrides or an estimated / pre-SAP10 EPC ⇒ Calculated = Effective, there being no lodged SAP10 figure to compare against). The calculator is therefore the mechanism that produces Effective Performance, having superseded the old ML-API rebaseliner. The calculator is **load-bearing**: for `sap_version < 10.2` (lodged under a superseded methodology) its output *is* the Effective Performance; for `≥ 10.2` the API's lodged figures are kept **only when the picture is a pristine lodged cert** — when Landlord Overrides or EPC Prediction changed the physical state (Rebaselining trigger (b)/(c)), the calculator's output *is* the Effective Performance even at ≥ 10.2 (ADR-0039), because the accredited lodged figure no longer describes the dwelling. Divergence (SAP > 0.5, PEUI/CO2 beyond tolerance) is logged **only in that pristine-lodged case** — the one case where lodged-vs-calc is a meaningful validation signal (see [[sap-spec-version]]); when the calculator's output is adopted there is nothing to validate against. It is load-bearing for **Bill Derivation regardless of version** (the EPC lodges no per-end-use kWh), so a calculator strict-raise **aborts the batch** and the un-mapped cert is fixed immediately. ADR-0009 introduced the term, amended by ADR-0010, realized by ADR-0013 (whose shadow stepping-stone is superseded) and ADR-0014.
 _Avoid_: calculator output, computed performance, worksheet performance, SAP10 output, calculated value-set (it is not a stored third set)
 
 **SAP10 Calculation**:

docs/adr/0039-override-aware-rebaselining.md

@@ -0,0 +1,27 @@
+# Effective baseline rebaselines off the calculator when overrides or prediction moved the picture
+
+## Context
+
+**Rebaselining** (CONTEXT.md) is meant to fire on three triggers: (a) the cert was lodged under a superseded methodology (`sap_version < 10.2`), (b) **Landlord Overrides / Site Notes changed the physical state**, or (c) the picture was **estimated (EPC Prediction) or remapped**. On any trigger the calculator's scoring of the Effective EPC *is* the Effective Performance; otherwise Effective equals the lodged accredited figure.
+
+The code only implemented trigger (a). `CalculatorRebaseliner` branched solely on `sap_version`: for a cert at `sap_version >= 10.2` it **kept the lodged figure** and ran the calculator only to log divergence — regardless of whether Landlord Overrides had changed the dwelling, or the picture was a synthesised prediction. So a property's stored **Effective baseline echoed the lodged accredited headline** even when the authoritative picture (EPC + overrides) described a different dwelling.
+
+This surfaced as a portfolio-796 hit-list (e.g. property 725634): a lodged ASHP+solar cert scored B/81, but the landlord override reassigned it to electric storage heaters on an uninsulated cavity. The stored Effective stayed **B/81** (lodged echo) while the modelling plan modelled the override-applied picture at **C/D** — an incoherent "already B, post-works C/D" plan. The same applied to predicted properties, whose synthetic headline was trusted as if accredited.
+
+The "trust lodged at ≥10.2" rule (ADR-0037 / the *Calculated SAP10 Performance* glossary entry) exists for a real reason: the deterministic calculator under-reads genuine accredited certs (e.g. 74.6 vs an accredited 81 on the un-overridden 725634 ASHP cert), so for a **pristine** lodged cert the accredited figure is the better baseline. The bug was applying that rule unconditionally — including to certs the landlord had explicitly corrected.
+
+## Decision
+
+Thread a **`physical_state_changed`** signal (`Property.physical_state_changed` — true on Site Notes, Landlord Overrides, or EPC Prediction) from the `PropertyBaselineOrchestrator` into the Rebaseliner. The calculator output becomes Effective Performance whenever **any** trigger fired — `pre_sap10` (a) **or** `physical_state_changed` (b/c) — tagged `pre_sap10` / `physical_state_changed` / `both`. Only a **pristine lodged SAP ≥ 10.2 cert with no overrides** keeps its accredited figure (`reason = "none"`), and that is now the *only* case the calculator runs purely to validate (and the only case the accredited-vs-calc divergence warning is meaningful, so divergence is logged only there).
+
+This realises trigger (b)/(c) as the domain model always claimed — **restoring documented behaviour, not new policy**. "EPC + overrides is authoritative" (decided with Khalim): a landlord's correction is trusted over the lodged cert, and a mistaken override is a *data* problem fixed at source (with the client), not papered over by the baseline.
+
+`physical_state_changed` is deliberately coarse — **any** resolved override trips it, not only a "material" one. A no-op override (asserting what the cert already says) would re-introduce the calculator's known under-read into the baseline; we accept that because landlords rarely enter no-op overrides, and "the assembled picture is authoritative" is the simpler, more defensible rule than a materiality threshold.
+
+## Consequences
+
+- Re-running the baseline flips every overridden / predicted property's Effective from its lodged/synthetic headline to the calculator's scoring of EPC + overrides. **Many legitimately change band** (725634 → D, 739060 → D, 721985 → C, …). The displayed baseline and the modelling plan now derive from the *same* picture, so they agree — the incoherent "B baseline, C/D plan" is gone.
+- FE/product should expect a **visible baseline shift** on overridden/predicted properties — this is correct Rebaselining. Pristine lodged ≥10.2 certs are unchanged.
+- A wrong override now *visibly* downgrades a property (725634 was really an ASHP cert). That is intended: the override is authoritative for modelling, and bad override data is a separate, surfaced concern — an override-plausibility scanner (anomalies like "another dwelling above", community-heated + gas-boiler) is its own ticket.
+- The baseline is only correct on a **faithful** Effective EPC: this decision is paired with [ADR-0040](0040-nested-field-epc-persistence-fidelity.md) (the DB EPC must round-trip without dropping PV arrays / floor flags), and the orchestrator must read the same faithful picture the plan modelled.
+- Numbers finalise only after deploy + re-model (the handler re-fetches the live cert and re-saves it faithfully, then the baseline recomputes). The `rebaseline_reason` enum (`none` / `pre_sap10` / `physical_state_changed` / `both`) already existed in the FE schema; this is a backend-only change.

docs/adr/0040-nested-field-epc-persistence-fidelity.md

@@ -0,0 +1,29 @@
+# EPC persistence must round-trip nested scoring fields, guarded recursively
+
+## Context
+
+The modelling pipeline scores the **DB-stored** `EpcPropertyData` (ADR-0003: the DB is the read model). ADR-0036 added a round-trip-fidelity guarantee — an `EpcPropertyData` saved to the `epc_property` tables, reloaded and mapped back must reconstruct the original — backed by a deep-equality round-trip test and a structural coverage guard.
+
+Both had a blind spot: they only covered **top-level** `EpcPropertyData` fields. The round-trip test only catches a dropped field when a *fixture populates it*, and the chosen fixtures (RdSAP-21.0.0/21.0.1) happened not to exercise certain fields; the structural guard inspected only `dataclasses.fields(EpcPropertyData)`, never recursing into nested objects. So fields on nested sub-objects were dropped **silently**, corrupting the scored baseline:
+
+- `sap_energy_source.photovoltaic_arrays` — no table existed; a dwelling's solar PV was dropped on save, **−12 SAP points** on a real ASHP+solar property (725634: `calc(live)` 74.6 → `calc(reloaded)` 62.1).
+- `SapFloorDimension.is_exposed_floor` / `is_above_partially_heated_space` — no columns; a `True` flag reloaded as the `False` default, **flipping the floor's heat-loss path** (+8 SAP points on 721534).
+- Recursing the guard then surfaced **seven more calculator-read nested fields** with no column (community-heating fuel + CHP fraction, alt-wall `is_sheltered`, wall insulation thermal conductivity, `pv_diverter_present`, measured cylinder volume, AP50 air permeability) — the same silent-drop class.
+
+A lossy read model is corrosive: the handler models off the faithful **in-memory** live cert (so the *plan* is right) but the baseline re-reads the lossy DB projection, so the two disagree for reasons unrelated to any override — compounding the ADR-0039 baseline confusion.
+
+## Decision
+
+Make the `epc_property` projection **faithful for every scoring field**, and make the guard able to *see* nested gaps:
+
+1. **New storage** (FE/Drizzle migrations, applied additively): `epc_photovoltaic_array` child table (one row per array, ordered by `array_index`); `epc_floor_dimension.{is_exposed_floor, is_above_partially_heated_space}`; and columns for the seven calculator-read nested fields. Python SQLModel + `save`/`_compose` mappers wire each, with deep-equality round-trip tests.
+2. **Recursive structural guard**: the ADR-0036 coverage guard now walks **every domain dataclass reachable from `EpcPropertyData`** (through Optionals/lists) and asserts each field is either reconstructed by a `_compose`/`_to_*` mapper or on an allow-list keyed by `Class.field`. Adding a field at any depth now forces a conscious persist-or-justify decision.
+
+Fields the calculator does **not** read stay allow-listed as dormant (e.g. `SapBuildingPart.{floor_u_value, roof_u_value, wall_u_value}`); a few structural gaps (`SapRoofWindow`, room-in-roof detail, alt-wall `u_value`/`wall_thickness_mm`) remain FE-column-pending and are explicitly allow-listed, not silent. The top-level `extract_fans_count` (a mirror of the persisted `sap_ventilation` copy the calculator actually reads) is reconstructed from that column so the object round-trips complete with no exception.
+
+## Consequences
+
+- The DB EPC round-trips **exactly for every scoring field** — verified against the live schema: 725634 and 721534 now reload with SAP delta `+0.00` (was −12.5 / +8.2). Residual non-equality is limited to allow-listed dormant fields and a sub-millimetre `real`-column float artifact, none scoring-relevant.
+- This is a **prerequisite for [ADR-0039](0039-override-aware-rebaselining.md)**: storing `calc(EPC + overrides)` as the Effective baseline is only correct if the stored EPC is faithful. Until a property is re-modelled (handler re-fetches + re-saves), its existing DB row stays lossy, so the live baseline finalises only after deploy + re-model.
+- A class of silent persistence drops can no longer hide: the recursive guard fails CI on any unjustified nested gap.
+- New EPC scoring fields now carry a persistence cost (column + mapper + round-trip assertion) by construction — the intended friction.

docs/migrations/epc-property-round-trip-fidelity.md

@@ -58,14 +58,60 @@ now fails if any `EpcPropertyData` field is neither reconstructed by `_compose`
 allow-list — so the gaps below are no longer "latent until a fixture exercises them", they are
 explicit and tracked.
 
-**Still open (follow-up issues):** the remaining §2 structural tables (room-in-roof detail, PV
-arrays, **roof windows** — `sap_roof_windows`) + §3 nested-wall fields
-(`SapAlternativeWall.u_value`/`wall_thickness_mm`) + `SapFloorDimension` exposed-floor flags + the
+**Still open (follow-up issues):** the remaining §2 structural tables (room-in-roof detail,
+**roof windows** — `sap_roof_windows`) + §3 nested-wall fields
+(`SapAlternativeWall.u_value`/`wall_thickness_mm`) + the
 solar-HW collector columns (`solar_hw_collector_orientation` / `_pitch_deg` / `_overshading`, §3.1).
 Each is on the structural-guard allow-list with a reason until its columns land.
 
 ---
 
+## Slice 2 — PV arrays + floor heat-loss flags (#TBD)
+
+Closed two **scoring-relevant** round-trip gaps that were corrupting the modelling baseline by up
+to ~12 SAP points on a real Hyde property (725634, an ASHP + solar dwelling whose DB-read picture
+scored ~12 points below the in-memory cert it was modelled from):
+
+- **§2 `epc_photovoltaic_array` child table** — `sap_energy_source.photovoltaic_arrays` (a list of
+  `peak_power` kWp + `pitch`/`overshading`/`orientation` SAP codes) had no table, so every array was
+  dropped on save and the dwelling lost its PV generation credit. Now persisted (one row per array,
+  ordered by `array_index`) and reconstructed in `_to_energy_source`. **Round-trip test asserts
+  deep-equality** (`test_photovoltaic_arrays_round_trip`).
+- **§3 `epc_floor_dimension` flags** — `is_exposed_floor` + `is_above_partially_heated_space` had no
+  columns, so a `True` flag round-tripped back to the `False` default and silently flipped the
+  floor's heat-loss path. Now persisted and reconstructed in `_to_floor_dimension`. **Round-trip
+  test** `test_floor_dimension_heat_loss_flags_round_trip`.
+
+DB migration: FE branch `feature/epc-pv-and-floor-heatloss-schema` (additive — new table + two
+`NOT NULL DEFAULT false` columns; no backfill). Run `drizzle-kit generate`/migrate **before**
+deploying this backend code.
+
+### Recursive-guard gaps (surfaced AND closed)
+
+Generalising the ADR-0036 structural guard to recurse into **nested** dataclasses (it previously
+saw only `EpcPropertyData`'s top-level fields — which is exactly how the PV-array and floor-flag
+drops stayed latent) surfaced **7 pre-existing, calculator-read** nested fields with no FE column —
+silent-drop gaps of the **same class as the PV bug**. All 7 are now **closed**: FE columns added
+(additive migration), Python SQLModel + save/`_compose` wired, and a deep-equality round-trip test
+(`test_calculator_read_fields_round_trip`) pins them. Their allow-list entries were removed, so the
+guard now *enforces* their reconstruction.
+
+| Field | Lives on | DB column (table) |
+|---|---|---|
+| `community_heating_boiler_fuel_type`, `community_heating_chp_fraction` | `MainHeatingDetail` | `epc_main_heating_detail` |
+| `is_sheltered` | `SapAlternativeWall` | `epc_building_part.alt_wall_{1,2}_is_sheltered` |
+| `wall_insulation_thermal_conductivity` | `SapBuildingPart` | `epc_building_part` (jsonb) |
+| `pv_diverter_present` | `SapEnergySource` | `epc_property.energy_pv_diverter_present` |
+| `cylinder_volume_measured_l` | `SapHeating` | `epc_property.heating_cylinder_volume_measured_l` |
+| `air_permeability_ap50_m3_h_m2` | `SapVentilation` | `epc_property.ventilation_air_permeability_ap50_m3_h_m2` |
+
+(Still allow-listed as dormant — not calculator-read, no column: `SapBuildingPart.{floor_u_value,
+roof_u_value, wall_u_value, wall_is_basement, rafter_insulation_thickness}`,
+`SapAlternativeWall.is_basement`, `SapHeating.cylinder_heat_loss`. And `SapAlternativeWall.{u_value,
+wall_thickness_mm}` remain FE-column-pending.)
+
+---
+
 ## 1. Type fidelity — convert `Union[int, str]` code columns to JSONB
 
 These columns hold SAP/RdSAP categorical codes that are **`int` from the gov API** and **`str`

domain/property/property.py

@@ -84,6 +84,19 @@ class Property:
             "no source path to model from"
         )
 
+    @property
+    def physical_state_changed(self) -> bool:
+        """True when the Effective EPC was assembled from something other than a
+        pristine lodged cert — Site Notes superseded it, Landlord Overrides were
+        folded on, or it was synthesised by EPC Prediction (Rebaselining trigger
+        (b)/(c), CONTEXT.md). The Rebaseliner uses this to adopt the calculator's
+        output over the accredited lodged figure even at SAP >= 10.2. A pristine
+        lodged EPC with no overrides returns False."""
+        path = self.source_path
+        if path == "site_notes" or path == "predicted":
+            return True
+        return bool(self.landlord_overrides)
+
     @property
     def effective_epc(self) -> EpcPropertyData:
         """The EpcPropertyData the modelling pipeline scores against.

domain/property_baseline/calculator_rebaseliner.py

@@ -4,7 +4,11 @@ import logging
 from typing import TYPE_CHECKING, Optional
 
 from domain.property_baseline.performance import Performance
-from domain.property_baseline.rebaseliner import Rebaseliner, RebaselineResult
+from domain.property_baseline.rebaseliner import (
+    Rebaseliner,
+    RebaselineReason,
+    RebaselineResult,
+)
 
 if TYPE_CHECKING:
     from datatypes.epc.domain.epc_property_data import EpcPropertyData
@@ -50,7 +54,12 @@ class CalculatorRebaseliner(Rebaseliner):
         self._calculator = calculator
 
     def rebaseline(
-        self, property_id: int, effective_epc: "EpcPropertyData", lodged: Performance
+        self,
+        property_id: int,
+        effective_epc: "EpcPropertyData",
+        lodged: Performance,
+        *,
+        physical_state_changed: bool = False,
     ) -> RebaselineResult:
         # A raise (UnmappedSapCode, etc.) propagates: the calculator is
         # load-bearing, so the batch aborts and the cert is fixed at once. The
@@ -58,10 +67,24 @@ class CalculatorRebaseliner(Rebaseliner):
         # regardless of whether lodged or calculated figures win (ADR-0013/0014).
         result: SapResult = self._calculator.calculate(effective_epc)
         sap_version: Optional[float] = effective_epc.sap_version
-        if sap_version is not None and sap_version < _MIN_TRUSTED_SAP_VERSION:
+        pre_sap10 = sap_version is not None and sap_version < _MIN_TRUSTED_SAP_VERSION
+        # The calculator output IS Effective Performance whenever a Rebaselining
+        # trigger fired: (a) a superseded methodology (pre-SAP10), or (b)/(c) the
+        # physical state was changed by Landlord Overrides / Prediction. Only a
+        # pristine lodged SAP >= 10.2 cert keeps its accredited figure — that is
+        # the *only* case where the calculator runs purely to validate (and where
+        # its known divergence from the accredited register would mislead).
+        if pre_sap10 or physical_state_changed:
+            reason: RebaselineReason = (
+                "both"
+                if pre_sap10 and physical_state_changed
+                else "pre_sap10"
+                if pre_sap10
+                else "physical_state_changed"
+            )
             return RebaselineResult(
                 effective=Performance.from_sap_result(result),
-                reason="pre_sap10",
+                reason=reason,
                 sap_result=result,
             )
         self._log_divergence(

domain/property_baseline/rebaseliner.py

@@ -59,8 +59,20 @@ class Rebaseliner(ABC):
 
     @abstractmethod
     def rebaseline(
-        self, property_id: int, effective_epc: EpcPropertyData, lodged: Performance
-    ) -> RebaselineResult: ...
+        self,
+        property_id: int,
+        effective_epc: EpcPropertyData,
+        lodged: Performance,
+        *,
+        physical_state_changed: bool = False,
+    ) -> RebaselineResult:
+        """Produce Effective Performance. ``physical_state_changed`` is True when
+        the Effective EPC was assembled from something other than a pristine
+        lodged cert — Landlord Overrides, Site Notes, or EPC Prediction moved the
+        physical picture (Rebaselining trigger (b)/(c)) — so the accredited lodged
+        figure no longer describes the dwelling and the calculator output wins
+        even at SAP >= 10.2."""
+        ...
 
 
 class StubRebaseliner(Rebaseliner):
@@ -76,7 +88,12 @@ class StubRebaseliner(Rebaseliner):
     """
 
     def rebaseline(
-        self, property_id: int, effective_epc: EpcPropertyData, lodged: Performance
+        self,
+        property_id: int,
+        effective_epc: EpcPropertyData,
+        lodged: Performance,
+        *,
+        physical_state_changed: bool = False,
     ) -> RebaselineResult:
         sap_version = effective_epc.sap_version
         if sap_version is not None and sap_version < _SAP10_FLOOR:
@@ -84,4 +101,11 @@ class StubRebaseliner(Rebaseliner):
                 f"Property needs rebaselining (pre-SAP10 cert, sap_version="
                 f"{sap_version}); this stub does not run the calculator"
             )
+        # A physical-state change needs the calculator this stub does not run;
+        # raise rather than fabricate a "none" that ignores the overrides.
+        if physical_state_changed:
+            raise RebaselineNotImplemented(
+                "Property needs rebaselining (physical state changed by overrides "
+                "/ prediction); this stub does not run the calculator"
+            )
         return RebaselineResult(effective=lodged, reason="none", sap_result=None)

infrastructure/postgres/epc_property_table.py

@@ -10,6 +10,7 @@ from datatypes.epc.domain.epc_property_data import (
     EpcPropertyData,
     EnergyElement,
     MainHeatingDetail,
+    PhotovoltaicArray,
     RenewableHeatIncentive,
     SapBuildingPart,
     SapFloorDimension,
@@ -138,6 +139,9 @@ class EpcPropertyModel(SQLModel, table=True):
     )
     energy_pv_percent_roof_area: Optional[int] = Field(default=None)
     energy_pv_battery_capacity: Optional[float] = Field(default=None)
+    # An EV/PV diverter present on the dwelling (SAP §M; the calculator reads it
+    # via cert_to_inputs). Non-optional bool defaulting False, matching the domain.
+    energy_pv_diverter_present: bool = Field(default=False)
     energy_wind_turbine_hub_height: Optional[float] = Field(default=None)
     energy_wind_turbine_rotor_diameter: Optional[float] = Field(default=None)
 
@@ -161,6 +165,7 @@ class EpcPropertyModel(SQLModel, table=True):
         default=None, sa_column=Column(JSONB, nullable=True)
     )
     heating_cylinder_insulation_thickness_mm: Optional[int] = Field(default=None)
+    heating_cylinder_volume_measured_l: Optional[int] = Field(default=None)
     heating_wwhrs_index_number_1: Optional[int] = Field(default=None)
     heating_wwhrs_index_number_2: Optional[int] = Field(default=None)
     heating_shower_outlet_type: Optional[Union[int, str]] = Field(
@@ -192,6 +197,7 @@ class EpcPropertyModel(SQLModel, table=True):
     ventilation_suspended_timber_floor_sealed: Optional[bool] = Field(default=None)
     ventilation_has_draught_lobby: Optional[bool] = Field(default=None)
     ventilation_air_permeability_ap4_m3_h_m2: Optional[float] = Field(default=None)
+    ventilation_air_permeability_ap50_m3_h_m2: Optional[float] = Field(default=None)
     ventilation_mechanical_ventilation_kind: Optional[str] = Field(default=None)
     mechanical_ventilation: Optional[int] = Field(default=None)
     mechanical_vent_duct_type: Optional[int] = Field(default=None)
@@ -340,6 +346,7 @@ class EpcPropertyModel(SQLModel, table=True):
                 pv.none_or_no_details.percent_roof_area if pv else None
             ),
             energy_pv_battery_capacity=pvb.pv_battery.battery_capacity if pvb else None,
+            energy_pv_diverter_present=es.pv_diverter_present,
             energy_wind_turbine_hub_height=wt.hub_height if wt else None,
             energy_wind_turbine_rotor_diameter=wt.rotor_diameter if wt else None,
             heating_cylinder_size=h.cylinder_size,
@@ -351,6 +358,7 @@ class EpcPropertyModel(SQLModel, table=True):
             heating_secondary_fuel_type=h.secondary_fuel_type,
             heating_secondary_heating_type=h.secondary_heating_type,
             heating_cylinder_insulation_thickness_mm=h.cylinder_insulation_thickness_mm,
+            heating_cylinder_volume_measured_l=h.cylinder_volume_measured_l,
             heating_wwhrs_index_number_1=h.instantaneous_wwhrs.wwhrs_index_number1,
             heating_wwhrs_index_number_2=h.instantaneous_wwhrs.wwhrs_index_number2,
             heating_shower_outlet_type=shower.shower_outlet_type if shower else None,
@@ -384,6 +392,9 @@ class EpcPropertyModel(SQLModel, table=True):
             ventilation_air_permeability_ap4_m3_h_m2=(
                 v.air_permeability_ap4_m3_h_m2 if v else None
             ),
+            ventilation_air_permeability_ap50_m3_h_m2=(
+                v.air_permeability_ap50_m3_h_m2 if v else None
+            ),
             ventilation_mechanical_ventilation_kind=(
                 v.mechanical_ventilation_kind if v else None
             ),
@@ -543,6 +554,10 @@ class EpcMainHeatingDetailModel(SQLModel, table=True):
     main_heating_data_source: Optional[int] = Field(default=None)
     condensing: Optional[bool] = Field(default=None)
     weather_compensator: Optional[bool] = Field(default=None)
+    # Community-heating fields (SAP Table 4d; the calculator reads them via
+    # cert_to_inputs for a community-heated dwelling).
+    community_heating_boiler_fuel_type: Optional[int] = Field(default=None)
+    community_heating_chp_fraction: Optional[float] = Field(default=None)
 
     @classmethod
     def from_domain(
@@ -568,6 +583,8 @@ class EpcMainHeatingDetailModel(SQLModel, table=True):
             main_heating_data_source=detail.main_heating_data_source,
             condensing=detail.condensing,
             weather_compensator=detail.weather_compensator,
+            community_heating_boiler_fuel_type=detail.community_heating_boiler_fuel_type,
+            community_heating_chp_fraction=detail.community_heating_chp_fraction,
         )
 
 
@@ -600,6 +617,11 @@ class EpcBuildingPartModel(SQLModel, table=True):
     wall_insulation_thickness: Optional[Union[str, int]] = Field(
         default=None, sa_column=Column(JSONB, nullable=True)
     )
+    # Union[int, str] SAP code (int from the API, str from Site Notes) — JSONB to
+    # preserve the type on round-trip, like the insulation-thickness siblings.
+    wall_insulation_thermal_conductivity: Optional[Union[int, str]] = Field(
+        default=None, sa_column=Column(JSONB, nullable=True)
+    )
     floor_heat_loss: Optional[int] = Field(default=None)
     floor_insulation_thickness: Optional[str] = Field(default=None)
     flat_roof_insulation_thickness: Optional[Union[str, int]] = Field(
@@ -626,12 +648,14 @@ class EpcBuildingPartModel(SQLModel, table=True):
     alt_wall_1_insulation_type: Optional[int] = Field(default=None)
     alt_wall_1_thickness_measured: Optional[str] = Field(default=None)
     alt_wall_1_insulation_thickness: Optional[str] = Field(default=None)
+    alt_wall_1_is_sheltered: Optional[bool] = Field(default=None)
     alt_wall_2_area: Optional[float] = Field(default=None)
     alt_wall_2_dry_lined: Optional[str] = Field(default=None)
     alt_wall_2_construction: Optional[int] = Field(default=None)
     alt_wall_2_insulation_type: Optional[int] = Field(default=None)
     alt_wall_2_thickness_measured: Optional[str] = Field(default=None)
     alt_wall_2_insulation_thickness: Optional[str] = Field(default=None)
+    alt_wall_2_is_sheltered: Optional[bool] = Field(default=None)
 
     @classmethod
     def from_domain(
@@ -652,6 +676,7 @@ class EpcBuildingPartModel(SQLModel, table=True):
             wall_dry_lined=part.wall_dry_lined,
             wall_thickness_mm=part.wall_thickness_mm,
             wall_insulation_thickness=part.wall_insulation_thickness,
+            wall_insulation_thermal_conductivity=part.wall_insulation_thermal_conductivity,
             floor_heat_loss=part.floor_heat_loss,
             floor_insulation_thickness=part.floor_insulation_thickness,
             flat_roof_insulation_thickness=part.flat_roof_insulation_thickness,
@@ -676,6 +701,7 @@ class EpcBuildingPartModel(SQLModel, table=True):
             alt_wall_1_insulation_thickness=(
                 aw1.wall_insulation_thickness if aw1 else None
             ),
+            alt_wall_1_is_sheltered=aw1.is_sheltered if aw1 else None,
             alt_wall_2_area=aw2.wall_area if aw2 else None,
             alt_wall_2_dry_lined=aw2.wall_dry_lined if aw2 else None,
             alt_wall_2_construction=aw2.wall_construction if aw2 else None,
@@ -684,6 +710,7 @@ class EpcBuildingPartModel(SQLModel, table=True):
             alt_wall_2_insulation_thickness=(
                 aw2.wall_insulation_thickness if aw2 else None
             ),
+            alt_wall_2_is_sheltered=aw2.is_sheltered if aw2 else None,
         )
 
 
@@ -702,6 +729,11 @@ class EpcFloorDimensionModel(SQLModel, table=True):
     heat_loss_perimeter_m: float
     floor_insulation: Optional[int] = Field(default=None)
     floor_construction: Optional[int] = Field(default=None)
+    # Heat-loss flags (SAP 10.2 §3.3): a floor over outside air / over a
+    # partially-heated space takes a different U-value path. Default False
+    # matches the domain default for an ordinary ground floor.
+    is_exposed_floor: bool = Field(default=False)
+    is_above_partially_heated_space: bool = Field(default=False)
 
     @classmethod
     def from_domain(
@@ -716,6 +748,8 @@ class EpcFloorDimensionModel(SQLModel, table=True):
             heat_loss_perimeter_m=dim.heat_loss_perimeter_m,
             floor_insulation=dim.floor_insulation,
             floor_construction=dim.floor_construction,
+            is_exposed_floor=dim.is_exposed_floor,
+            is_above_partially_heated_space=dim.is_above_partially_heated_space,
         )
 
 
@@ -771,6 +805,37 @@ class EpcWindowModel(SQLModel, table=True):
         )
 
 
+class EpcPhotovoltaicArrayModel(SQLModel, table=True):
+    __tablename__: ClassVar[str] = "epc_photovoltaic_array"  # pyright: ignore[reportIncompatibleVariableOverride]
+
+    id: Optional[int] = Field(default=None, primary_key=True)
+    epc_property_id: int = Field(foreign_key="epc_property.id", nullable=False)
+
+    # List position on `sap_energy_source.photovoltaic_arrays`. Persisted so the
+    # array order is recoverable for round-trip equality (a dwelling can carry
+    # several arrays at different pitch/orientation).
+    array_index: int
+    # kWp as a float (e.g. 3.24). pitch / overshading / orientation are SAP
+    # enumeration CODES (small categorical ints), not physical degrees.
+    peak_power: float
+    pitch: int
+    overshading: int
+    orientation: Optional[int] = Field(default=None)
+
+    @classmethod
+    def from_domain(
+        cls, array: PhotovoltaicArray, array_index: int, epc_property_id: int
+    ) -> EpcPhotovoltaicArrayModel:
+        return cls(
+            epc_property_id=epc_property_id,
+            array_index=array_index,
+            peak_power=array.peak_power,
+            pitch=array.pitch,
+            overshading=array.overshading,
+            orientation=array.orientation,
+        )
+
+
 class EpcEnergyElementModel(SQLModel, table=True):
     __tablename__: ClassVar[str] = "epc_energy_element"  # pyright: ignore[reportIncompatibleVariableOverride]
 

orchestration/property_baseline_orchestrator.py

@@ -52,7 +52,10 @@ class PropertyBaselineOrchestrator:
                 effective_epc = prop.effective_epc
                 lodged = lodged_performance(effective_epc)
                 rebaselined = self._rebaseliner.rebaseline(
-                    property_id, effective_epc, lodged
+                    property_id,
+                    effective_epc,
+                    lodged,
+                    physical_state_changed=prop.physical_state_changed,
                 )
                 # No SapResult (the stub path) means no scored picture to price,
                 # so the bill stays None.

repositories/epc/epc_postgres_repository.py

@@ -14,6 +14,7 @@ from datatypes.epc.domain.epc_property_data import (
     EpcPropertyData,
     InstantaneousWwhrs,
     MainHeatingDetail,
+    PhotovoltaicArray,
     PhotovoltaicSupply,
     PhotovoltaicSupplyNoneOrNoDetails,
     PvBatteries,
@@ -41,6 +42,7 @@ from infrastructure.postgres.epc_property_table import (
     EpcFlatDetailsModel,
     EpcFloorDimensionModel,
     EpcMainHeatingDetailModel,
+    EpcPhotovoltaicArrayModel,
     EpcPropertyEnergyPerformanceModel,
     EpcPropertyModel,
     EpcRenewableHeatIncentiveModel,
@@ -140,6 +142,10 @@ class EpcPostgresRepository(EpcRepository):
                 self._session.add(EpcFloorDimensionModel.from_domain(dim, bp_id))
         for window in data.sap_windows:
             self._session.add(EpcWindowModel.from_domain(window, epc_property_id))
+        for index, array in enumerate(data.sap_energy_source.photovoltaic_arrays or []):
+            self._session.add(
+                EpcPhotovoltaicArrayModel.from_domain(array, index, epc_property_id)
+            )
 
         for element_type, elements in (
             ("roof", data.roofs),
@@ -211,6 +217,7 @@ class EpcPostgresRepository(EpcRepository):
             EpcMainHeatingDetailModel,
             EpcBuildingPartModel,
             EpcWindowModel,
+            EpcPhotovoltaicArrayModel,
             EpcFlatDetailsModel,
             EpcRenewableHeatIncentiveModel,
         ):
@@ -327,6 +334,13 @@ class EpcPostgresRepository(EpcRepository):
                 .order_by(EpcWindowModel.id)  # type: ignore[arg-type]
             ).all()
         )
+        pv_arrays_by = _group_by_epc(
+            self._session.exec(
+                select(EpcPhotovoltaicArrayModel)
+                .where(col(EpcPhotovoltaicArrayModel.epc_property_id).in_(epc_ids))
+                .order_by(EpcPhotovoltaicArrayModel.array_index)  # type: ignore[arg-type]
+            ).all()
+        )
         part_ids = [
             bp.id
             for parts in parts_by.values()
@@ -346,6 +360,7 @@ class EpcPostgresRepository(EpcRepository):
                 part_rows=parts_by.get(epc_id, []),
                 floor_dims_by_part=floor_dims_by_part,
                 window_rows=windows_by.get(epc_id, []),
+                pv_array_rows=pv_arrays_by.get(epc_id, []),
                 flat_row=flat_by.get(epc_id),
                 rhi_row=rhi_by.get(epc_id),
             )
@@ -407,6 +422,7 @@ class EpcPostgresRepository(EpcRepository):
             )
         ).first()
         window_rows = self._windows(epc_property_id)
+        pv_array_rows = self._pv_arrays(epc_property_id)
         floor_dims_by_part = self._floor_dims_by_part(
             [bp.id for bp in part_rows if bp.id is not None]
         )
@@ -418,6 +434,7 @@ class EpcPostgresRepository(EpcRepository):
             part_rows=part_rows,
             floor_dims_by_part=floor_dims_by_part,
             window_rows=window_rows,
+            pv_array_rows=pv_array_rows,
             flat_row=flat_row,
             rhi_row=rhi_row,
         )
@@ -432,6 +449,7 @@ class EpcPostgresRepository(EpcRepository):
         part_rows: list[EpcBuildingPartModel],
         floor_dims_by_part: dict[int, list[EpcFloorDimensionModel]],
         window_rows: list[EpcWindowModel],
+        pv_array_rows: list[EpcPhotovoltaicArrayModel],
         flat_row: Optional[EpcFlatDetailsModel],
         rhi_row: Optional[EpcRenewableHeatIncentiveModel],
     ) -> EpcPropertyData:
@@ -457,7 +475,7 @@ class EpcPostgresRepository(EpcRepository):
             door_count=p.door_count,
             sap_heating=self._to_sap_heating(p, heating_rows),
             sap_windows=[self._to_window(w) for w in window_rows],
-            sap_energy_source=self._to_energy_source(p),
+            sap_energy_source=self._to_energy_source(p, pv_array_rows),
             sap_building_parts=[
                 self._to_building_part(
                     bp, floor_dims_by_part.get(bp.id, []) if bp.id is not None else []
@@ -564,6 +582,12 @@ class EpcPostgresRepository(EpcRepository):
                 else None
             ),
             multiple_glazed_proportion=p.multiple_glazed_proportion,
+            # Top-level mirror of sap_ventilation.extract_fans_count (the mapper
+            # sets both from the same source). The deterministic calculator reads
+            # the sap_ventilation copy, not this one; reconstruct it from the same
+            # column anyway so EpcPropertyData round-trips complete, with no
+            # allow-list exception for a field that trivially mirrors a persisted one.
+            extract_fans_count=p.ventilation_extract_fans_count,
             calculation_software_version=p.calculation_software_version,
             mechanical_vent_duct_placement=p.mechanical_vent_duct_placement,
             mechanical_vent_duct_insulation=p.mechanical_vent_duct_insulation,
@@ -622,6 +646,18 @@ class EpcPostgresRepository(EpcRepository):
             ).all()
         )
 
+    @private
+    def _pv_arrays(
+        self, epc_property_id: int
+    ) -> list[EpcPhotovoltaicArrayModel]:
+        return list(
+            self._session.exec(
+                select(EpcPhotovoltaicArrayModel)
+                .where(EpcPhotovoltaicArrayModel.epc_property_id == epc_property_id)
+                .order_by(EpcPhotovoltaicArrayModel.array_index)  # type: ignore[arg-type]
+            ).all()
+        )
+
     @private
     def _to_energy_element(self, e: EpcEnergyElementModel) -> EnergyElement:
         return EnergyElement(
@@ -661,6 +697,7 @@ class EpcPostgresRepository(EpcRepository):
             secondary_fuel_type=p.heating_secondary_fuel_type,
             secondary_heating_type=p.heating_secondary_heating_type,
             cylinder_insulation_thickness_mm=p.heating_cylinder_insulation_thickness_mm,
+            cylinder_volume_measured_l=p.heating_cylinder_volume_measured_l,
             number_baths=p.heating_number_baths,
             number_baths_wwhrs=p.heating_number_baths_wwhrs,
             electric_shower_count=p.heating_electric_shower_count,
@@ -688,6 +725,8 @@ class EpcPostgresRepository(EpcRepository):
             main_heating_data_source=m.main_heating_data_source,
             condensing=m.condensing,
             weather_compensator=m.weather_compensator,
+            community_heating_boiler_fuel_type=m.community_heating_boiler_fuel_type,
+            community_heating_chp_fraction=m.community_heating_chp_fraction,
         )
 
     @private
@@ -738,6 +777,7 @@ class EpcPostgresRepository(EpcRepository):
             wall_dry_lined=bp.wall_dry_lined,
             wall_thickness_mm=bp.wall_thickness_mm,
             wall_insulation_thickness=bp.wall_insulation_thickness,
+            wall_insulation_thermal_conductivity=bp.wall_insulation_thermal_conductivity,
             sap_alternative_wall_1=self._to_alt_wall(bp, 1),
             sap_alternative_wall_2=self._to_alt_wall(bp, 2),
             floor_heat_loss=bp.floor_heat_loss,
@@ -789,6 +829,7 @@ class EpcPostgresRepository(EpcRepository):
             if n == 1
             else bp.alt_wall_2_insulation_thickness
         )
+        sheltered = bp.alt_wall_1_is_sheltered if n == 1 else bp.alt_wall_2_is_sheltered
         return SapAlternativeWall(
             wall_area=area,
             wall_dry_lined=_require(dry_lined, f"alt_wall_{n}_dry_lined"),
@@ -800,6 +841,9 @@ class EpcPostgresRepository(EpcRepository):
                 thickness_measured, f"alt_wall_{n}_thickness_measured"
             ),
             wall_insulation_thickness=insulation_thickness,
+            # Nullable column (added later than the alt wall itself); a row from
+            # before the column existed reads None → the domain default False.
+            is_sheltered=sheltered if sheltered is not None else False,
         )
 
     @private
@@ -812,11 +856,29 @@ class EpcPostgresRepository(EpcRepository):
             floor=f.floor,
             floor_insulation=f.floor_insulation,
             floor_construction=f.floor_construction,
+            is_exposed_floor=f.is_exposed_floor,
+            is_above_partially_heated_space=f.is_above_partially_heated_space,
         )
 
     @private
-    def _to_energy_source(self, p: EpcPropertyModel) -> SapEnergySource:
+    def _to_energy_source(
+        self, p: EpcPropertyModel, pv_array_rows: list[EpcPhotovoltaicArrayModel]
+    ) -> SapEnergySource:
         return SapEnergySource(
+            photovoltaic_arrays=(
+                [
+                    PhotovoltaicArray(
+                        peak_power=a.peak_power,
+                        pitch=a.pitch,
+                        overshading=a.overshading,
+                        orientation=a.orientation,
+                    )
+                    for a in sorted(pv_array_rows, key=lambda a: a.array_index)
+                ]
+                if pv_array_rows
+                else None
+            ),
+            pv_diverter_present=p.energy_pv_diverter_present,
             mains_gas=p.energy_mains_gas,
             meter_type=p.energy_meter_type,
             pv_battery_count=p.energy_pv_battery_count,
@@ -902,6 +964,7 @@ class EpcPostgresRepository(EpcRepository):
             suspended_timber_floor_sealed=p.ventilation_suspended_timber_floor_sealed,
             has_draught_lobby=p.ventilation_has_draught_lobby,
             air_permeability_ap4_m3_h_m2=p.ventilation_air_permeability_ap4_m3_h_m2,
+            air_permeability_ap50_m3_h_m2=p.ventilation_air_permeability_ap50_m3_h_m2,
             mechanical_ventilation_kind=p.ventilation_mechanical_ventilation_kind,
         )
 

tests/domain/property/test_property_physical_state_changed.py

@@ -0,0 +1,53 @@
+"""`Property.physical_state_changed` — the Rebaselining trigger (b)/(c) signal.
+
+True when the Effective EPC was assembled from something other than a pristine
+lodged cert: Landlord Overrides folded on, Site Notes superseding the cert, or an
+EPC-Prediction synthesis. The PropertyBaselineOrchestrator threads it into the
+Rebaseliner so an overridden / predicted SAP >= 10.2 picture rebaselines off the
+calculator instead of echoing the (now-stale) accredited lodged figure.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+from domain.epc.property_overlays.wall_type_overlay import wall_overlay_for
+from domain.property.property import Property, PropertyIdentity
+
+_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples"
+
+
+def _epc() -> EpcPropertyData:
+    raw: dict[str, Any] = json.loads(
+        (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
+    )
+    return EpcPropertyDataMapper.from_api_response(raw)
+
+
+def _identity() -> PropertyIdentity:
+    return PropertyIdentity(
+        portfolio_id=1, postcode="A0 0AA", address="1 Some Street", uprn=12345
+    )
+
+
+def test_pristine_lodged_cert_with_no_overrides_is_unchanged() -> None:
+    prop = Property(identity=_identity(), epc=_epc())
+    assert prop.physical_state_changed is False
+
+
+def test_lodged_cert_with_landlord_overrides_is_changed() -> None:
+    overlay = wall_overlay_for("Solid brick, with internal insulation", 0)
+    assert overlay is not None
+    prop = Property(identity=_identity(), epc=_epc(), landlord_overrides=[overlay])
+    assert prop.physical_state_changed is True
+
+
+def test_predicted_property_is_changed() -> None:
+    # A neighbour-synthesised picture is assembled, not a real lodged cert, so it
+    # rebaselines off the calculator (trigger (c)).
+    prop = Property(identity=_identity(), epc=None, predicted_epc=_epc())
+    assert prop.physical_state_changed is True

tests/domain/property_baseline/test_calculator_rebaseliner.py

@@ -113,6 +113,75 @@ def test_a_10_2_cert_keeps_the_lodged_figures() -> None:
     assert result.sap_result is sap_result
 
 
+def test_a_10_2_cert_with_changed_physical_state_uses_the_calculator() -> None:
+    # Arrange — a 10.2 cert whose physical state was changed by Landlord
+    # Overrides (or assembled by Prediction): the accredited lodged figure no
+    # longer describes the dwelling, so the calculator's output becomes Effective
+    # Performance (Rebaselining trigger (b)/(c), CONTEXT.md), not the lodged value.
+    sap_result = _sap_result(
+        sap_score=67, co2_kg_per_yr=2100.0, primary_energy_kwh_per_m2=210.0
+    )
+    rebaseliner = CalculatorRebaseliner(_StubCalculator(sap_result))
+    epc = _epc(sap_version=10.2)
+
+    # Act
+    result = rebaseliner.rebaseline(
+        property_id=10,
+        effective_epc=epc,
+        lodged=_lodged(),
+        physical_state_changed=True,
+    )
+
+    # Assert — calculated Performance wins, tagged physical_state_changed.
+    assert result.effective == Performance(
+        sap_score=67, epc_band=Epc.D, co2_emissions=2.1, primary_energy_intensity=210
+    )
+    assert result.reason == "physical_state_changed"
+    assert result.sap_result is sap_result
+
+
+def test_a_pre_10_2_cert_with_changed_physical_state_is_reason_both() -> None:
+    # Arrange — both triggers fire: a superseded methodology (sap_version < 10.2)
+    # AND a physical-state change. The calculator wins (as for either alone); the
+    # reason records both.
+    rebaseliner = CalculatorRebaseliner(_StubCalculator(_sap_result(sap_score=65)))
+    epc = _epc(sap_version=10.0)
+
+    # Act
+    result = rebaseliner.rebaseline(
+        property_id=10,
+        effective_epc=epc,
+        lodged=_lodged(),
+        physical_state_changed=True,
+    )
+
+    # Assert
+    assert result.effective.sap_score == 65
+    assert result.reason == "both"
+
+
+def test_a_10_2_cert_with_changed_state_does_not_log_divergence(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # Arrange — when we ADOPT the calculator output (physical state changed),
+    # there is no accredited lodged figure to validate against, so no divergence
+    # warning is emitted (calc 90 vs lodged 72 would otherwise warn).
+    rebaseliner = CalculatorRebaseliner(_StubCalculator(_sap_result(sap_score=90)))
+    epc = _epc(sap_version=10.2)
+
+    # Act
+    with caplog.at_level(logging.WARNING):
+        rebaseliner.rebaseline(
+            property_id=42,
+            effective_epc=epc,
+            lodged=_lodged(),
+            physical_state_changed=True,
+        )
+
+    # Assert — no divergence log when the calculator output is adopted.
+    assert caplog.records == []
+
+
 def test_a_10_2_cert_logs_divergence_when_the_calculator_disagrees(
     caplog: pytest.LogCaptureFixture,
 ) -> None:

tests/domain/property_baseline/test_portfolio_796_override_rebaseline.py

@@ -0,0 +1,192 @@
+"""Regression: portfolio-796 properties whose Landlord Overrides moved them off
+their lodged headline must rebaseline off the calculator (not echo the lodged
+figure).
+
+These seven properties (a hit-list Khalim raised) each displayed an Effective
+baseline equal to their *lodged accredited* SAP — band B/C — while the modelling
+plan modelled the override-applied picture (often a different band), producing an
+incoherent "already B, post-works C/D" plan. The cause: the rebaseliner kept the
+lodged figure for any SAP >= 10.2 cert, ignoring the overrides (and, for the
+predicted ones, the synthetic headline). The fix: when a physical-state trigger
+fired — Landlord Overrides or Prediction — the calculator output IS the Effective
+baseline, so the displayed baseline and the plan agree.
+
+This test pins, per property, that (1) its real override set still resolves to
+overlays and trips `physical_state_changed`, and (2) the rebaseliner adopts the
+calculator output (reason physical_state_changed / both), not the lodged figure.
+A representative end-to-end case scores a real cert through the live calculator.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+import pytest
+
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+from domain.property.property import Property, PropertyIdentity
+from domain.property_baseline.calculator_rebaseliner import CalculatorRebaseliner
+from domain.property_baseline.performance import lodged_performance
+from domain.sap10_calculator.calculator import Sap10Calculator
+from repositories.property.landlord_override_overlays import overlays_from
+from repositories.property.property_overrides_reader import (
+    ResolvedPropertyOverride,
+    ResolvedPropertyOverrides,
+)
+
+_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples"
+
+
+def _cert() -> EpcPropertyData:
+    """A real, scorable lodged cert used as the base picture (and, for the
+    predicted cases, as a stand-in synthesised EPC)."""
+    raw: dict[str, Any] = json.loads(
+        (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
+    )
+    return EpcPropertyDataMapper.from_api_response(raw)
+
+
+def _overrides(*pairs: tuple[str, str]) -> ResolvedPropertyOverrides:
+    return ResolvedPropertyOverrides(
+        rows=tuple(
+            ResolvedPropertyOverride(
+                override_component=component, building_part=0, override_value=value
+            )
+            for component, value in pairs
+        )
+    )
+
+
+@dataclass(frozen=True)
+class _Case:
+    pid: int
+    source: Literal["lodged", "predicted"]
+    overrides: ResolvedPropertyOverrides
+
+
+# The seven properties' real Landlord Override sets (as stored in
+# property_overrides), captured verbatim from portfolio 796.
+_CASES: tuple[_Case, ...] = (
+    _Case(725634, "lodged", _overrides(
+        ("main_heating_system", "Electric storage heaters, slimline"),
+        ("wall_type", "Cavity wall, as built, no insulation (assumed)"),
+        ("main_fuel", "electricity"),
+        ("water_heating", "Electric immersion, electricity"),
+        ("glazing", "Double glazing, pre-2002"),
+        ("property_type", "House"),
+        ("built_form_type", "Mid-Terrace"),
+        ("construction_age_band", "E"),
+    )),
+    _Case(721534, "lodged", _overrides(
+        ("main_heating_system", "Gas boiler, combi"),
+        ("wall_type", "System built, as built, no insulation (assumed)"),
+        ("main_fuel", "mains gas"),
+        ("water_heating", "From main system, mains gas"),
+        ("glazing", "Double glazing, 2002 or later"),
+        ("property_type", "Flat"),
+        ("built_form_type", "Mid-Terrace"),
+        ("construction_age_band", "K"),
+    )),
+    _Case(721985, "predicted", _overrides(
+        ("main_heating_system", "Gas boiler, combi"),
+        ("wall_type", "Solid brick, as built, no insulation (assumed)"),
+        ("main_fuel", "mains gas"),
+        ("water_heating", "From main system, mains gas"),
+        ("glazing", "Double glazing, 2002 or later"),
+        ("property_type", "Maisonette"),
+        ("built_form_type", "Semi-Detached"),
+        ("construction_age_band", "B"),
+    )),
+    _Case(721993, "predicted", _overrides(
+        ("main_heating_system", "Gas boiler, combi"),
+        ("wall_type", "Solid brick, as built, no insulation (assumed)"),
+        ("main_fuel", "mains gas"),
+        ("water_heating", "From main system, mains gas"),
+        ("glazing", "Single glazing"),
+        ("property_type", "Maisonette"),
+        ("built_form_type", "Semi-Detached"),
+        ("construction_age_band", "B"),
+    )),
+    _Case(735096, "predicted", _overrides(
+        ("main_heating_system", "Gas boiler, regular"),
+        ("wall_type", "Cavity wall, as built, insulated (assumed)"),
+        ("main_fuel", "electricity"),
+        ("water_heating", "Electric immersion, electricity"),
+        ("glazing", "Double glazing, 2002 or later"),
+        ("property_type", "Flat"),
+        ("built_form_type", "Enclosed Mid-Terrace"),
+        ("construction_age_band", "K"),
+    )),
+    _Case(735220, "predicted", _overrides(
+        ("main_heating_system", "Gas boiler, regular"),
+        ("wall_type", "Cavity wall, as built, insulated (assumed)"),
+        ("main_fuel", "electricity"),
+        ("water_heating", "Electric immersion, electricity"),
+        ("glazing", "Double glazing, 2002 or later"),
+        ("property_type", "Flat"),
+        ("built_form_type", "Enclosed Mid-Terrace"),
+        ("construction_age_band", "J"),
+    )),
+    _Case(739060, "predicted", _overrides(
+        ("main_heating_system", "Gas CPSU"),
+        ("wall_type", "System built, as built, no insulation (assumed)"),
+        ("main_fuel", "mains gas (community)"),
+        ("water_heating", "From main system, mains gas"),
+        ("glazing", "Double glazing, pre-2002"),
+        ("property_type", "Flat"),
+        ("built_form_type", "Mid-Terrace"),
+        ("construction_age_band", "K"),
+    )),
+)
+
+
+def _property_for(case: _Case) -> Property:
+    overlays = overlays_from(case.overrides)
+    identity = PropertyIdentity(
+        portfolio_id=796, postcode="A0 0AA", address="", uprn=case.pid
+    )
+    if case.source == "lodged":
+        return Property(identity=identity, epc=_cert(), landlord_overrides=overlays)
+    return Property(
+        identity=identity, epc=None, predicted_epc=_cert(), landlord_overrides=overlays
+    )
+
+
+@pytest.mark.parametrize("case", _CASES, ids=lambda c: str(c.pid))
+def test_override_set_resolves_and_trips_physical_state_changed(case: _Case) -> None:
+    # The stored override values must still resolve to overlays (a renamed enum
+    # value would silently stop overriding), and the resulting Property must trip
+    # the Rebaselining trigger (b)/(c) — the signal that stops the baseline
+    # echoing the lodged headline.
+    overlays = overlays_from(case.overrides)
+    assert overlays, f"property {case.pid}: no override resolved to an overlay"
+    assert _property_for(case).physical_state_changed is True
+
+
+@pytest.mark.parametrize("case", _CASES, ids=lambda c: str(c.pid))
+def test_property_rebaselines_off_the_calculator_not_the_lodged_headline(
+    case: _Case,
+) -> None:
+    # The load-bearing guarantee: with a physical-state change, the Effective
+    # baseline is the calculator's scoring of EPC + overrides — so the displayed
+    # baseline and the modelled plan derive from the same picture and agree.
+    prop = _property_for(case)
+    effective_epc = prop.effective_epc
+    lodged = lodged_performance(effective_epc)
+    rebaseliner = CalculatorRebaseliner(Sap10Calculator())
+
+    result = rebaseliner.rebaseline(
+        case.pid,
+        effective_epc,
+        lodged,
+        physical_state_changed=prop.physical_state_changed,
+    )
+
+    assert result.reason in ("physical_state_changed", "both")
+    assert result.sap_result is not None
+    # Effective is the calculator's output, not the lodged accredited headline.
+    assert result.effective.sap_score == result.sap_result.sap_score

tests/orchestration/test_ara_first_run_pipeline_integration.py

@@ -364,12 +364,18 @@ def test_modelling_optimises_and_persists_a_multi_measure_plan(
     assert plan.cost_of_works is not None
     assert plan.cost_of_works > 0.0
     # Plan-level energy/bill figures derived from the post-package bill vs the
-    # baseline bill at the run's Fuel Rates (ADR-0014 amendment). The package
-    # improves the property, so it consumes less energy and costs less to run.
+    # baseline bill at the run's Fuel Rates (ADR-0014 amendment). The max-gain
+    # fallback package is ASHP-led on a *gas* dwelling whose suspended floor is
+    # correctly modelled as exposed (is_exposed_floor now round-trips through
+    # persistence — #TBD): it saves a large amount of delivered energy, but the
+    # gas→electricity fuel switch trades cheap gas kWh for pricier electricity
+    # kWh, so the *bill* is roughly neutral (marginally negative). The exact
+    # value is pinned by the telescoping cascade below; here we only assert the
+    # figure is produced (sign is not load-bearing for a SAP-max fallback).
     assert plan.post_energy_bill is not None and plan.post_energy_bill > 0.0
     assert plan.post_energy_consumption is not None
     assert plan.post_energy_consumption > 0.0
-    assert plan.energy_bill_savings is not None and plan.energy_bill_savings > 0.0
+    assert plan.energy_bill_savings is not None
     assert plan.energy_consumption_savings is not None
     assert plan.energy_consumption_savings > 0.0
 

tests/orchestration/test_property_baseline_orchestrator.py

@@ -158,7 +158,12 @@ class _ScoringRebaseliner(Rebaseliner):
         self._result = result
 
     def rebaseline(
-        self, property_id: int, effective_epc: EpcPropertyData, lodged: Performance
+        self,
+        property_id: int,
+        effective_epc: EpcPropertyData,
+        lodged: Performance,
+        *,
+        physical_state_changed: bool = False,
     ) -> RebaselineResult:
         return RebaselineResult(
             effective=lodged, reason="none", sap_result=self._result

tests/repositories/epc/test_epc_persistence_field_coverage.py

@@ -1,16 +1,23 @@
-"""Structural guard: every `EpcPropertyData` field must round-trip.
+"""Structural guard: every `EpcPropertyData` field — top-level AND nested — must
+round-trip.
 
 The deep-equality round-trip test (`test_epc_round_trip.py`) only catches a
 dropped field when a *fixture populates it* — the existing gaps (conservatory,
-roof windows, solar-HW collector) were latent precisely because no fixture
-exercised them (see `docs/migrations/epc-property-round-trip-fidelity.md`).
+roof windows, solar-HW collector, **PV arrays**, **floor heat-loss flags**) were
+latent precisely because no fixture exercised them (see
+`docs/migrations/epc-property-round-trip-fidelity.md`).
 
-This guard closes that blind spot structurally: it asserts that every field on
-`EpcPropertyData` is either reconstructed by `EpcPostgresRepository._compose`
-(the DB → domain mapper) or listed — with a reason — in the allow-list below.
-Adding a new domain field therefore forces a conscious persist-or-justify
-decision; persisting a previously-gapped field means deleting its allow-list
-entry. ADR-0036.
+This guard closes that blind spot structurally. It walks every domain dataclass
+reachable from `EpcPropertyData` (transitively, through Optionals and lists) and
+asserts each field is either reconstructed by an `EpcPostgresRepository`
+`_compose` / `_to_*` mapper or listed — with a reason — in the allow-list below.
+The earlier version only inspected `EpcPropertyData`'s *top-level* fields, so a
+dropped field on a *nested* object (`SapEnergySource.photovoltaic_arrays`,
+`SapFloorDimension.is_above_partially_heated_space`) slipped straight through.
+
+Adding a new domain field — at any depth — therefore forces a conscious
+persist-or-justify decision; persisting a previously-gapped field means deleting
+its allow-list entry. ADR-0036.
 """
 
 from __future__ import annotations
@@ -19,74 +26,151 @@ import ast
 import dataclasses
 import inspect
 import textwrap
+import typing
 
 from datatypes.epc.domain.epc_property_data import EpcPropertyData
 from repositories.epc.epc_postgres_repository import EpcPostgresRepository
 
-# Fields deliberately NOT reconstructed by `_compose`, each with its reason.
+# Fields NOT reconstructed by the repo mappers, each with its reason. Keyed by
+# ``ClassName.field`` (a single field) or ``ClassName`` (the whole dataclass is
+# an un-persisted, tracked gap awaiting its FE columns / child table).
 _UNPERSISTED_ALLOWLIST: dict[str, str] = {
-    # Redundant top-level field: the calculator reads
-    # `sap_ventilation.extract_fans_count` (which round-trips via
-    # `_to_ventilation`), never this top-level duplicate.
-    "extract_fans_count": "redundant; scoring uses sap_ventilation.extract_fans_count",
     # Not read by the calculator (dormant); no DB column yet.
-    "air_tightness": "dormant — not read by the calculator; no FE column",
-    "lzc_energy_sources": "dormant — not read by the calculator; no FE column",
-    # Scoring-relevant round-trip gaps awaiting FE columns / child table
+    "EpcPropertyData.air_tightness": "dormant — not read by the calculator; no FE column",
+    "EpcPropertyData.lzc_energy_sources": "dormant — not read by the calculator; no FE column",
+    # Scoring-relevant round-trip gaps awaiting FE columns / child tables
     # (tracked follow-ups; see docs/migrations/epc-property-round-trip-fidelity.md).
-    "sap_roof_windows": "FE child table pending — tracked round-trip gap",
-    "solar_hw_collector_orientation": "FE column pending — tracked round-trip gap",
-    "solar_hw_collector_pitch_deg": "FE column pending — tracked round-trip gap",
-    "solar_hw_overshading": "FE column pending — tracked round-trip gap",
+    "EpcPropertyData.sap_roof_windows": "FE child table pending — tracked round-trip gap",
+    "SapRoofWindow": "FE child table pending — tracked round-trip gap (sap_roof_windows)",
+    "EpcPropertyData.solar_hw_collector_orientation": "FE column pending — tracked round-trip gap",
+    "EpcPropertyData.solar_hw_collector_pitch_deg": "FE column pending — tracked round-trip gap",
+    "EpcPropertyData.solar_hw_overshading": "FE column pending — tracked round-trip gap",
+    "SapAlternativeWall.u_value": "FE column pending — tracked round-trip gap",
+    "SapAlternativeWall.wall_thickness_mm": "FE column pending — tracked round-trip gap",
+    # Room-in-roof detail: only floor_area + age_band reconstruct from the
+    # inlined columns; the geometry / surface detail awaits an FE child table
+    # (tracked round-trip gap, docs/migrations §2).
+    "SapRoomInRoof.common_wall_height_m": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoof.common_wall_length_m": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoof.detailed_surfaces": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoof.gable_1_height_m": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoof.gable_1_length_m": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoof.gable_2_height_m": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoof.gable_2_length_m": "FE child table pending — room-in-roof detail",
+    "SapRoomInRoofSurface": "FE child table pending — room-in-roof surface detail",
+    # --- Nested gaps the calculator does NOT read (dormant); no FE column.
+    "SapAlternativeWall.is_basement": "dormant — not read by the calculator; no FE column",
+    "SapBuildingPart.floor_u_value": "dormant — not read by the calculator (column exists, unread)",
+    "SapBuildingPart.rafter_insulation_thickness": "dormant — not read by the calculator; no FE column",
+    "SapBuildingPart.roof_u_value": "dormant — not read by the calculator; no FE column",
+    "SapBuildingPart.wall_is_basement": "dormant — not read by the calculator; no FE column",
+    "SapBuildingPart.wall_u_value": "dormant — not read by the calculator; no FE column",
+    "SapHeating.cylinder_heat_loss": "dormant — not read by the calculator; no FE column",
 }
 
 
-def _compose_reconstructed_fields() -> set[str]:
-    """The keyword names `_compose` passes to the top-level `EpcPropertyData(...)`
-    constructor — i.e. the fields it reconstructs from the DB row."""
-    source = textwrap.dedent(inspect.getsource(EpcPostgresRepository._compose))
+def _unwrap(annotation: object) -> list[object]:
+    """The concrete types inside an annotation — peeling `Optional`, `Union` and
+    `list[...]` so a field typed `Optional[list[SapWindow]]` yields `SapWindow`."""
+    args = typing.get_args(annotation)
+    if not args:
+        return [annotation]
+    origin = typing.get_origin(annotation)
+    out: list[object] = []
+    if origin is list:
+        out.extend(_unwrap(args[0]))
+    else:  # Union / Optional
+        for arg in args:
+            out.extend(_unwrap(arg))
+    return out
+
+
+def _reachable_dataclasses() -> dict[str, type]:
+    """Every domain dataclass reachable from `EpcPropertyData`, transitively,
+    keyed by class name."""
+    seen: dict[str, type] = {}
+    stack: list[type] = [EpcPropertyData]
+    while stack:
+        cls = stack.pop()
+        if not dataclasses.is_dataclass(cls) or cls.__name__ in seen:
+            continue
+        seen[cls.__name__] = cls
+        for field in dataclasses.fields(cls):
+            for inner in _unwrap(field.type):
+                if isinstance(inner, type) and dataclasses.is_dataclass(inner):
+                    stack.append(inner)
+    return seen
+
+
+def _reconstructed_by_class() -> dict[str, set[str]]:
+    """For each domain-dataclass name, the keyword argument names passed to ANY
+    `ClassName(...)` construction inside `EpcPostgresRepository` — i.e. the fields
+    its `_compose` / `_to_*` mappers reconstruct from the DB rows."""
+    source = textwrap.dedent(inspect.getsource(EpcPostgresRepository))
     tree = ast.parse(source)
+    by_class: dict[str, set[str]] = {}
     for node in ast.walk(tree):
-        if (
-            isinstance(node, ast.Call)
-            and isinstance(node.func, ast.Name)
-            and node.func.id == "EpcPropertyData"
-        ):
-            return {kw.arg for kw in node.keywords if kw.arg is not None}
-    raise AssertionError("no EpcPropertyData(...) construction found in _compose")
+        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
+            kwargs = {kw.arg for kw in node.keywords if kw.arg is not None}
+            by_class.setdefault(node.func.id, set()).update(kwargs)
+    return by_class
 
 
 def test_every_epc_property_data_field_is_persisted_or_allowlisted() -> None:
-    # Arrange
-    all_fields = {f.name for f in dataclasses.fields(EpcPropertyData)}
-    reconstructed = _compose_reconstructed_fields()
+    # Arrange — every field of every reachable domain dataclass, and the fields
+    # the repo mappers reconstruct.
+    reachable = _reachable_dataclasses()
+    reconstructed = _reconstructed_by_class()
 
-    # Act — fields neither reconstructed nor explicitly excused.
-    silently_dropped = all_fields - reconstructed - set(_UNPERSISTED_ALLOWLIST)
+    # Act — fields neither reconstructed nor explicitly excused (per-field or
+    # whole-class allow-list entry).
+    silently_dropped: list[str] = []
+    for name, cls in reachable.items():
+        if name in _UNPERSISTED_ALLOWLIST:  # whole class excused
+            continue
+        recon = reconstructed.get(name, set())
+        for field in dataclasses.fields(cls):
+            key = f"{name}.{field.name}"
+            if field.name not in recon and key not in _UNPERSISTED_ALLOWLIST:
+                silently_dropped.append(key)
 
-    # Assert — nothing falls through the round-trip unnoticed.
-    assert silently_dropped == set(), (
+    # Assert — nothing falls through the round-trip unnoticed, at any depth.
+    assert silently_dropped == [], (
         "EpcPropertyData field(s) are dropped on DB round-trip without a "
         f"documented reason: {sorted(silently_dropped)}. Either reconstruct them "
-        "in EpcPostgresRepository._compose, or add them to "
+        "in EpcPostgresRepository (_compose / a _to_* mapper), or add them to "
         "_UNPERSISTED_ALLOWLIST with a justification."
     )
 
 
 def test_allowlist_has_no_stale_entries() -> None:
     # Arrange
-    all_fields = {f.name for f in dataclasses.fields(EpcPropertyData)}
-    reconstructed = _compose_reconstructed_fields()
+    reachable = _reachable_dataclasses()
+    reconstructed = _reconstructed_by_class()
 
-    # Act — allow-list entries that are now persisted, or no longer fields.
-    redundant = {
-        name
-        for name in _UNPERSISTED_ALLOWLIST
-        if name in reconstructed or name not in all_fields
-    }
+    # Act — allow-list entries that are now fully reconstructed, or name a
+    # class/field that no longer exists.
+    redundant: list[str] = []
+    for entry in _UNPERSISTED_ALLOWLIST:
+        class_name, _, field_name = entry.partition(".")
+        cls = reachable.get(class_name)
+        if cls is None:
+            redundant.append(entry)
+            continue
+        if not field_name:
+            # Whole-class entry: stale once ANY field is reconstructed (the class
+            # is now wired into the round-trip).
+            if reconstructed.get(class_name):
+                redundant.append(entry)
+            continue
+        field_names = {f.name for f in dataclasses.fields(cls)}
+        if field_name not in field_names or field_name in reconstructed.get(
+            class_name, set()
+        ):
+            redundant.append(entry)
 
     # Assert — the allow-list stays honest as gaps are closed.
-    assert redundant == set(), (
+    assert redundant == [], (
         f"stale _UNPERSISTED_ALLOWLIST entries (now persisted or removed): "
         f"{sorted(redundant)} — delete them."
     )

tests/repositories/epc/test_epc_round_trip.py

@@ -86,6 +86,128 @@ def test_non_separated_conservatory_round_trips(db_engine: Engine) -> None:
     assert reloaded == original
 
 
+def test_photovoltaic_arrays_round_trip(db_engine: Engine) -> None:
+    # SAP 10.2 Appendix M — a dwelling's solar PV arrays generate electricity that
+    # offsets demand, worth a large slice of the SAP score (≈12 points on an
+    # electrically-heated dwelling). `sap_energy_source.photovoltaic_arrays` had
+    # no table, so every array was dropped on save (persist != score). We inject
+    # two arrays (distinct, so ORDER is observable) onto a clean PV-free fixture
+    # so the ONLY thing that can break deep-equality is the array list itself.
+    from dataclasses import replace
+
+    from datatypes.epc.domain.epc_property_data import PhotovoltaicArray
+
+    # Arrange — a green fixture with no PV, plus two ordered arrays.
+    original = _load_epc("RdSAP-Schema-21.0.1")
+    assert original.sap_energy_source.photovoltaic_arrays is None, (
+        "fixture must start PV-free so the array list is the only variable"
+    )
+    arrays = [
+        PhotovoltaicArray(peak_power=3.24, pitch=3, overshading=1, orientation=3),
+        PhotovoltaicArray(peak_power=1.5, pitch=2, overshading=2, orientation=None),
+    ]
+    original = replace(
+        original,
+        sap_energy_source=replace(
+            original.sap_energy_source, photovoltaic_arrays=arrays
+        ),
+    )
+
+    # Act
+    with Session(db_engine) as session:
+        epc_property_id = EpcPostgresRepository(session).save(original)
+        session.commit()
+    with Session(db_engine) as session:
+        reloaded = EpcPostgresRepository(session).get(epc_property_id)
+
+    # Assert — both arrays survive in order, deep-equal.
+    assert reloaded == original
+
+
+def test_calculator_read_fields_round_trip(db_engine: Engine) -> None:
+    # Seven fields the SAP calculator reads (cert_to_inputs / heat_transmission)
+    # that had no DB column, so they were silently dropped on save (persist !=
+    # score). FE columns now exist; inject all seven onto a clean fixture and
+    # prove they survive the round-trip deep-equal.
+    from dataclasses import replace
+
+    # Arrange
+    original = _load_epc("RdSAP-Schema-21.0.1")
+    bp0 = original.sap_building_parts[0]
+    assert bp0.sap_alternative_wall_1 is not None, "fixture must carry an alt wall"
+    bp0 = replace(
+        bp0,
+        sap_alternative_wall_1=replace(bp0.sap_alternative_wall_1, is_sheltered=True),
+        wall_insulation_thermal_conductivity=35,  # Union[int,str] → JSONB int
+    )
+    heating = original.sap_heating
+    mh0 = replace(
+        heating.main_heating_details[0],
+        community_heating_boiler_fuel_type=35,
+        community_heating_chp_fraction=0.35,
+    )
+    heating = replace(
+        heating,
+        main_heating_details=[mh0, *heating.main_heating_details[1:]],
+        cylinder_volume_measured_l=180,
+    )
+    original = replace(
+        original,
+        sap_building_parts=[bp0, *original.sap_building_parts[1:]],
+        sap_heating=heating,
+        sap_energy_source=replace(original.sap_energy_source, pv_diverter_present=True),
+        sap_ventilation=replace(
+            original.sap_ventilation, air_permeability_ap50_m3_h_m2=7.5
+        ),
+    )
+
+    # Act
+    with Session(db_engine) as session:
+        epc_property_id = EpcPostgresRepository(session).save(original)
+        session.commit()
+    with Session(db_engine) as session:
+        reloaded = EpcPostgresRepository(session).get(epc_property_id)
+
+    # Assert — all seven calculator-read fields survive, deep-equal.
+    assert reloaded == original
+
+
+def test_floor_dimension_heat_loss_flags_round_trip(db_engine: Engine) -> None:
+    # SAP 10.2 §3.3 — `is_exposed_floor` and `is_above_partially_heated_space`
+    # are heat-loss flags on a `SapFloorDimension`: the calculator routes a floor
+    # over outside air / over a partially-heated space through a different
+    # U-value path. They had no `epc_floor_dimension` column, so a True flag
+    # round-tripped back to the `False` default and silently flipped the floor's
+    # heat loss (persist != score). We force both True on a clean fixture so the
+    # ONLY thing that can break deep-equality is these two flags.
+    from dataclasses import replace
+
+    # Arrange — a green fixture with a floor dimension, both flags forced True.
+    original = _load_epc("RdSAP-Schema-21.0.0")
+    assert original.sap_building_parts, "fixture must have a building part"
+    bp0 = original.sap_building_parts[0]
+    assert bp0.sap_floor_dimensions, "fixture building part must have a floor dimension"
+    dim0 = replace(
+        bp0.sap_floor_dimensions[0],
+        is_exposed_floor=True,
+        is_above_partially_heated_space=True,
+    )
+    bp0 = replace(bp0, sap_floor_dimensions=[dim0, *bp0.sap_floor_dimensions[1:]])
+    original = replace(
+        original, sap_building_parts=[bp0, *original.sap_building_parts[1:]]
+    )
+
+    # Act
+    with Session(db_engine) as session:
+        epc_property_id = EpcPostgresRepository(session).save(original)
+        session.commit()
+    with Session(db_engine) as session:
+        reloaded = EpcPostgresRepository(session).get(epc_property_id)
+
+    # Assert — both flags survive the round-trip, deep-equal.
+    assert reloaded == original
+
+
 def test_building_part_wall_insulation_thickness_preserves_int(
     db_engine: Engine,
 ) -> None: