tradein-mvp/backend/app/services/house_dedup_merge.py

@@ -29,6 +29,26 @@ WHAT this is:
     250 m of the keeper (ST_DistanceSphere on the WGS84 geom). Same-canon houses >250 m away — or
     with NULL geom on either side — are left as separate rows (conservative; never over-merges).
 
+TWO PASSES (2026-07-02 follow-up): the SAME cluster→keeper→re-point→carry-identity→delete
+    pipeline now runs TWICE inside one transaction, parametrised by the cluster-key expression
+    (`_mapping_sql`) — no 300-line copy-paste:
+      1. FIAS pass — clusters by lower(NULLIF(house_fias_id, '')) (the ФИАС/ГАР building UUID,
+         DaData-backfilled 2026-07-02 for 5 122 houses). Only non-null fias shared by >1 house_id
+         cluster. Catches duplicates the canon MISSES: slash-collapse («Сулимова, 32» vs
+         «Сулимова, 3/2» → same canon «сулимова32») and посёлок truncation (Кедровка/Шувакиш
+         colliding with a same-named ЕКБ street).
+      2. CANON pass — the canonical-address clustering above, now with a CROSS-FIAS GUARD: within a
+         canon cluster a loser is NOT merged when it AND the keeper both carry a non-null but
+         DIFFERENT house_fias_id — provably different buildings the canon collapsed (the
+         slash-collapse class). Critical anti-over-merge fix. The 250 m geo guard applies to both
+         passes (after the backfill, fias rows almost always have geom → conservative).
+
+IDENTITY CARRY-OVER (each pass, BEFORE deleting losers): the keeper's NULL identity / geo-QC
+    fields are filled from its losers with COALESCE semantics (keeper value wins; donor = the
+    loser with geom, then min id): house_fias_id, cadastral_number, gar_house_guid, gar_flat_count,
+    gar_matched_at, gar_match_method, dadata_qc_geo, dadata_qc_house, dadata_enriched_at. Without
+    this the merge would silently drop DaData/ГАР enrichment that lived only on a loser row.
+
 KEEPER RULE (one canonical row per cluster), in priority order:
     1. geom NOT NULL first      (a geocoded row is the better survivor)
     2. most linked listings     (the row the corpus already points at)
@@ -83,6 +103,7 @@ from __future__ import annotations
 import logging
 import time
 from dataclasses import dataclass, field
+from typing import Any
 
 from sqlalchemy import text
 from sqlalchemy.orm import Session
@@ -113,21 +134,47 @@ _KEEPER_ORDER = f"""
     h.id ASC
 """
 
