tradein-mvp/backend/app/services/matching/conflict_resolution.py

@@ -113,12 +113,50 @@ LISTING_FIELD_PRIORITY: dict[str, list[str] | str] = {
 }
 
 
+def _freshest_non_null(
+    candidates: dict[str, Any],
+    timestamps: dict[str, Any] | None,
+) -> Any:
+    """Pick the non-null candidate from the freshest source (max last_seen_at).
+
+    Deterministic replacement for dict-insertion-order ``first_non_null`` (#1539).
+    Among sources with a non-null value:
+
+    * if ``timestamps`` is provided, choose the value whose source has the most
+      recent ``last_seen_at``; ties (equal / missing timestamps) break by source
+      name (lexicographic) so the result is reproducible regardless of dict order;
+    * if no usable timestamps are available, fall back to first non-null in
+      sorted-key order (still deterministic, unlike raw insertion order).
+    """
+    non_null = {src: v for src, v in candidates.items() if v is not None}
+    if not non_null:
+        return None
+    timestamps = timestamps or {}
+
+    timed = [src for src in non_null if timestamps.get(src) is not None]
+    if timed:
+        # Freshest last_seen_at wins; equal timestamps tie-break by source name
+        # (lexicographically smallest) so the result never depends on dict order.
+        latest = max(timestamps[src] for src in timed)
+        winner = min(src for src in timed if timestamps[src] == latest)
+        return non_null[winner]
+    # No timestamps anywhere: deterministic by sorted source name.
+    return non_null[min(non_null)]
+
+
 def _resolve(
     priority: dict[str, list[str] | str],
     field: str,
     candidates: dict[str, Any],
+    timestamps: dict[str, Any] | None = None,
 ) -> Any:
-    """Pick value from candidates per priority dict semantics."""
+    """Pick value from candidates per priority dict semantics.
+
+    ``candidates`` maps source name -> value for ``field``. ``timestamps`` maps
+    source name -> ``last_seen_at`` (any comparable, usually ``datetime``); when
+    supplied it makes conflict resolution prefer the freshest source instead of
+    relying on non-deterministic dict insertion order (#1539).
+    """
     if not candidates:
         return None
     rule = priority.get(field, "first_non_null")
