Prefer the newer of a stored survey and a fetched gov cert 🟥

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

applications/modelling_e2e/handler.py

@@ -387,6 +387,15 @@ def _dedupe_skipped(
     return unique
 
 
+def _newer_lodged(
+    stored: Optional[EpcPropertyData], fetched: Optional[EpcPropertyData]
+) -> tuple[Optional[EpcPropertyData], bool]:
+    """ADR-0001 Recency Tie-Break over the two lodged sources. Returns
+    (chosen, chosen_is_fetched) — the bool gates the EPC save, so a stored EPC
+    that wins is never re-persisted."""
+    raise NotImplementedError
+
+
 def _predict_epc(
     *,
     property_id: int,

tests/applications/modelling_e2e/test_handler.py

@@ -7,7 +7,11 @@ is needed. One test per distinct behaviour path.
 
 from __future__ import annotations
 
+import dataclasses
+import json
 from contextlib import ExitStack
+from datetime import date
+from pathlib import Path
 from typing import Any, Iterator, Optional
 from unittest.mock import MagicMock, call, patch
 from uuid import UUID, uuid4
@@ -18,6 +22,8 @@ from pydantic import ValidationError
 from applications.modelling_e2e.modelling_e2e_trigger_body import (
     ModellingE2ETriggerBody,
 )
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.mapper import EpcPropertyDataMapper
 from domain.tasks.subtasks import SubTask
 from repositories.epc.epc_postgres_repository import EpcSaveRequest
 
@@ -1403,9 +1409,38 @@ def test_cohort_cache_prevents_duplicate_candidates_for_calls() -> None:
 
 
 # ---------------------------------------------------------------------------
-# refetch_epc flag
+# Recency Tie-Break between the two lodged sources (ADR-0001)
 # ---------------------------------------------------------------------------
 
+_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples"
+
+
+def _lodged_epc(inspection: str) -> EpcPropertyData:
+    """A real lodged EPC surveyed on `inspection` — the tie-break reads only
+    inspection_date, but a real cert keeps the test honest about the type."""
+    raw: dict[str, Any] = json.loads(
+        (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
+    )
+    epc = EpcPropertyDataMapper.from_api_response(raw)
+    return dataclasses.replace(epc, inspection_date=date.fromisoformat(inspection))
+
+
+def test_stored_survey_newer_than_gov_cert_wins_the_tie_break() -> None:
+    """The bug in #1589: a newer stored survey (e.g. PasHub site notes) must beat
+    an older gov-register cert, and must not be flagged for re-persisting."""
+    # Arrange
+    from applications.modelling_e2e.handler import _newer_lodged
+
+    stored = _lodged_epc("2026-05-01")
+    fetched = _lodged_epc("2023-12-01")
+
+    # Act
+    chosen, chosen_is_fetched = _newer_lodged(stored, fetched)
+
+    # Assert
+    assert chosen is stored
+    assert chosen_is_fetched is False
+
 
 def test_refetch_epc_false_with_stored_epc_skips_api_call() -> None:
     """refetch_epc=False + stored lodged EPC present: EpcClientService.get_by_uprn