-# Build the loser→keeper mapping as a TEMP table for this transaction. Only addresses shared
+# Cluster-key expressions, one per merge pass. The pipeline (build mapping → keeper → re-point
-# by >1 house_id are clustered; the keeper is rn=1 per cluster, losers are rn>1.
+# children → carry identity → delete losers → backfill) is IDENTICAL for both passes; only the
-_BUILD_MAPPING_SQL = text(
+# cluster key differs, parametrised into `_mapping_sql`, so there is NO copy-paste of the pipeline.
-    f"""
+#
-    CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
+# FIAS key (#1772 follow-up, 2026-07-02): the ФИАС/ГАР building UUID (DaData backfill). Catches
-    WITH clustered AS (
+# duplicates the canon misses — slash-collapse («Сулимова, 32» vs «Сулимова, 3/2») and посёлок
-        SELECT
+# truncation (Кедровка/Шувакиш). Only non-null fias clusters.
-            id,
+_FIAS_KEY_EXPR = """
+            CASE WHEN NULLIF(house_fias_id, '') IS NOT NULL
+                 THEN 'fias:' || lower(house_fias_id)
+                 END
+"""
+
+# CANON key: canonical address over the CLEAN address (short_address→full_address→address). The
+# canon collapses spelling/район variants of the SAME building (ул→улица, strips город/район noise,
+# preserves корпус); the digit guard drops street-less degenerate canons; blank canons never join.
+_CANON_KEY_EXPR = """
             CASE WHEN NULLIF(
                        tradein_canon_addr(COALESCE(short_address, full_address, address)), ''
                      ) IS NOT NULL
                   AND tradein_canon_addr(COALESCE(short_address, full_address, address)) ~ '[0-9]'
                  THEN 'addr:'
                       || tradein_canon_addr(COALESCE(short_address, full_address, address))
-                 END AS cluster_key
+                 END
+"""
+
+
+def _mapping_sql(cluster_key_case: str) -> str:
+    """Render the loser→keeper mapping SQL for one pass, given its cluster-key CASE expression.
+
+    Only cluster keys shared by >1 house_id form a cluster; the keeper is rn=1 per cluster, losers
+    are rn>1. The final mapping is filtered by the 250 m geo guard AND the CROSS-FIAS guard (both
+    no-ops for the fias pass, where every clustered row shares one fias by construction).
+    `cluster_key_case` is a STATIC module constant (never runtime data) — no value injection.
+    """
+    return f"""
+    CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
+    WITH clustered AS (
+        SELECT
+            id,
+            {cluster_key_case} AS cluster_key
         FROM houses
     ),
     clusters_with_count AS (
@@ -155,6 +202,7 @@ _BUILD_MAPPING_SQL = text(
             dh.cluster_key,
             lower(trim(COALESCE(h.short_address, h.full_address, h.address))) AS norm_address,
             h.geom AS loser_geom,
+            h.house_fias_id AS loser_fias,
             ROW_NUMBER() OVER (
                 PARTITION BY dh.cluster_key
                 ORDER BY {_KEEPER_ORDER}
@@ -166,7 +214,11 @@ _BUILD_MAPPING_SQL = text(
             first_value(h.geom) OVER (
                 PARTITION BY dh.cluster_key
                 ORDER BY {_KEEPER_ORDER}
-            ) AS keeper_geom
+            ) AS keeper_geom,
+            first_value(h.house_fias_id) OVER (
+                PARTITION BY dh.cluster_key
+                ORDER BY {_KEEPER_ORDER}
+            ) AS keeper_fias
         FROM dup_houses dh
         JOIN houses h ON h.id = dh.id
         LEFT JOIN listing_counts lc ON lc.house_id = dh.id
@@ -179,6 +231,10 @@ _BUILD_MAPPING_SQL = text(
     -- the same canon-cluster but >250 m away, or with NULL geom on either side, are left as
     -- separate rows (conservative — never over-merges). Cluster_key также требует цифру в canon
     -- (номер дома) — исключает вырожденные canon вида «екатеринбург»/«сооружение».
+    -- CROSS-FIAS guard (#1772 follow-up): never merge two rows that BOTH carry a non-null but
+    -- DIFFERENT house_fias_id — they are provably different buildings the cluster key happened to
+    -- collapse (canon slash-collapse «Сулимова, 32»/«Сулимова, 3/2»). No-op for the fias pass (one
+    -- fias per cluster) and for canon clusters where at most one side has a fias.
     SELECT id AS loser_id, keeper_id, norm_address
     FROM ranked
     WHERE rn > 1
@@ -186,8 +242,18 @@ _BUILD_MAPPING_SQL = text(
       AND keeper_geom IS NOT NULL
       AND loser_geom IS NOT NULL
       AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250
+      AND NOT (
+          NULLIF(loser_fias, '') IS NOT NULL
+          AND NULLIF(keeper_fias, '') IS NOT NULL
+          AND lower(loser_fias) <> lower(keeper_fias)
+      )
     """
-)
+
+
+# Canon-pass mapping — kept as a module constant for backward-compatible introspection/tests.
+_BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR))
+# Fias-pass mapping — same pipeline, clustered by the ФИАС building UUID (runs first).
+_BUILD_MAPPING_SQL_FIAS = text(_mapping_sql(_FIAS_KEY_EXPR))
 
 # Each step keys off _1772_dup_mapping → empty mapping ⇒ 0 rows touched ⇒ idempotent no-op.
 _STEPS: list[tuple[str, str]] = [
@@ -430,6 +496,63 @@ _BACKFILL_ALIASES_SQL = text(
     """
 )
 
+# Carry identity / geo-QC / ГАР-enrichment onto the keeper BEFORE deleting losers (#1772
+# follow-up). COALESCE — the keeper's own populated value always wins; a NULL keeper field is
+# filled from a donor loser. Donor = the loser with geom first, then min id (deterministic):
+# array_agg(... ORDER BY geom-present DESC, id ASC) FILTER (WHERE <field> NOT NULL) [1] takes the
+# first non-null value in that order. Without this the merge would drop DaData/ГАР enrichment that
+# lived only on a loser row (e.g. house_fias_id backfilled onto the loser but not the keeper).
+_CARRY_OVER_IDENTITY_SQL = text(
+    """
+    UPDATE houses k
+    SET
+        house_fias_id      = COALESCE(k.house_fias_id, d.house_fias_id),
+        cadastral_number   = COALESCE(k.cadastral_number, d.cadastral_number),
+        gar_house_guid     = COALESCE(k.gar_house_guid, d.gar_house_guid),
+        gar_flat_count     = COALESCE(k.gar_flat_count, d.gar_flat_count),
+        gar_matched_at     = COALESCE(k.gar_matched_at, d.gar_matched_at),
+        gar_match_method   = COALESCE(k.gar_match_method, d.gar_match_method),
+        dadata_qc_geo      = COALESCE(k.dadata_qc_geo, d.dadata_qc_geo),
+        dadata_qc_house    = COALESCE(k.dadata_qc_house, d.dadata_qc_house),
+        dadata_enriched_at = COALESCE(k.dadata_enriched_at, d.dadata_enriched_at)
+    FROM (
+        SELECT
+            m.keeper_id,
+            (array_agg(l.house_fias_id
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.house_fias_id IS NOT NULL))[1] AS house_fias_id,
+            (array_agg(l.cadastral_number
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.cadastral_number IS NOT NULL))[1] AS cadastral_number,
+            (array_agg(l.gar_house_guid
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.gar_house_guid IS NOT NULL))[1] AS gar_house_guid,
+            (array_agg(l.gar_flat_count
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.gar_flat_count IS NOT NULL))[1] AS gar_flat_count,
+            (array_agg(l.gar_matched_at
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.gar_matched_at IS NOT NULL))[1] AS gar_matched_at,
+            (array_agg(l.gar_match_method
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.gar_match_method IS NOT NULL))[1] AS gar_match_method,
+            (array_agg(l.dadata_qc_geo
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.dadata_qc_geo IS NOT NULL))[1] AS dadata_qc_geo,
+            (array_agg(l.dadata_qc_house
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.dadata_qc_house IS NOT NULL))[1] AS dadata_qc_house,
+            (array_agg(l.dadata_enriched_at
+                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
+                FILTER (WHERE l.dadata_enriched_at IS NOT NULL))[1] AS dadata_enriched_at
+        FROM _1772_dup_mapping m
+        JOIN houses l ON l.id = m.loser_id
+        GROUP BY m.keeper_id
+    ) d
+    WHERE k.id = d.keeper_id
+    """
+)
+
 
 @dataclass
 class DedupMergeResult:
@@ -455,14 +578,83 @@ class DedupMergeResult:
         }
 
 
-def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
+def _run_merge_pass(
-    """Cluster houses by canonical address and merge duplicates onto one canonical keeper.
+    db: Session,
+    *,
+    build_sql: Any,
+    pass_label: str,
+    result: DedupMergeResult,
+) -> None:
+    """Run ONE merge pass (fias- or canon-key) inside the caller's open transaction.
 
-    Re-implements migration 108's proven collision-safe pipeline as a RECURRING job:
+    Builds a fresh loser→keeper mapping for this pass's cluster key, re-points every FK child
-        cluster (canon address + 250 m geo guard) → pick keeper →
+    (UNIQUE-collision-safe), carries identity/enrichment onto the keeper, deletes the losers and
-        re-point children (UNIQUE-collision-safe) → delete losers → backfill sources/aliases.
+    backfills sources/aliases. Accumulates counters onto `result`. NEVER commits/rolls back — the
-    All in ONE transaction. dry_run=True computes counts then ROLLS BACK (no writes).
+    caller owns the single transaction wrapping both passes.
-    Idempotent: a clean table yields an empty mapping → every statement is a 0-row no-op.
+    """
+    # Fresh mapping for this pass. ON COMMIT DROP only fires at txn end, so drop the temp table
+    # explicitly — the second pass must rebuild the same-named table within the one transaction.
+    db.execute(text("DROP TABLE IF EXISTS _1772_dup_mapping"))
+    db.execute(build_sql)
+
+    mapping = db.execute(
+        text(
+            "SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping "
+            "ORDER BY keeper_id, loser_id"
+        )
+    ).all()
+    if not mapping:
+        logger.info("merge_duplicate_houses: pass=%s no clusters found (no-op)", pass_label)
+        return
+
+    result.losers_deleted += len(mapping)
+    result.clusters_merged += len({row.keeper_id for row in mapping})
+
+    # Audit log: every loser→keeper move with its address, for traceability.
+    for row in mapping:
+        logger.info(
+            "merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r",
+            pass_label,
+            row.loser_id,
+            row.keeper_id,
+            row.norm_address,
+        )
+
+    for label, sql in _STEPS:
+        res = db.execute(text(sql))
+        rowcount = res.rowcount or 0
+        if label == "listings":
+            result.listings_repointed += rowcount
+        elif label.endswith("(collision-delete)") or label.endswith("(dedup)"):
+            result.children_deleted += rowcount
+        elif label.endswith("(re-point)") or label in (
+            "house_placement_history",
+            "house_reviews",
+            "house_reliability_checks",
+            "external_valuations",
+        ):
+            result.children_repointed += rowcount
+        logger.debug("merge_duplicate_houses: pass=%s step=%s rows=%d", pass_label, label, rowcount)
+
+    # Carry identity/enrichment onto the keeper BEFORE the losers vanish, then delete + backfill.
+    db.execute(_CARRY_OVER_IDENTITY_SQL)
+    db.execute(_DELETE_LOSERS_SQL)
+    db.execute(_BACKFILL_SOURCES_SQL)
+    db.execute(_BACKFILL_ALIASES_SQL)
+
+
+def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
+    """Cluster houses by fias UUID, then by canonical address, merging dups onto one keeper.
+
+    Re-implements migration 108's proven collision-safe pipeline as a RECURRING TWO-PASS job:
+        1. FIAS pass  — cluster by lower(NULLIF(house_fias_id, '')) (catches slash-collapse /
+           посёлок canon bugs the address canon misses).
+        2. CANON pass — cluster by canonical address (250 m geo guard + cross-fias anti-over-merge
+           guard).
+    Each pass: pick keeper → re-point children (UNIQUE-collision-safe) → carry identity onto the
+    keeper → delete losers → backfill sources/aliases. BOTH passes run in ONE transaction.
+    dry_run=True computes counts then ROLLS BACK (no writes). Idempotent: a clean table yields an
+    empty mapping in each pass → every statement is a 0-row no-op.
 
     Returns the counter dict (DedupMergeResult.to_counters()).
     """
@@ -470,56 +662,21 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i
     result = DedupMergeResult(dry_run=dry_run)
 
     try:
-        db.execute(_BUILD_MAPPING_SQL)
+        # Pass 1: cluster by the ФИАС building UUID (runs first — most precise building identity).
+        _run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL_FIAS, pass_label="fias", result=result)
+        # Pass 2: cluster by canonical address, with the cross-fias anti-over-merge guard.
+        _run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL, pass_label="canon", result=result)
 
-        mapping = db.execute(
+        if result.losers_deleted == 0:
-            text(
+            # Clean table — both passes empty. Roll back (we only opened temp tables).
-                "SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping "
-                "ORDER BY keeper_id, loser_id"
-            )
-        ).all()
-        result.losers_deleted = len(mapping)
-        result.clusters_merged = len({row.keeper_id for row in mapping})
-
-        if not mapping:
-            # Clean table — nothing to merge. Still roll back (we only opened a temp table).
             db.rollback()
             result.duration_sec = time.monotonic() - start
             logger.info(
-                "merge_duplicate_houses: no duplicate-address clusters found (no-op) dry_run=%s",
+                "merge_duplicate_houses: no duplicate clusters found (no-op) dry_run=%s",
                 dry_run,
             )
             return result.to_counters()
 
-        # Audit log: every loser→keeper move with its address, for traceability.
-        for row in mapping:
-            logger.info(
-                "merge_duplicate_houses: merge loser_id=%d → keeper_id=%d address=%r",
-                row.loser_id,
-                row.keeper_id,
-                row.norm_address,
-            )
-
-        for label, sql in _STEPS:
-            res = db.execute(text(sql))
-            rowcount = res.rowcount or 0
-            if label == "listings":
-                result.listings_repointed = rowcount
-            elif label.endswith("(collision-delete)") or label.endswith("(dedup)"):
-                result.children_deleted += rowcount
-            elif label.endswith("(re-point)") or label in (
-                "house_placement_history",
-                "house_reviews",
-                "house_reliability_checks",
-                "external_valuations",
-            ):
-                result.children_repointed += rowcount
-            logger.debug("merge_duplicate_houses: step=%s rows=%d", label, rowcount)
-
-        db.execute(_DELETE_LOSERS_SQL)
-        db.execute(_BACKFILL_SOURCES_SQL)
-        db.execute(_BACKFILL_ALIASES_SQL)
-
         if dry_run:
             db.rollback()
             logger.info(

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -32,6 +32,8 @@ _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
 _MIGRATION_135 = _SQL_DIR / "135_scrape_schedules_seed_house_dedup_merge.sql"
 
 _MAPPING_SQL = str(hdm._BUILD_MAPPING_SQL.text)
+_FIAS_MAPPING_SQL = str(hdm._BUILD_MAPPING_SQL_FIAS.text)
+_CARRY_SQL = str(hdm._CARRY_OVER_IDENTITY_SQL.text)
 _STEP_SQL = {label: sql for label, sql in hdm._STEPS}
 _ALL_STEP_SQL = "\n".join(sql for _, sql in hdm._STEPS)
 _SERVICE_SRC = inspect.getsource(hdm)
@@ -235,6 +237,104 @@ def test_no_fstring_or_format_in_parametrised_sql() -> None:
     assert ".format(" not in _SERVICE_SRC
 
 
+# ── Fias-key pass + cross-fias guard + identity carry-over (#1772 follow-up) ──
+
+
+def test_fias_pass_clusters_by_house_fias_id() -> None:
+    """A second cluster key: lower(NULLIF(house_fias_id,'')) — the ФИАС/ГАР building UUID.
+
+    The fias pass catches duplicates the address canon misses (slash-collapse «Сулимова, 32» vs
+    «Сулимова, 3/2»; посёлок truncation). Only non-null fias shared by >1 house_id clusters.
+    """
+    flat = _flat(_FIAS_MAPPING_SQL)
+    assert "'fias:' || lower(house_fias_id)" in flat
+    assert "NULLIF(house_fias_id, '') IS NOT NULL" in flat
+    # same collision-safe pipeline (same temp table + >1 grouping), NOT the canon cluster key.
+    assert "CREATE TEMP TABLE _1772_dup_mapping" in flat
+    assert "HAVING count(*) > 1" in flat
+    # the fias pass keys on the fias UUID, never on the 'addr:' canon key.
+    assert "'addr:'" not in flat
+
+
+def test_both_passes_share_one_pipeline_no_copy_paste() -> None:
+    """The two mappings differ ONLY in the cluster-key expression (parametrised, no copy-paste)."""
+    canon = _flat(_MAPPING_SQL)
+    fias = _flat(_FIAS_MAPPING_SQL)
+    assert "'addr:'" in canon and "'addr:'" not in fias
+    assert "'fias:'" in fias and "'fias:'" not in canon
+    # both carry the same downstream skeleton (keeper rule, geo guard, cross-fias guard).
+    for token in (
+        "clusters_with_count",
+        "listing_counts",
+        "ranked AS",
+        "AS keeper_geom",
+        "ST_DistanceSphere(loser_geom, keeper_geom) <= 250",
+    ):
+        assert token in canon and token in fias
+
+
+def test_cross_fias_guard_blocks_slash_collapse_over_merge() -> None:
+    """Canon pass must NOT merge two rows that BOTH carry a non-null but DIFFERENT house_fias_id.
+
+    This is the critical anti-over-merge fix for the canon slash-collapse class: «Сулимова, 32»
+    and «Сулимова, 3/2» share the canon «сулимова32» but are different buildings — their distinct
+    fias UUIDs veto the merge.
+    """
+    flat = _flat(_MAPPING_SQL)
+    assert "h.house_fias_id AS loser_fias" in flat
+    assert "AS keeper_fias" in flat
+    assert "first_value(h.house_fias_id) OVER" in flat
+    # the guard: NOT (both non-null AND lower-different).
+    assert "NULLIF(loser_fias, '') IS NOT NULL" in flat
+    assert "NULLIF(keeper_fias, '') IS NOT NULL" in flat
+    assert "lower(loser_fias) <> lower(keeper_fias)" in flat
+
+
+def test_identity_carry_over_fills_all_enrichment_fields() -> None:
+    """Before deleting losers, the keeper's NULL identity / geo-QC fields are filled from losers."""
+    flat = _flat(_CARRY_SQL)
+    assert "UPDATE houses k" in flat
+    for col in (
+        "house_fias_id",
+        "cadastral_number",
+        "gar_house_guid",
+        "gar_flat_count",
+        "gar_matched_at",
+        "gar_match_method",
+        "dadata_qc_geo",
+        "dadata_qc_house",
+        "dadata_enriched_at",
+    ):
+        # COALESCE keeper-first ⇒ a populated keeper field is NEVER overwritten by a loser.
+        assert f"COALESCE(k.{col}, d.{col})" in flat, f"identity field {col} not carried over"
+    # deterministic donor selection: loser with geom first, then min id.
+    assert "ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC" in flat
+    assert "FROM _1772_dup_mapping m" in flat
+    assert "JOIN houses l ON l.id = m.loser_id" in flat
+
+
+def test_carry_over_runs_before_delete_losers_in_each_pass() -> None:
+    """Identity carry-over must execute BEFORE the loser rows are deleted, in every pass."""
+    src = inspect.getsource(hdm._run_merge_pass)
+    assert "_CARRY_OVER_IDENTITY_SQL" in src
+    assert src.index("_CARRY_OVER_IDENTITY_SQL") < src.index("_DELETE_LOSERS_SQL")
+
+
+def test_merge_runs_fias_then_canon_pass_in_one_transaction() -> None:
+    """merge_duplicate_houses runs the fias pass first, then the canon pass, in ONE transaction."""
+    src = inspect.getsource(hdm.merge_duplicate_houses)
+    assert 'pass_label="fias"' in src and 'pass_label="canon"' in src
+    assert src.index('pass_label="fias"') < src.index('pass_label="canon"')
+    assert "_BUILD_MAPPING_SQL_FIAS" in src
+    # single transaction wrapping BOTH passes: exactly one commit.
+    assert src.count("db.commit()") == 1
+
+
+def test_carry_over_no_psycopg_v3_colon_colon_cast() -> None:
+    """psycopg v3: the carry-over SQL must not use the :param::type bound-param cast form."""
+    assert not re.search(r":\w+::", _CARRY_SQL)
+
+
 # ── Service signature + dry_run / idempotent contract ─────────────────────────
 
 
@@ -288,6 +388,7 @@ class _FakeDB:
     def __init__(self, mapping_rows: list[_Row], step_rowcount: int = 1):
         self._mapping_rows = mapping_rows
         self._step_rowcount = step_rowcount
+        self._mapping_served = False
         self.commits = 0
         self.rollbacks = 0
         self.executed: list[str] = []
@@ -298,8 +399,13 @@ class _FakeDB:
         if "CREATE TEMP TABLE" in sql:
             return _FakeResult()
         if "SELECT loser_id, keeper_id, norm_address" in sql:
+            # The service now runs TWO passes (fias, then canon). Model «fias pass found the
+            # duplicates, canon pass is clean»: serve the scripted mapping once, empty afterwards.
+            if self._mapping_served:
+                return _FakeResult(rows=[])
+            self._mapping_served = True
             return _FakeResult(rows=list(self._mapping_rows))
-        # any UPDATE/DELETE/INSERT step
+        # any UPDATE/DELETE/INSERT step (incl. DROP TABLE, carry-over, delete, backfill)
         return _FakeResult(rowcount=self._step_rowcount)
 
     def commit(self) -> None:
@@ -714,3 +820,107 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None:
         db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900010 AND 900015"))
         db.commit()
         db.close()
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_fias_pass_cross_guard_and_identity_carryover() -> None:
+    """End-to-end on a real DB for the #1772 follow-up (fias pass + cross-fias guard + carry-over):
+
+      A. same house_fias_id, DIFFERENT canon (different streets)  → MERGED by the FIAS pass;
+         the loser's listing re-points onto the single survivor.
+      B. same canon (slash-collapse «32»/«3/2»), BOTH fias filled and DIFFERENT → NOT merged
+         (cross-fias guard vetoes — provably different buildings), both survive.
+      C. same canon, fias only on the LOSER → MERGED by the canon pass; the loser's
+         house_fias_id + gar_house_guid CARRY OVER onto the fias-less keeper.
+      D. a second full run is idempotent (nothing left to merge).
+
+    Synthetic streets «*1772» avoid colliding with any real prod address/alias.
+    """
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    try:
+        db.execute(
+            _t(
+                "INSERT INTO houses "
+                "(id, source, ext_house_id, address, lat, lon, house_fias_id, gar_house_guid, "
+                " dadata_enriched_at) VALUES "
+                # A — same fias, different canon (different streets), ~10 m apart → FIAS-pass merge
+                "(900020,'avito','EXT-F-K','ФиасОдин1772, 10',   56.84000,60.60000,"
+                " 'F-SAME-1772',NULL,NULL),"
+                "(900021,'cian', 'EXT-F-L','СовсемДругая1772, 77',56.84009,60.60000,"
+                " 'F-SAME-1772',NULL,NULL),"
+                # B — same canon (slash-collapse), DIFFERENT fias → cross-fias guard BLOCKS
+                "(900022,'avito','EXT-B-1','Клара1772, 32',56.84000,60.60000,"
+                " 'F-B1-1772',NULL,NULL),"
+                "(900023,'cian', 'EXT-B-2','Клара1772, 3/2',56.84009,60.60000,"
+                " 'F-B2-1772',NULL,NULL),"
+                # C — same canon, fias only on the loser → canon-pass merge + carry-over
+                "(900024,'avito','EXT-C-K','Донбасс1772, 8',56.84000,60.60000,NULL,NULL,NULL),"
+                "(900025,'cian', 'EXT-C-L','Донбасс1772, 8',56.84009,60.60000,"
+                " 'F-CARRY-1772','G-CARRY-1772',NOW())"
+            )
+        )
+        # Listings: A keeper + A loser (loser's must re-point); C keeper (gives it the listing-count
+        # tie-break so the fias-LESS row wins the keeper slot and receives the carried-over fias).
+        db.execute(
+            _t(
+                "INSERT INTO listings "
+                "(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) VALUES "
+                "(910020,'avito','http://t/f/k','LF-K','dh-f-keep',5000000,900020),"
+                "(910021,'cian', 'http://t/f/l','LF-L','dh-f-lose',6000000,900021),"
+                "(910024,'avito','http://t/c/k','LC-K','dh-c-keep',5500000,900024)"
+            )
+        )
+        db.commit()
+
+        out = hdm.merge_duplicate_houses(db, dry_run=False)
+        assert out["losers_deleted"] >= 2  # at least A + C merged
+
+        ids = [
+            r.id
+            for r in db.execute(
+                _t("SELECT id FROM houses WHERE id BETWEEN 900020 AND 900025 ORDER BY id")
+            ).all()
+        ]
+        # A: fias merge — different canon, same fias → the loser is gone.
+        assert 900020 in ids and 900021 not in ids, "same fias must merge across differing canons"
+        # B: cross-fias guard — same canon but different fias → BOTH survive.
+        assert 900022 in ids and 900023 in ids, "different fias must veto a same-canon merge"
+        # C: canon merge — same canon, one fias → the fias-less keeper survives.
+        assert 900024 in ids and 900025 not in ids, "same-canon single-fias pair must merge"
+
+        # A: the merged loser's listing re-pointed onto the surviving keeper.
+        repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910021")).scalar()
+        assert repointed == 900020
+
+        # C: identity carry-over — the loser's fias + ГАР guid moved onto the fias-less keeper.
+        keeper_c = db.execute(
+            _t("SELECT house_fias_id, gar_house_guid FROM houses WHERE id = 900024")
+        ).one()
+        assert keeper_c.house_fias_id == "F-CARRY-1772", "loser's house_fias_id must carry over"
+        assert keeper_c.gar_house_guid == "G-CARRY-1772", "loser's gar_house_guid must carry over"
+
+        # D: idempotent second run — nothing left to merge.
+        again = hdm.merge_duplicate_houses(db, dry_run=False)
+        assert again["losers_deleted"] == 0
+    finally:
+        db.rollback()
+        db.execute(_t("DELETE FROM listings WHERE id IN (910020,910021,910024)"))
+        db.execute(
+            _t(
+                "DELETE FROM house_sources WHERE ext_id IN "
+                "('EXT-F-K','EXT-F-L','EXT-B-1','EXT-B-2','EXT-C-K','EXT-C-L')"
+            )
+        )
+        db.execute(
+            _t(
+                "DELETE FROM house_address_aliases WHERE normalized_address IN "
+                "('фиасодин1772, 10','совсемдругая1772, 77','клара1772, 32','клара1772, 3/2',"
+                "'донбасс1772, 8')"
+            )
+        )
+        db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900020 AND 900025"))
+        db.commit()
+        db.close()