@@ -144,10 +182,7 @@ def _resolve(
                 uniq.append(x)
             return uniq
         if rule == "first_non_null":
-            for v in candidates.values():
+            return _freshest_non_null(candidates, timestamps)
-                if v is not None:
-                    return v
-            return None
         if rule == "max":
             non_null = [v for v in candidates.values() if v is not None]
             return max(non_null) if non_null else None
@@ -180,25 +215,42 @@ def _resolve(
                     candidates,
                 )
             return median
-        # Unknown rule
+        # Unknown rule — same deterministic freshness fallback as first_non_null.
-        return next((v for v in candidates.values() if v is not None), None)
+        return _freshest_non_null(candidates, timestamps)
 
-    # Priority list: pick value from highest-ranked source present (non-null)
+    # Priority list: pick value from highest-ranked source present (non-null).
+    # Explicit per-field ranking is authoritative and beats freshness.
     for src in rule:
         if src in candidates and candidates[src] is not None:
             return candidates[src]
-    # Fallback: any non-null
+    # Fallback: no ranked source present → freshest non-null (deterministic).
-    return next((v for v in candidates.values() if v is not None), None)
+    return _freshest_non_null(candidates, timestamps)
 
 
-def resolve_house_field(field: str, candidates: dict[str, Any]) -> Any:
+def resolve_house_field(
-    """Pick canonical house field value per HOUSE_FIELD_PRIORITY."""
+    field: str,
-    return _resolve(HOUSE_FIELD_PRIORITY, field, candidates)
+    candidates: dict[str, Any],
+    timestamps: dict[str, Any] | None = None,
+) -> Any:
+    """Pick canonical house field value per HOUSE_FIELD_PRIORITY.
+
+    ``timestamps`` (source -> last_seen_at) makes first_non_null / fallback
+    resolution prefer the freshest source deterministically (#1539).
+    """
+    return _resolve(HOUSE_FIELD_PRIORITY, field, candidates, timestamps)
 
 
-def resolve_listing_field(field: str, candidates: dict[str, Any]) -> Any:
+def resolve_listing_field(
-    """Pick canonical listing field value per LISTING_FIELD_PRIORITY."""
+    field: str,
-    return _resolve(LISTING_FIELD_PRIORITY, field, candidates)
+    candidates: dict[str, Any],
+    timestamps: dict[str, Any] | None = None,
+) -> Any:
+    """Pick canonical listing field value per LISTING_FIELD_PRIORITY.
+
+    ``timestamps`` (source -> last_seen_at) makes first_non_null / fallback
+    resolution prefer the freshest source deterministically (#1539).
+    """
+    return _resolve(LISTING_FIELD_PRIORITY, field, candidates, timestamps)
 
 
 # ---------------------------------------------------------------------------

tradein-mvp/backend/tests/matching/test_conflict_resolution.py

@@ -1,5 +1,7 @@
 """Per-field priority resolution tests."""
 
+from datetime import datetime
+
 from app.services.matching.conflict_resolution import (
     HOUSE_FIELD_PRIORITY,
     LISTING_FIELD_PRIORITY,
@@ -212,3 +214,116 @@ class TestYandexListingPriority:
         ]
         for key in yandex_keys:
             assert key in HOUSE_FIELD_PRIORITY, f"{key!r} missing from HOUSE_FIELD_PRIORITY"
+
+
+# ---------------------------------------------------------------------------
+# Freshness-based conflict resolution (#1539) — last_seen_at wins, deterministic
+# ---------------------------------------------------------------------------
+
+
+class TestFreshnessResolution:
+    """first_non_null / fallback must prefer the freshest source, not dict order."""
+
+    def test_first_non_null_fresher_source_wins(self) -> None:
+        # kadastr_num uses the 'first_non_null' rule.
+        candidates = {"avito": "66:1:1:OLD", "cian": "66:1:1:NEW"}
+        timestamps = {
+            "avito": datetime(2026, 1, 1),
+            "cian": datetime(2026, 6, 1),  # fresher
+        }
+        out = resolve_listing_field("kadastr_num", candidates, timestamps)
+        assert out == "66:1:1:NEW"
+
+    def test_first_non_null_fresher_source_wins_regardless_of_order(self) -> None:
+        # Same data, opposite insertion order — result must not flip.
+        candidates = {"cian": "66:1:1:NEW", "avito": "66:1:1:OLD"}
+        timestamps = {
+            "avito": datetime(2026, 1, 1),
+            "cian": datetime(2026, 6, 1),
+        }
+        out = resolve_listing_field("kadastr_num", candidates, timestamps)
+        assert out == "66:1:1:NEW"
+
+    def test_freshness_picks_avito_when_avito_is_newer(self) -> None:
+        candidates = {"avito": "AVITO", "cian": "CIAN"}
+        timestamps = {
+            "avito": datetime(2026, 6, 1),  # fresher
+            "cian": datetime(2026, 1, 1),
+        }
+        out = resolve_listing_field("kadastr_num", candidates, timestamps)
+        assert out == "AVITO"
+
+    def test_null_only_returns_none(self) -> None:
+        out = resolve_listing_field(
+            "kadastr_num",
+            {"avito": None, "cian": None},
+            {"avito": datetime(2026, 6, 1), "cian": datetime(2026, 1, 1)},
+        )
+        assert out is None
+
+    def test_null_source_skipped_even_if_fresher(self) -> None:
+        # cian is fresher but null → avito's non-null value wins.
+        candidates = {"avito": "VALUE", "cian": None}
+        timestamps = {
+            "avito": datetime(2026, 1, 1),
+            "cian": datetime(2026, 6, 1),
+        }
+        out = resolve_listing_field("kadastr_num", candidates, timestamps)
+        assert out == "VALUE"
+
+    def test_tie_breaks_by_source_name_deterministic(self) -> None:
+        # Equal timestamps → lexicographically smallest source name wins.
+        ts = datetime(2026, 6, 1)
+        out1 = resolve_listing_field(
+            "kadastr_num",
+            {"avito": "A", "cian": "C"},
+            {"avito": ts, "cian": ts},
+        )
+        out2 = resolve_listing_field(
+            "kadastr_num",
+            {"cian": "C", "avito": "A"},
+            {"cian": ts, "avito": ts},
+        )
+        assert out1 == "A"  # "avito" < "cian"
+        assert out1 == out2  # order-independent
+
+    def test_no_timestamps_falls_back_deterministically(self) -> None:
+        # Without timestamps, fallback is deterministic by sorted source name.
+        out1 = resolve_listing_field("kadastr_num", {"avito": "A", "cian": "C"})
+        out2 = resolve_listing_field("kadastr_num", {"cian": "C", "avito": "A"})
+        assert out1 == "A"
+        assert out1 == out2
+
+    def test_partial_timestamps_timed_source_preferred(self) -> None:
+        # Only one source carries a timestamp → it is treated as the freshest.
+        candidates = {"avito": "A", "cian": "C"}
+        timestamps = {"cian": datetime(2026, 6, 1)}  # avito has none
+        out = resolve_listing_field("kadastr_num", candidates, timestamps)
+        assert out == "C"
+
+    def test_house_unknown_field_freshness(self) -> None:
+        # Unknown house field also uses freshness fallback.
+        candidates = {"src_a": 1, "src_b": 2}
+        timestamps = {"src_a": datetime(2026, 1, 1), "src_b": datetime(2026, 6, 1)}
+        assert resolve_house_field("totally_unknown", candidates, timestamps) == 2
+
+    def test_priority_list_fallback_uses_freshness(self) -> None:
+        # No ranked source present for 'lat' → freshest non-null wins.
+        candidates = {"unranked_old": 56.80, "unranked_new": 56.90}
+        timestamps = {
+            "unranked_old": datetime(2026, 1, 1),
+            "unranked_new": datetime(2026, 6, 1),
+        }
+        out = resolve_house_field("lat", candidates, timestamps)
+        assert out == 56.90
+
+    def test_priority_list_ranking_beats_freshness(self) -> None:
+        # Explicit ranking is authoritative: cian_serp ranked above yandex even if
+        # yandex is fresher.
+        candidates = {"cian_serp": 56.83, "yandex_realty_nb": 56.85}
+        timestamps = {
+            "cian_serp": datetime(2026, 1, 1),
+            "yandex_realty_nb": datetime(2026, 6, 1),  # fresher but lower-ranked
+        }
+        out = resolve_house_field("lat", candidates, timestamps)
+        assert out == 56.83