An expired-source EPC occupies the predicted slot without stranding rows 🟩

EpcSource widens to "expired" (ADR-0054; the column is TEXT — no
migration). "predicted"/"expired" form one slot family: _slot_sources
routes every slot read and slot-clearing delete through the family, so a
re-ingestion flipping the flavour replaces the row instead of stranding
its sibling. FakeEpcRepo mirrors the family.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

repositories/epc/epc_postgres_repository.py

@@ -50,9 +50,20 @@ from infrastructure.postgres.epc_property_table import (
     EpcRenewableHeatIncentiveModel,
     EpcWindowModel,
 )
-from repositories.epc.epc_repository import EpcRepository, EpcSource
+from repositories.epc.epc_repository import (
+    PREDICTED_SLOT_SOURCES,
+    EpcRepository,
+    EpcSource,
+)
 from utilities.private import private
 
+
+def _slot_sources(source: EpcSource) -> tuple[EpcSource, ...]:
+    """The source family sharing `source`'s slot: "predicted" and "expired" are
+    one slot whose flavour a re-ingestion may flip, so every slot read and
+    slot-clearing delete addresses the family (ADR-0054)."""
+    return PREDICTED_SLOT_SOURCES if source in PREDICTED_SLOT_SOURCES else (source,)
+
 _T = TypeVar("_T")
 
 
@@ -320,7 +331,7 @@ class EpcPostgresRepository(EpcRepository):
             for i in self._session.exec(
                 select(EpcPropertyModel.id)
                 .where(col(EpcPropertyModel.property_id).in_(property_ids))
-                .where(EpcPropertyModel.source == source)
+                .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
             ).all()
             if i is not None
         ]
@@ -367,7 +378,7 @@ class EpcPostgresRepository(EpcRepository):
             for i in self._session.exec(
                 select(EpcPropertyModel.id)
                 .where(EpcPropertyModel.property_id == property_id)
-                .where(EpcPropertyModel.source == source)
+                .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
             ).all()
             if i is not None
         ]
@@ -417,7 +428,7 @@ class EpcPostgresRepository(EpcRepository):
         row = self._session.exec(
             select(EpcPropertyModel)
             .where(EpcPropertyModel.property_id == property_id)
-            .where(EpcPropertyModel.source == source)
+            .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
             .order_by(EpcPropertyModel.id)  # type: ignore[arg-type]
         ).first()
         if row is None or row.id is None:
@@ -444,7 +455,7 @@ class EpcPostgresRepository(EpcRepository):
         parents = self._session.exec(
             select(EpcPropertyModel)
             .where(col(EpcPropertyModel.property_id).in_(property_ids))
-            .where(EpcPropertyModel.source == source)
+            .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
             .order_by(EpcPropertyModel.id)  # type: ignore[arg-type]
         ).all()
         parent_by_property: dict[int, EpcPropertyModel] = {}

repositories/epc/epc_repository.py

@@ -9,8 +9,15 @@ if TYPE_CHECKING:
     from repositories.epc.epc_postgres_repository import EpcSaveRequest
 
 # Provenance of a persisted EPC picture (ADR-0031): a real "lodged" EPC, or a
-# "predicted" one synthesised by EPC Prediction. A property can hold one of each.
-EpcSource = Literal["lodged", "predicted"]
+# "predicted" one synthesised by EPC Prediction — "expired" is a prediction
+# enhanced by an expired Historic EPC's stable attributes (ADR-0054). A
+# property holds one lodged picture and one predicted-slot picture.
+EpcSource = Literal["lodged", "predicted", "expired"]
+
+# The predicted slot's source family: "predicted" and "expired" occupy the SAME
+# slot (a re-ingestion may flip the flavour), so slot reads and slot-clearing
+# deletes must always address the family, never one member (ADR-0054).
+PREDICTED_SLOT_SOURCES: tuple[EpcSource, ...] = ("predicted", "expired")
 
 
 class EpcRepository(ABC):

tests/orchestration/fakes.py

@@ -21,7 +21,11 @@ from repositories.epc.epc_postgres_repository import EpcSaveRequest
 from repositories.plan.plan_repository import PlanRepository, PlanSaveRequest
 from repositories.product.product_repository import ProductRepository
 from repositories.property_baseline.property_baseline_repository import PropertyBaselineRepository
-from repositories.epc.epc_repository import EpcRepository, EpcSource
+from repositories.epc.epc_repository import (
+    PREDICTED_SLOT_SOURCES,
+    EpcRepository,
+    EpcSource,
+)
 from repositories.property.property_repository import (
     PropertyIdentityInsert,
     PropertyRepository,
@@ -96,7 +100,7 @@ class FakeEpcRepo(EpcRepository):
         if property_id is not None:
             slot = (
                 self._predicted_by_property
-                if source == "predicted"
+                if source in PREDICTED_SLOT_SOURCES
                 else self._by_property
             )
             slot[property_id] = data