Break lodged-UPRN duplicates by latest inspection_date, then id

inspection_date is non-null on epc_property, so it is a better winner for
a shared-UPRN collision than insertion order: the latest-surveyed lodged
row wins. id DESC stays as the deterministic secondary, so a same-UPRN,
same-inspection_date tie falls to the highest id.

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

repositories/epc/epc_postgres_repository.py

@@ -418,14 +418,18 @@ class EpcPostgresRepository(EpcRepository):
 
     def get_for_property(self, uprn: Optional[int]) -> Optional[EpcPropertyData]:
         # UPRN is not unique on epc_property, so the tie-break is load-bearing: the
-        # most recently ingested (highest-id) lodged row wins.
+        # latest-surveyed lodged row wins (inspection_date is non-null), with id as
+        # the deterministic secondary.
         if uprn is None:
             return None
         row = self._session.exec(
             select(EpcPropertyModel)
             .where(EpcPropertyModel.uprn == uprn)
             .where(col(EpcPropertyModel.source).in_(_slot_sources("lodged")))
-            .order_by(col(EpcPropertyModel.id).desc())
+            .order_by(
+                col(EpcPropertyModel.inspection_date).desc(),
+                col(EpcPropertyModel.id).desc(),
+            )
         ).first()
         if row is None or row.id is None:
             return None
@@ -450,10 +454,12 @@ class EpcPostgresRepository(EpcRepository):
             select(EpcPropertyModel)
             .where(col(EpcPropertyModel.uprn).in_(uprns))
             .where(col(EpcPropertyModel.source).in_(_slot_sources("lodged")))
-            .order_by(col(EpcPropertyModel.id).desc())
+            .order_by(
+                col(EpcPropertyModel.inspection_date).desc(),
+                col(EpcPropertyModel.id).desc(),
+            )
         ).all()
-        # UPRN is not unique, so setdefault keeps the most recently ingested
-        # (highest-id) row per UPRN.
+        # UPRN is not unique, so setdefault keeps the latest-surveyed row per UPRN.
         by_uprn: dict[int, EpcPropertyModel] = {}
         for parent in parents:
             if parent.uprn is not None and parent.id is not None:

tests/repositories/epc/test_epc_lodged_uprn_collision.py

@@ -1,10 +1,12 @@
 """Lodged EPC reads key on UPRN, which is not unique on epc_property. When several
-lodged rows share a UPRN, the most recently ingested (highest-id) row wins."""
+lodged rows share a UPRN, the latest-surveyed (most recent inspection_date) row
+wins — regardless of insertion order."""
 
 from __future__ import annotations
 
 import json
 from dataclasses import replace
+from datetime import date
 from pathlib import Path
 from typing import Any
 
@@ -28,11 +30,19 @@ def _load_epc() -> EpcPropertyData:
 def _seed_shared_uprn(db_engine: Engine) -> None:
     # Two lodged EPCs share UPRN 500 under different property_ids. The write path
     # dedupes per property_id, not UPRN, so both rows persist — a real collision.
+    # The latest-surveyed EPC is saved FIRST (lower id) so a passing test proves
+    # inspection_date, not insertion order, decides the winner.
     epc = _load_epc()
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
-        repo.save(replace(epc, uprn=500, status="older"), property_id=100)
-        repo.save(replace(epc, uprn=500, status="newer"), property_id=101)
+        repo.save(
+            replace(epc, uprn=500, inspection_date=date(2022, 6, 1), status="newer"),
+            property_id=100,
+        )
+        repo.save(
+            replace(epc, uprn=500, inspection_date=date(2018, 6, 1), status="older"),
+            property_id=101,
+        )
         session.commit()