feat(tradein): recurring house-dedup merge (schedule dormant) (#1772) (#1933)

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

@@ -0,0 +1,567 @@
+"""Recurring, idempotent, collision-safe house-deduplication merge (#1772).
+
+WHY (recurrence, not a one-shot):
+    Migration 108_merge_duplicate_houses.sql did a ONE-TIME merge of duplicate `houses`
+    rows, but the duplicates RECUR: the matching pipeline keeps creating them
+    (Tier-3 geo-jitter splits one physical building across rows + per-source ext_id INSERTs).
+    A single migration cannot fix a steady-state inflow — we need a RECURRING merge that
+    re-runs on a schedule and is a no-op when there is nothing to merge.
+
+WHAT this is:
+    A direct, set-based reimplementation of 108's proven
+        cluster → pick canonical → re-point children (UNIQUE-collision-safe) → delete losers
+    pipeline, run inside ONE transaction so a crash leaves the table untouched.
+
+    Cluster key: EXACT `houses.address` (cadastral_number is 100% NULL on prod — confirmed
+    in migration 040 — so address is the real building key). Rows whose address is NULL/blank
+    are never clustered (cluster_key NULL → ignored). Only addresses shared by >1 house_id
+    form a cluster.
+
+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)
+    3. most-populated fields     (the richest metadata row)
+    4. min(id)                  (deterministic tie-break)
+
+CHILD-COLLISION HANDLING — the migration-133 lesson:
+    Migration 133 broke on prod because a child UNIQUE constraint was missed. EVERY FK child
+    of houses(id) is handled below. The FK set was re-audited against the LIVE schema
+    (grep "REFERENCES houses" data/sql/*.sql) — it is the SAME 11 children 108 handled; no
+    new FK child was added after 108. For each UNIQUE-constrained child we DELETE the loser
+    rows that would collide with the keeper (deduping by the TARGET key so dup-vs-dup
+    collisions inside clusters >2 are also caught) BEFORE re-pointing the survivors.
+
+    IMPORTANT divergence from 108: houses_price_dynamics's UNIQUE was REPLACED by
+    migration 029 — it is now the 6-column
+        (house_id, source, room_count, prices_type, period, month_date)
+    NOT 108's stale 3-column (house_id, month_date, source). We dedup on the LIVE 6-column
+    key; deduping on the old key would still violate the live constraint on re-point.
+
+    Per-child handling (loser=L, keeper=K):
+      listings.house_id_fk              no UNIQUE         → plain re-point
+      house_placement_history.house_id  no UNIQUE         → plain re-point
+      house_reviews.house_id            no blocking UNIQUE→ plain re-point
+      house_reliability_checks.house_id no blocking UNIQUE→ plain re-point
+      external_valuations.house_id      no blocking UNIQUE→ plain re-point
+      house_sources             UNIQUE(ext_source,ext_id) → delete colliding L, re-point rest
+      house_address_aliases     UNIQUE(normalized_address)→ delete colliding L, re-point rest
+      houses_price_dynamics     UNIQUE(6-col, migr 029)   → dedup by target 6-col key, re-point
+      house_imv_evaluations     UNIQUE(house_id)          → dedup by target house_id, re-point
+      house_suggestions         UNIQUE(house_id,ext_item_id)→ dedup by target key, re-point
+      address_mismatch_audit    UNIQUE(house_id,audit_batch)→ dedup by target key, re-point
+
+BACKFILL (reduces recurrence):
+    After deleting losers we backfill house_sources + house_address_aliases onto the keeper
+    (same as 108) so the matching pipeline's Tier-1/Tier-2 finds the keeper next scrape and
+    does not immediately re-split it.
+
+IDEMPOTENCY:
+    Every UPDATE/DELETE keys off a temp mapping of (loser→keeper). On a clean table the
+    mapping is empty → every statement touches 0 rows → no-op. Re-running is safe.
+
+DRY-RUN:
+    dry_run=True computes the same mapping and counts, then ROLLS BACK — no writes. Used by
+    the orchestrator to preview a merge before enabling the (DESTRUCTIVE) schedule.
+
+psycopg v3: all SQL uses CAST(:x AS type), never the colon-colon bound-param cast form.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Significant columns counted to score per-row completeness when picking the keeper.
+# Mirrors 108's completeness expression (all confirmed present in the live houses schema).
+_COMPLETENESS_EXPR = """
+    (h.address IS NOT NULL)::int +
+    (h.lat IS NOT NULL)::int +
+    (h.lon IS NOT NULL)::int +
+    (h.year_built IS NOT NULL)::int +
+    (h.house_type IS NOT NULL)::int +
+    (h.total_floors IS NOT NULL)::int +
+    (h.cadastral_number IS NOT NULL)::int +
+    (h.developer_name IS NOT NULL)::int +
+    (h.rating_score IS NOT NULL)::int +
+    (h.avito_id_hash IS NOT NULL)::int
+"""
+
+# Keeper ORDER BY, shared by the ROW_NUMBER() rank and the first_value() keeper pick so they
+# agree row-for-row. Priority: geom present → most linked listings → most-populated → min id.
+_KEEPER_ORDER = f"""
+    (h.geom IS NOT NULL) DESC,
+    listing_cnt DESC,
+    ({_COMPLETENESS_EXPR}) DESC,
+    h.id ASC
+"""
+
+# Build the loser→keeper mapping as a TEMP table for this transaction. Only addresses shared
+# by >1 house_id are clustered; the keeper is rn=1 per cluster, losers are rn>1.
+_BUILD_MAPPING_SQL = text(
+    f"""
+    CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
+    WITH clustered AS (
+        SELECT
+            id,
+            CASE WHEN NULLIF(lower(trim(address)), '') IS NOT NULL
+                 THEN 'addr:' || lower(trim(address)) END AS cluster_key
+        FROM houses
+    ),
+    clusters_with_count AS (
+        SELECT cluster_key, count(*) AS n
+        FROM clustered
+        WHERE cluster_key IS NOT NULL
+        GROUP BY cluster_key
+        HAVING count(*) > 1
+    ),
+    dup_houses AS (
+        SELECT h.id, cl.cluster_key
+        FROM houses h
+        JOIN clustered cl ON cl.id = h.id
+        JOIN clusters_with_count cw ON cw.cluster_key = cl.cluster_key
+    ),
+    listing_counts AS (
+        SELECT house_id_fk AS house_id, count(*) AS listing_cnt
+        FROM listings
+        WHERE house_id_fk IS NOT NULL
+        GROUP BY house_id_fk
+    ),
+    ranked AS (
+        SELECT
+            dh.id,
+            dh.cluster_key,
+            lower(trim(h.address)) AS norm_address,
+            ROW_NUMBER() OVER (
+                PARTITION BY dh.cluster_key
+                ORDER BY {_KEEPER_ORDER}
+            ) AS rn,
+            first_value(dh.id) OVER (
+                PARTITION BY dh.cluster_key
+                ORDER BY {_KEEPER_ORDER}
+            ) AS keeper_id
+        FROM dup_houses dh
+        JOIN houses h ON h.id = dh.id
+        LEFT JOIN listing_counts lc ON lc.house_id = dh.id
+    )
+    SELECT id AS loser_id, keeper_id, norm_address
+    FROM ranked
+    WHERE rn > 1
+      AND id <> keeper_id
+    """
+)
+
+# Each step keys off _1772_dup_mapping → empty mapping ⇒ 0 rows touched ⇒ idempotent no-op.
+_STEPS: list[tuple[str, str]] = [
+    # ── Plain re-point (no UNIQUE on the FK column) ───────────────────────────
+    (
+        "listings",
+        """
+        UPDATE listings l
+        SET house_id_fk = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE l.house_id_fk = m.loser_id
+        """,
+    ),
+    (
+        "house_placement_history",
+        """
+        UPDATE house_placement_history hph
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hph.house_id = m.loser_id
+        """,
+    ),
+    (
+        "house_reviews",
+        """
+        UPDATE house_reviews hr
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hr.house_id = m.loser_id
+        """,
+    ),
+    (
+        "house_reliability_checks",
+        """
+        UPDATE house_reliability_checks hrc
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hrc.house_id = m.loser_id
+        """,
+    ),
+    (
+        "external_valuations",
+        """
+        UPDATE external_valuations ev
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE ev.house_id = m.loser_id
+        """,
+    ),
+    # ── UNIQUE(ext_source, ext_id): delete colliding losers, re-point rest ─────
+    (
+        "house_sources(collision-delete)",
+        """
+        DELETE FROM house_sources hs
+        USING _1772_dup_mapping m
+        WHERE hs.house_id = m.loser_id
+          AND EXISTS (
+              SELECT 1 FROM house_sources hs2
+              WHERE hs2.house_id = m.keeper_id
+                AND hs2.ext_source = hs.ext_source
+                AND hs2.ext_id = hs.ext_id
+          )
+        """,
+    ),
+    (
+        "house_sources(re-point)",
+        """
+        UPDATE house_sources hs
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hs.house_id = m.loser_id
+        """,
+    ),
+    # ── UNIQUE(normalized_address): delete colliding losers, re-point rest ─────
+    (
+        "house_address_aliases(collision-delete)",
+        """
+        DELETE FROM house_address_aliases haa
+        USING _1772_dup_mapping m
+        WHERE haa.house_id = m.loser_id
+          AND EXISTS (
+              SELECT 1 FROM house_address_aliases haa2
+              WHERE haa2.house_id = m.keeper_id
+                AND haa2.normalized_address = haa.normalized_address
+          )
+        """,
+    ),
+    (
+        "house_address_aliases(re-point)",
+        """
+        UPDATE house_address_aliases haa
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE haa.house_id = m.loser_id
+        """,
+    ),
+    # ── UNIQUE(house_id, source, room_count, prices_type, period, month_date) ──
+    # LIVE 6-col key (migration 029 replaced 108's stale 3-col). Dedup by the TARGET key
+    # (keeper survives, else min id) so dup-vs-dup collisions in clusters >2 are also caught.
+    (
+        "houses_price_dynamics(dedup)",
+        """
+        DELETE FROM houses_price_dynamics t
+        USING (
+            SELECT t2.id,
+                   ROW_NUMBER() OVER (
+                       PARTITION BY COALESCE(m.keeper_id, t2.house_id),
+                                    t2.source, t2.room_count, t2.prices_type,
+                                    t2.period, t2.month_date
+                       ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
+                   ) AS rn
+            FROM houses_price_dynamics t2
+            LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
+        ) d
+        WHERE t.id = d.id AND d.rn > 1
+        """,
+    ),
+    (
+        "houses_price_dynamics(re-point)",
+        """
+        UPDATE houses_price_dynamics hpd
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hpd.house_id = m.loser_id
+        """,
+    ),
+    # ── UNIQUE(house_id): one evaluation per keeper ───────────────────────────
+    (
+        "house_imv_evaluations(dedup)",
+        """
+        DELETE FROM house_imv_evaluations t
+        USING (
+            SELECT t2.id,
+                   ROW_NUMBER() OVER (
+                       PARTITION BY COALESCE(m.keeper_id, t2.house_id)
+                       ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
+                   ) AS rn
+            FROM house_imv_evaluations t2
+            LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
+        ) d
+        WHERE t.id = d.id AND d.rn > 1
+        """,
+    ),
+    (
+        "house_imv_evaluations(re-point)",
+        """
+        UPDATE house_imv_evaluations hie
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hie.house_id = m.loser_id
+        """,
+    ),
+    # ── UNIQUE(house_id, ext_item_id) ─────────────────────────────────────────
+    (
+        "house_suggestions(dedup)",
+        """
+        DELETE FROM house_suggestions t
+        USING (
+            SELECT t2.id,
+                   ROW_NUMBER() OVER (
+                       PARTITION BY COALESCE(m.keeper_id, t2.house_id), t2.ext_item_id
+                       ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
+                   ) AS rn
+            FROM house_suggestions t2
+            LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
+        ) d
+        WHERE t.id = d.id AND d.rn > 1
+        """,
+    ),
+    (
+        "house_suggestions(re-point)",
+        """
+        UPDATE house_suggestions hs
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE hs.house_id = m.loser_id
+        """,
+    ),
+    # ── UNIQUE(house_id, audit_batch) ─────────────────────────────────────────
+    (
+        "address_mismatch_audit(dedup)",
+        """
+        DELETE FROM address_mismatch_audit t
+        USING (
+            SELECT t2.id,
+                   ROW_NUMBER() OVER (
+                       PARTITION BY COALESCE(m.keeper_id, t2.house_id), t2.audit_batch
+                       ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
+                   ) AS rn
+            FROM address_mismatch_audit t2
+            LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
+        ) d
+        WHERE t.id = d.id AND d.rn > 1
+        """,
+    ),
+    (
+        "address_mismatch_audit(re-point)",
+        """
+        UPDATE address_mismatch_audit ama
+        SET house_id = m.keeper_id
+        FROM _1772_dup_mapping m
+        WHERE ama.house_id = m.loser_id
+        """,
+    ),
+]
+
+# Delete the loser houses — all FK children are re-pointed or CASCADE by now.
+_DELETE_LOSERS_SQL = text(
+    """
+    DELETE FROM houses h
+    USING _1772_dup_mapping m
+    WHERE h.id = m.loser_id
+    """
+)
+
+# Backfill house_sources for keeper avito-rows so Tier-1 (source_exact) finds the keeper next
+# scrape (reduces recurrence). Same shape as 108 Step 5.
+_BACKFILL_SOURCES_SQL = text(
+    """
+    INSERT INTO house_sources (house_id, ext_source, ext_id, confidence, matched_method, matched_at)
+    SELECT h.id, h.source, h.ext_house_id, 1.0, 'backfill_dedup_merge', NOW()
+    FROM houses h
+    WHERE h.source = 'avito'
+      AND h.ext_house_id IS NOT NULL
+      AND h.id IN (SELECT DISTINCT keeper_id FROM _1772_dup_mapping)
+    ON CONFLICT (ext_source, ext_id) DO NOTHING
+    """
+)
+
+# Backfill normalized-address aliases onto the keeper so Tier-2 (address) finds it next scrape.
+_BACKFILL_ALIASES_SQL = text(
+    """
+    INSERT INTO house_address_aliases (house_id, normalized_address, fingerprint, source)
+    SELECT h.id, lower(trim(h.address)), NULL, 'backfill_dedup_merge'
+    FROM houses h
+    WHERE h.address IS NOT NULL
+      AND length(trim(h.address)) >= 5
+      AND h.id IN (SELECT DISTINCT keeper_id FROM _1772_dup_mapping)
+    ON CONFLICT (normalized_address) DO NOTHING
+    """
+)
+
+
+@dataclass
+class DedupMergeResult:
+    """Counters from one merge pass (also the dry-run preview)."""
+
+    clusters_merged: int = 0  # distinct keepers that absorbed at least one loser
+    losers_deleted: int = 0  # duplicate house rows removed (0 on a clean table)
+    listings_repointed: int = 0  # listings.house_id_fk moved loser→keeper
+    children_deleted: int = 0  # collision/dedup deletions across all UNIQUE children
+    children_repointed: int = 0  # survivor child rows moved loser→keeper
+    dry_run: bool = False
+    duration_sec: float = field(default=0.0)
+
+    def to_counters(self) -> dict[str, int]:
+        return {
+            "clusters_merged": self.clusters_merged,
+            "losers_deleted": self.losers_deleted,
+            "listings_repointed": self.listings_repointed,
+            "children_deleted": self.children_deleted,
+            "children_repointed": self.children_repointed,
+            "dry_run": int(self.dry_run),
+            "duration_sec": int(self.duration_sec),
+        }
+
+
+def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
+    """Cluster houses by exact address and merge duplicates onto one canonical keeper.
+
+    Re-implements migration 108's proven collision-safe pipeline as a RECURRING job:
+        cluster (exact address) → pick keeper → re-point children (UNIQUE-collision-safe) →
+        delete losers → backfill sources/aliases onto keeper.
+    All in ONE transaction. dry_run=True computes counts then ROLLS BACK (no writes).
+    Idempotent: a clean table yields an empty mapping → every statement is a 0-row no-op.
+
+    Returns the counter dict (DedupMergeResult.to_counters()).
+    """
+    start = time.monotonic()
+    result = DedupMergeResult(dry_run=dry_run)
+
+    try:
+        db.execute(_BUILD_MAPPING_SQL)
+
+        mapping = db.execute(
+            text(
+                "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",
+                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(
+                "merge_duplicate_houses: DRY-RUN computed clusters=%d losers=%d "
+                "listings_repointed=%d children_deleted=%d children_repointed=%d — ROLLED BACK",
+                result.clusters_merged,
+                result.losers_deleted,
+                result.listings_repointed,
+                result.children_deleted,
+                result.children_repointed,
+            )
+        else:
+            db.commit()
+            logger.info(
+                "merge_duplicate_houses: COMMITTED clusters=%d losers=%d "
+                "listings_repointed=%d children_deleted=%d children_repointed=%d",
+                result.clusters_merged,
+                result.losers_deleted,
+                result.listings_repointed,
+                result.children_deleted,
+                result.children_repointed,
+            )
+    except Exception:
+        logger.exception("merge_duplicate_houses: FAILED — rolling back")
+        try:
+            db.rollback()
+        except Exception:
+            logger.exception("merge_duplicate_houses: rollback also failed")
+        raise
+
+    result.duration_sec = time.monotonic() - start
+    return result.to_counters()
+
+
+# ── Run lifecycle wrapper (scheduler entrypoint) ─────────────────────────────
+def run_house_dedup_merge(db: Session, *, run_id: int, params: dict) -> dict[str, int]:
+    """Run-lifecycle wrapper for the recurring house-dedup merge (sync, DB-only).
+
+    Launched by the in-app scheduler (source='house_dedup_merge') via
+    trigger_house_dedup_merge_run, or manually. Mirrors run_cadastral_geo_match: pure
+    internal DB op, finalises scrape_runs (mark_done / mark_failed) with counters.
+
+    Params (default_params jsonb):
+        dry_run: compute + return counts then ROLLBACK without writing (default false).
+                 The deploy seeds the schedule DISABLED; the orchestrator can flip a single
+                 manual run to dry_run=true to preview before enabling.
+    """
+    from app.services import scrape_runs as runs_mod
+
+    dry_run = bool(params.get("dry_run", False))
+    counters: dict[str, int] = {
+        "clusters_merged": 0,
+        "losers_deleted": 0,
+        "listings_repointed": 0,
+        "children_deleted": 0,
+        "children_repointed": 0,
+        "dry_run": int(dry_run),
+    }
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+        counters = merge_duplicate_houses(db, dry_run=dry_run)
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "run_house_dedup_merge: run_id=%d DONE clusters=%d losers=%d dry_run=%s",
+            run_id,
+            counters.get("clusters_merged", 0),
+            counters.get("losers_deleted", 0),
+            dry_run,
+        )
+        return counters
+    except Exception as exc:
+        logger.exception("run_house_dedup_merge: run_id=%d FAILED", run_id)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

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

@@ -110,6 +110,21 @@ Sources:
                               использует собственную curl_cffi-сессию; window 15:00-17:00 UTC —
                               после avito_full_load 13-15 UTC и утреннего Avito-блока, дома
                               уже geocoded/enriched дневными sweep'ами)
+  - house_dedup_merge       → run_house_dedup_merge
+                              (services/house_dedup_merge.py, #1772; recurring, idempotent,
+                              collision-safe house-deduplication merge. Clusters houses by
+                              EXACT address (cadastral_number is 100% NULL), picks one canonical
+                              keeper per cluster (geom NOT NULL → most listings → most-populated
+                              → min id), re-points all FK children with UNIQUE-collision handling
+                              — reuses migration 108's proven pattern — deletes loser rows,
+                              backfills house_sources/house_address_aliases onto the keeper so the
+                              matching pipeline finds the keeper next scrape. All in ONE
+                              transaction; dry_run param computes counts then ROLLBACKs.
+                              DESTRUCTIVE (DELETEs duplicate house rows) but idempotent: clean
+                              table → empty mapping → 0-row no-op. Seed 135 ships the schedule
+                              DISABLED (enabled=false) — deploy is neutral; enable deliberately
+                              after a validated dry-run + manual run; weekly window 04:00-05:00
+                              UTC — quiet hour, no scraper/proxy contention)
 """
 
 from __future__ import annotations
@@ -1654,6 +1669,53 @@ async def trigger_house_imv_backfill_run(db: Session, schedule_row: dict[str, An
     return run_id
 
 
+async def trigger_house_dedup_merge_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch run_house_dedup_merge в executor (sync DB-only task, #1772).
+
+    Recurring, idempotent, collision-safe house-deduplication merge: clusters `houses` by
+    EXACT address, picks one canonical keeper per cluster (geom NOT NULL → most listings →
+    most-populated → min id), re-points all FK children with UNIQUE-collision handling
+    (reuses migration 108's proven pattern), deletes the loser rows, backfills
+    house_sources/house_address_aliases onto the keeper — all in ONE transaction.
+
+    DESTRUCTIVE (DELETEs duplicate house rows) but idempotent: a clean table yields an empty
+    mapping → every statement is a 0-row no-op. Seed 135 ships the schedule DISABLED
+    (enabled=false) — the deploy is neutral; the orchestrator dry-runs + does one validated
+    manual run, then enables deliberately. default_params.dry_run=true previews without writes.
+
+    Sync task (set-based UPDATE/DELETE, no async HTTP) — run in run_in_executor by the same
+    pattern as trigger_cadastral_geo_match_run. run_house_dedup_merge owns the scrape_runs
+    lifecycle (mark_done/mark_failed).
+
+    Returns run_id или None (skip — already running).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.services.house_dedup_merge import run_house_dedup_merge
+
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(
+                None,
+                lambda: run_house_dedup_merge(run_db, run_id=run_id, params=params),
+            )
+        except Exception:
+            logger.exception("scheduler: run_house_dedup_merge crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered house_dedup_merge run_id=%d", run_id)
+    return run_id
+
+
 def get_due_schedules(db: Session) -> list[dict[str, Any]]:
     """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
     rows = (
@@ -1737,6 +1799,8 @@ async def scheduler_loop() -> None:
                         await trigger_cadastral_geo_match_run(db, sch)
                     elif source == "house_imv_backfill":
                         await trigger_house_imv_backfill_run(db, sch)
+                    elif source == "house_dedup_merge":
+                        await trigger_house_dedup_merge_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

tradein-mvp/backend/data/sql/135_scrape_schedules_seed_house_dedup_merge.sql

@@ -0,0 +1,79 @@
+-- 135_scrape_schedules_seed_house_dedup_merge.sql
+-- Scheduler seed for the recurring house-deduplication merge (#1772).
+--
+-- WHAT (source='house_dedup_merge'):
+--   trigger_house_dedup_merge_run (scheduler.py) → run_house_dedup_merge
+--   (services/house_dedup_merge.py) → merge_duplicate_houses. Recurring, idempotent,
+--   collision-safe merge of duplicate `houses` rows:
+--     1. Cluster houses by EXACT address (cadastral_number is 100% NULL on prod — confirmed
+--        in migration 040 — so address is the real building key). Only addresses shared by
+--        >1 house_id form a cluster.
+--     2. Pick one canonical keeper per cluster: geom NOT NULL → most linked listings →
+--        most-populated fields → min(id).
+--     3. Re-point every FK child of houses(id) onto the keeper with UNIQUE-collision handling
+--        (reuses migration 108's proven cluster→canonical→re-point→delete pattern). The 11
+--        children + their constraints (listings, house_placement_history, house_reviews,
+--        house_reliability_checks, external_valuations, house_sources, house_address_aliases,
+--        houses_price_dynamics [LIVE 6-col UNIQUE per migration 029], house_imv_evaluations,
+--        house_suggestions, address_mismatch_audit).
+--     4. DELETE the loser house rows; backfill house_sources + house_address_aliases onto the
+--        keeper so the matching pipeline (Tier-1/2) finds the keeper next scrape (reduces
+--        recurrence). ALL in ONE transaction.
+--   Idempotent: a clean table yields an empty mapping → every statement is a 0-row no-op.
+--   dry_run param computes the same counts then ROLLBACKs (no writes).
+--
+-- WHY RECURRING (not a one-shot migration):
+--   Migration 108 did a ONE-TIME merge, but duplicates RECUR — the matching pipeline keeps
+--   creating them (Tier-3 geo-jitter + per-source ext_id INSERTs). A migration cannot fix a
+--   steady-state inflow; this schedule re-runs and is a no-op when there is nothing to merge.
+--
+-- *** enabled=false (DORMANT) — DEPLOY MUST BE NEUTRAL ***
+--   This is an auto-recurring DESTRUCTIVE merge (it DELETEs duplicate house rows). The deploy
+--   ships it DISABLED so nothing fires automatically. The orchestrator dry-runs it
+--   (default_params.dry_run=true on a manual run) + does ONE validated manual run, then flips
+--   enabled=true deliberately:
+--       UPDATE scrape_schedules SET enabled = true WHERE source = 'house_dedup_merge';
+--
+-- Window 04:00-05:00 UTC (07:00-08:00 MSK):
+--   - Quiet hour: after the overnight scraper/sweep block and before the morning Avito block
+--     (city_sweep 06-07, detail_backfill 09-12). Pure internal DB op — no scraper/proxy
+--     contention. interval_days=7 → weekly cadence (default_params).
+--   - next_run_at = tomorrow 04:00 UTC (NOT NULL — follows the seed convention; harmless while
+--     enabled=false, since get_due_schedules filters enabled=true first).
+--
+-- default_params:
+--   dry_run       -- false: actually merge. The orchestrator overrides to true on a manual run
+--                    to preview counts without writing.
+--   interval_days -- 7: weekly cadence (compute_next_run_at reads this).
+--
+-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)),
+--   108_merge_duplicate_houses.sql (the proven collision-safe pattern this reuses),
+--   029_extend_matching_valuation_dynamics.sql (LIVE 6-col houses_price_dynamics UNIQUE).
+-- Idempotent: ON CONFLICT (source) DO NOTHING.
+-- Runner applies migrations WITHOUT --single-transaction, so wrap in explicit BEGIN/COMMIT.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'house_dedup_merge',
+    false,
+    4,
+    5,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 4)) AT TIME ZONE 'UTC',
+    '{"dry_run": false, "interval_days": 7}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (replaces cron-script setup). Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), yandex_address_backfill (#855, EKB pilot), sber_index_pull (#887, monthly), rosreestr_quarter_poll (#888, monthly), cian_city_sweep (dormant, #973), yandex_newbuilding_sweep (dormant, #974), geocode_missing_listings (#1: listings geom backfill, all sources), avito_detail_backfill (#1551: nightly detail-enrichment backfill for legacy avito listings), house_imv_backfill (#854: nightly bulk Avito IMV at house level → house_imv_evaluations), house_dedup_merge (#1772: recurring collision-safe merge of duplicate houses by exact address — DORMANT enabled=false, DESTRUCTIVE, enable deliberately after a validated dry-run + manual run).';
+
+COMMIT;

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -0,0 +1,590 @@
+"""Tests for the recurring house-deduplication merge (#1772).
+
+Convention mirrors tests/tasks/test_cadastral_geo_match.py and
+tests/test_house_imv_backfill_scheduler.py: the merge is SQL-heavy and the CI gate has no
+live Postgres, so the bulk of coverage is STATIC analysis of the emitted SQL (text() clauses
++ inspect.getsource) plus a fake-db behavioural pass driving the dry_run / idempotent / count
+logic. An OPTIONAL real-Postgres behavioural test asserts the actual merge semantics (re-point
+listings, dedup colliding children WITHOUT a UNIQUE violation, delete loser, keeper survives,
+idempotent second run, dry_run no-write) and self-SKIPS when no DB is reachable.
+
+The static block is the migration-133 guard: it asserts EVERY FK child of houses(id) is
+handled, and that the UNIQUE-constrained children dedup by the TARGET key before re-pointing.
+"""
+
+from __future__ import annotations
+
+import inspect
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import house_dedup_merge as hdm
+from app.services import scheduler
+
+_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)
+_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)
+
+
+def _flat(sql: str) -> str:
+    return re.sub(r"\s+", " ", sql)
+
+
+# ── Cluster key + keeper rule ─────────────────────────────────────────────────
+
+
+def test_cluster_key_is_exact_address_not_cadastral() -> None:
+    """Spec: cadastral is unpopulated → cluster by EXACT address only."""
+    flat = _flat(_MAPPING_SQL)
+    assert "'addr:' || lower(trim(address))" in flat
+    # cadastral_number must NOT be part of the cluster key (it is 100% NULL).
+    assert "cadastral_number" not in _flat(
+        _MAPPING_SQL[: _MAPPING_SQL.index("ranked")]
+    )  # not in cluster construction
+    # Only addresses shared by >1 house form a cluster.
+    assert "HAVING count(*) > 1" in flat
+
+
+def test_keeper_rule_priority_order() -> None:
+    """Keeper: geom NOT NULL → most linked listings → most-populated → min(id)."""
+    order = _flat(hdm._KEEPER_ORDER)
+    assert "(h.geom IS NOT NULL) DESC" in order
+    assert "listing_cnt DESC" in order
+    assert "h.id ASC" in order
+    # geom precedes listing count precedes id (ordering matters).
+    assert order.index("geom IS NOT NULL") < order.index("listing_cnt")
+    assert order.index("listing_cnt") < order.index("h.id ASC")
+    # completeness is counted from the live houses columns.
+    for col in ("year_built", "house_type", "total_floors", "developer_name", "rating_score"):
+        assert col in hdm._COMPLETENESS_EXPR
+
+
+def test_mapping_selects_losers_only() -> None:
+    """The mapping is loser_id→keeper_id; the keeper itself is excluded (rn>1, id<>keeper)."""
+    flat = _flat(_MAPPING_SQL)
+    assert "SELECT id AS loser_id, keeper_id" in flat
+    assert "WHERE rn > 1" in flat
+    assert "id <> keeper_id" in flat
+
+
+def test_mapping_temp_table_drops_on_commit() -> None:
+    """TEMP ... ON COMMIT DROP — the mapping never leaks across runs (idempotent setup)."""
+    assert "CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP" in _flat(_MAPPING_SQL)
+
+
+# ── EVERY FK child of houses(id) is handled (the migration-133 guard) ──────────
+
+# The authoritative FK-child set (grep "REFERENCES houses" data/sql/*.sql). If a new child
+# is ever added this list must grow with it — the test then forces a matching merge step.
+_FK_CHILDREN = {
+    "listings": "house_id_fk",
+    "house_placement_history": "house_id",
+    "house_reviews": "house_id",
+    "house_reliability_checks": "house_id",
+    "external_valuations": "house_id",
+    "house_sources": "house_id",
+    "house_address_aliases": "house_id",
+    "houses_price_dynamics": "house_id",
+    "house_imv_evaluations": "house_id",
+    "house_suggestions": "house_id",
+    "address_mismatch_audit": "house_id",
+}
+
+
+def test_every_fk_child_of_houses_is_repointed() -> None:
+    """Each FK child must be re-pointed loser→keeper somewhere in the merge steps."""
+    for table in _FK_CHILDREN:
+        assert table in _ALL_STEP_SQL, f"FK child {table} is never touched by the merge"
+
+
+def test_unique_constrained_children_dedup_before_repoint() -> None:
+    """UNIQUE-constrained children must DELETE colliding losers before re-pointing survivors.
+
+    Missing one of these is exactly what broke migration 133. We assert each has both a
+    collision/dedup DELETE step and a re-point UPDATE step, and that the DELETE precedes
+    the UPDATE in step order.
+    """
+    unique_children = [
+        ("house_sources", "house_sources(collision-delete)", "house_sources(re-point)"),
+        (
+            "house_address_aliases",
+            "house_address_aliases(collision-delete)",
+            "house_address_aliases(re-point)",
+        ),
+        (
+            "houses_price_dynamics",
+            "houses_price_dynamics(dedup)",
+            "houses_price_dynamics(re-point)",
+        ),
+        (
+            "house_imv_evaluations",
+            "house_imv_evaluations(dedup)",
+            "house_imv_evaluations(re-point)",
+        ),
+        ("house_suggestions", "house_suggestions(dedup)", "house_suggestions(re-point)"),
+        (
+            "address_mismatch_audit",
+            "address_mismatch_audit(dedup)",
+            "address_mismatch_audit(re-point)",
+        ),
+    ]
+    labels = [label for label, _ in hdm._STEPS]
+    for _table, del_label, rep_label in unique_children:
+        assert del_label in _STEP_SQL, f"{del_label} missing"
+        assert rep_label in _STEP_SQL, f"{rep_label} missing"
+        assert "DELETE FROM" in _STEP_SQL[del_label]
+        assert "UPDATE" in _STEP_SQL[rep_label]
+        assert labels.index(del_label) < labels.index(rep_label)
+
+
+def test_price_dynamics_dedup_uses_live_6col_key_not_stale_3col() -> None:
+    """Migration 029 replaced the price_dynamics UNIQUE with a 6-col key — dedup on it.
+
+    Deduping on 108's stale (house_id, month_date, source) would still violate the live
+    6-col constraint on re-point. The dedup PARTITION must include all 6 dimension columns.
+    """
+    sql = _flat(_STEP_SQL["houses_price_dynamics(dedup)"])
+    for col in ("source", "room_count", "prices_type", "period", "month_date"):
+        assert col in sql, f"price_dynamics dedup missing dimension column {col}"
+    # partition over the TARGET keeper (COALESCE(keeper_id, house_id)) so dup-vs-dup is caught.
+    assert "PARTITION BY COALESCE(m.keeper_id, t2.house_id)" in sql
+
+
+def test_collision_dedup_partitions_on_target_keeper() -> None:
+    """All dedup steps partition on the keeper target so cluster-of-3 dup-vs-dup collisions
+    are caught, with the keeper's own row surviving (keeper_id IS NULL DESC)."""
+    for label in (
+        "houses_price_dynamics(dedup)",
+        "house_imv_evaluations(dedup)",
+        "house_suggestions(dedup)",
+        "address_mismatch_audit(dedup)",
+    ):
+        sql = _flat(_STEP_SQL[label])
+        assert "PARTITION BY COALESCE(m.keeper_id, t2.house_id)" in sql
+        assert "(m.keeper_id IS NULL) DESC" in sql
+
+
+def test_losers_deleted_and_keeper_backfilled() -> None:
+    """Loser houses deleted; keeper backfilled with sources + aliases to reduce recurrence."""
+    assert "DELETE FROM houses" in _flat(str(hdm._DELETE_LOSERS_SQL.text))
+    bf_sources = _flat(str(hdm._BACKFILL_SOURCES_SQL.text))
+    assert "INSERT INTO house_sources" in bf_sources
+    assert "ON CONFLICT (ext_source, ext_id) DO NOTHING" in bf_sources
+    bf_aliases = _flat(str(hdm._BACKFILL_ALIASES_SQL.text))
+    assert "INSERT INTO house_address_aliases" in bf_aliases
+    assert "ON CONFLICT (normalized_address) DO NOTHING" in bf_aliases
+
+
+def test_no_psycopg_v3_colon_colon_cast() -> None:
+    """psycopg v3: never :param::type — must use CAST(:param AS type)."""
+    assert not re.search(r":\w+::", _MAPPING_SQL)
+    assert not re.search(r":\w+::", _ALL_STEP_SQL)
+    assert not re.search(r":\w+::", _SERVICE_SRC)
+
+
+def test_no_fstring_or_format_in_parametrised_sql() -> None:
+    """SQL identifiers are static; no .format()/f-string interpolation of *values* into SQL."""
+    # The only f-strings build static column-list fragments (_KEEPER_ORDER / completeness),
+    # never user/runtime values — assert no '.format(' value-injection into SQL text.
+    assert ".format(" not in _SERVICE_SRC
+
+
+# ── Service signature + dry_run / idempotent contract ─────────────────────────
+
+
+def test_merge_signature_and_dry_run_default() -> None:
+    sig = inspect.signature(hdm.merge_duplicate_houses)
+    assert "dry_run" in sig.parameters
+    assert sig.parameters["dry_run"].default is False
+    # dry_run is keyword-only (the `*` in the spec).
+    assert sig.parameters["dry_run"].kind is inspect.Parameter.KEYWORD_ONLY
+
+
+def test_service_logs_every_merge_for_audit() -> None:
+    """Every loser→keeper move is logged with its address (audit trail)."""
+    assert "loser_id=%d → keeper_id=%d address=%r" in _SERVICE_SRC
+
+
+def test_single_transaction_commit_or_rollback() -> None:
+    """One transaction: dry_run → rollback; real → commit; failure → rollback + re-raise."""
+    src = inspect.getsource(hdm.merge_duplicate_houses)
+    assert "db.rollback()" in src
+    assert "db.commit()" in src
+    assert "raise" in src  # failures re-raise, never swallowed
+
+
+# ── Fake-db behavioural: dry_run / idempotent / counts (no Postgres) ──────────
+
+
+class _FakeResult:
+    def __init__(self, rowcount: int = 0, rows: list[Any] | None = None, scalar: Any = None):
+        self.rowcount = rowcount
+        self._rows = rows or []
+        self._scalar = scalar
+
+    def all(self) -> list[Any]:
+        return self._rows
+
+    def scalar(self) -> Any:
+        return self._scalar
+
+
+class _Row:
+    def __init__(self, loser_id: int, keeper_id: int, norm_address: str):
+        self.loser_id = loser_id
+        self.keeper_id = keeper_id
+        self.norm_address = norm_address
+
+
+class _FakeDB:
+    """Session stand-in: build-mapping + a scripted SELECT result, then per-step rowcounts."""
+
+    def __init__(self, mapping_rows: list[_Row], step_rowcount: int = 1):
+        self._mapping_rows = mapping_rows
+        self._step_rowcount = step_rowcount
+        self.commits = 0
+        self.rollbacks = 0
+        self.executed: list[str] = []
+
+    def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
+        sql = str(getattr(clause, "text", clause))
+        self.executed.append(sql)
+        if "CREATE TEMP TABLE" in sql:
+            return _FakeResult()
+        if "SELECT loser_id, keeper_id, norm_address" in sql:
+            return _FakeResult(rows=list(self._mapping_rows))
+        # any UPDATE/DELETE/INSERT step
+        return _FakeResult(rowcount=self._step_rowcount)
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:
+        self.rollbacks += 1
+
+
+def test_idempotent_no_op_when_no_dups() -> None:
+    """Empty mapping → no commit (only the temp table was opened) → all-zero counters."""
+    db = _FakeDB(mapping_rows=[])
+    out = hdm.merge_duplicate_houses(db, dry_run=False)  # type: ignore[arg-type]
+    assert out["clusters_merged"] == 0
+    assert out["losers_deleted"] == 0
+    assert out["listings_repointed"] == 0
+    assert db.commits == 0  # nothing to commit
+    assert db.rollbacks == 1  # the no-op rollback
+
+
+def test_dry_run_computes_counts_but_rolls_back() -> None:
+    """dry_run=True → counts populated, ZERO commits, exactly one rollback (no writes)."""
+    rows = [_Row(2, 1, "ул. ленина, 5"), _Row(3, 1, "ул. ленина, 5")]
+    db = _FakeDB(mapping_rows=rows, step_rowcount=2)
+    out = hdm.merge_duplicate_houses(db, dry_run=True)  # type: ignore[arg-type]
+    assert out["dry_run"] == 1
+    assert out["clusters_merged"] == 1  # both losers → one keeper
+    assert out["losers_deleted"] == 2
+    assert out["listings_repointed"] == 2  # the 'listings' step rowcount
+    assert db.commits == 0  # dry-run NEVER commits
+    assert db.rollbacks == 1
+
+
+def test_real_merge_commits() -> None:
+    """dry_run=False with dups → exactly one commit, no rollback."""
+    rows = [_Row(2, 1, "ул. мира, 10")]
+    db = _FakeDB(mapping_rows=rows, step_rowcount=1)
+    out = hdm.merge_duplicate_houses(db, dry_run=False)  # type: ignore[arg-type]
+    assert out["dry_run"] == 0
+    assert out["losers_deleted"] == 1
+    assert db.commits == 1
+    assert db.rollbacks == 0
+
+
+# ── Run-lifecycle wrapper ─────────────────────────────────────────────────────
+
+
+def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None:
+    marked: dict[str, Any] = {}
+    from app.services import scrape_runs as runs_mod
+
+    monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(
+        runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None)
+    monkeypatch.setattr(
+        hdm,
+        "merge_duplicate_houses",
+        lambda _db, dry_run=False: {"clusters_merged": 3, "losers_deleted": 5, "dry_run": 0},
+    )
+
+    out = hdm.run_house_dedup_merge(object(), run_id=42, params={"dry_run": False})  # type: ignore[arg-type]
+    assert out["clusters_merged"] == 3
+    assert marked["run_id"] == 42
+    assert marked["counters"]["losers_deleted"] == 5
+
+
+def test_run_wrapper_passes_dry_run_param(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+    from app.services import scrape_runs as runs_mod
+
+    monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None)
+
+    def _fake_merge(_db: Any, dry_run: bool = False) -> dict[str, int]:
+        captured["dry_run"] = dry_run
+        return {"dry_run": int(dry_run)}
+
+    monkeypatch.setattr(hdm, "merge_duplicate_houses", _fake_merge)
+    hdm.run_house_dedup_merge(object(), run_id=1, params={"dry_run": True})  # type: ignore[arg-type]
+    assert captured["dry_run"] is True
+
+
+def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    failed: dict[str, Any] = {}
+    from app.services import scrape_runs as runs_mod
+
+    monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(
+        runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
+    )
+
+    def _boom(_db: Any, dry_run: bool = False) -> dict[str, int]:
+        raise RuntimeError("merge exploded")
+
+    monkeypatch.setattr(hdm, "merge_duplicate_houses", _boom)
+
+    class _DB:
+        def rollback(self) -> None:
+            pass
+
+    with pytest.raises(RuntimeError):
+        hdm.run_house_dedup_merge(_DB(), run_id=9, params={})  # type: ignore[arg-type]
+    assert failed["run_id"] == 9
+    assert "merge exploded" in failed["err"]
+
+
+# ── Scheduler wiring ──────────────────────────────────────────────────────────
+
+
+def test_scheduler_has_trigger_and_dispatch() -> None:
+    assert hasattr(scheduler, "trigger_house_dedup_merge_run")
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source == "house_dedup_merge"' in loop_src
+    assert "trigger_house_dedup_merge_run(db, sch)" in loop_src
+
+
+def test_trigger_claims_run_and_runs_in_executor() -> None:
+    src = inspect.getsource(scheduler.trigger_house_dedup_merge_run)
+    assert "_claim_run(db, schedule_row)" in src
+    assert "if run_id is None:" in src
+    # sync DB-only task → run_in_executor (mirror cadastral_geo_match, not a bare await).
+    assert "run_in_executor" in src
+    assert "run_house_dedup_merge" in src
+    # fresh session inside the spawned task + RUF006 keep-alive callback.
+    assert "run_db = SessionLocal()" in src
+    assert "task.add_done_callback(" in src
+    assert "run_db.close()" in src
+
+
+# ── Migration content ─────────────────────────────────────────────────────────
+
+
+def test_migration_135_seeds_schedule_disabled() -> None:
+    """The deploy must be NEUTRAL — schedule seeded enabled=false (DORMANT)."""
+    sql = _MIGRATION_135.read_text(encoding="utf-8")
+    assert "INSERT INTO scrape_schedules" in sql
+    assert "'house_dedup_merge'" in sql
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+    assert "BEGIN;" in sql and "COMMIT;" in sql
+    # enabled=false in the VALUES (not enabled=true). Find the source line's VALUES tuple.
+    flat = re.sub(r"\s+", " ", sql)
+    m = re.search(r"'house_dedup_merge',\s*(\w+)", flat)
+    assert m is not None and m.group(1) == "false", "schedule MUST be seeded enabled=false"
+
+
+def test_migration_135_weekly_window_and_dry_run_param() -> None:
+    sql = _MIGRATION_135.read_text(encoding="utf-8")
+    # weekly cadence + dry_run default param present.
+    assert '"interval_days": 7' in sql
+    assert '"dry_run": false' in sql
+    # next_run_at NOT NULL = tomorrow (seed convention; harmless while disabled).
+    assert "CURRENT_DATE + INTERVAL '1 day'" in sql
+    assert "make_interval(hours => 4)" in sql
+
+
+def test_migration_135_no_psycopg_v3_colon_colon_cast() -> None:
+    sql = _MIGRATION_135.read_text(encoding="utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+# ── Optional real-Postgres behavioural merge test (self-skips without a DB) ────
+
+
+def _live_session() -> Any | None:
+    """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None."""
+    try:
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+        if not dsn or "localhost:5432/test" in dsn:
+            return None
+        engine = create_engine(dsn, future=True)
+        conn = engine.connect()
+        from sqlalchemy import text as _t
+
+        conn.execute(_t("SELECT 1"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
+    """End-to-end on a real DB: 2 houses sharing an address + colliding children →
+    merge re-points listings, dedups the colliding children WITHOUT a UNIQUE violation,
+    deletes the loser, keeper survives with merged data; second run is a no-op; dry_run
+    makes no writes."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    try:
+        # Two houses at the SAME address. Keeper (geom present) should win.
+        db.execute(
+            _t(
+                "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES "
+                "(900001, 'avito', 'EXT-KEEP', 'тестдом 1772, 1', 56.84, 60.60),"
+                "(900002, 'cian',  'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)"
+            )
+        )
+        # listings pointing at BOTH (the loser's must be re-pointed). source_url, dedup_hash,
+        # price_rub are NOT NULL with no default (002_core_tables.sql); dedup_hash is also
+        # UNIQUE, so each row needs a distinct value.
+        db.execute(
+            _t(
+                "INSERT INTO listings "
+                "(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) "
+                "VALUES "
+                "(910001, 'avito', 'http://t/1772/k', 'L-KEEP', 'dh-1772-keep', 5000000, 900001),"
+                "(910002, 'cian',  'http://t/1772/l', 'L-LOSE', 'dh-1772-lose', 6000000, 900002)"
+            )
+        )
+        # COLLIDING children: both houses have a price_dynamics row whose 5 dimension fields
+        # (source, room_count, prices_type, period, month_date) are IDENTICAL and only house_id
+        # differs (loser vs keeper). After the loser re-points to the keeper, both rows would
+        # share the LIVE 6-col UNIQUE key → the dedup step must drop the loser's row first.
+        # price_per_sqm is the real NOT NULL value column (024_houses_price_dynamics.sql);
+        # there is no avg_price column.
+        db.execute(
+            _t(
+                "INSERT INTO houses_price_dynamics "
+                "(house_id, month_date, source, room_count, prices_type, period, price_per_sqm) "
+                "VALUES "
+                "(900001, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 100000),"
+                "(900002, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 999999)"
+            )
+        )
+        # house_sources UNIQUE(ext_source, ext_id) is GLOBAL (not scoped to house_id), so two
+        # rows can never share the key at insert time — a true same-key collision is impossible
+        # to seed. We give the loser and keeper DISTINCT source rows; the merge must re-point the
+        # loser's onto the keeper (the collision-delete guard is a defensive no-op here, and both
+        # rows survive on the keeper).
+        db.execute(
+            _t(
+                "INSERT INTO house_sources "
+                "(house_id, ext_source, ext_id, confidence, matched_method) VALUES "
+                "(900001, 'avito', 'SRC-KEEP', 1.0, 'test'),"
+                "(900002, 'cian',  'SRC-LOSE', 1.0, 'test')"
+            )
+        )
+        db.commit()
+
+        # ── dry_run makes NO writes ──
+        before = db.execute(_t("SELECT count(*) FROM houses WHERE id IN (900001,900002)")).scalar()
+        dry = hdm.merge_duplicate_houses(db, dry_run=True)
+        assert dry["losers_deleted"] >= 1
+        after_dry = db.execute(
+            _t("SELECT count(*) FROM houses WHERE id IN (900001,900002)")
+        ).scalar()
+        assert after_dry == before == 2, "dry_run must not delete anything"
+
+        # ── real merge ──
+        out = hdm.merge_duplicate_houses(db, dry_run=False)
+        assert out["losers_deleted"] >= 1
+        # loser gone, keeper survives.
+        survivors = db.execute(
+            _t("SELECT id FROM houses WHERE id IN (900001,900002) ORDER BY id")
+        ).all()
+        assert [r.id for r in survivors] == [900001]
+        # loser's listing re-pointed to keeper.
+        repointed = db.execute(
+            _t("SELECT house_id_fk FROM listings WHERE id = 910002")
+        ).scalar()
+        assert repointed == 900001
+        # price_dynamics deduped — exactly one row for the keeper at that 6-col key (no dup),
+        # and it is the KEEPER's own row (price_per_sqm=100000), not the loser's (999999):
+        # the dedup keeps keeper rows (keeper_id IS NULL DESC) over loser rows.
+        pd_rows = db.execute(
+            _t(
+                "SELECT price_per_sqm FROM houses_price_dynamics "
+                "WHERE house_id = 900001 AND month_date = DATE '2026-01-01' "
+                "AND source='cian' AND room_count='all' AND prices_type='priceSqm' "
+                "AND period='allTime'"
+            )
+        ).all()
+        assert len(pd_rows) == 1, "colliding price_dynamics must be deduped, not duplicated"
+        assert pd_rows[0].price_per_sqm == 100000, "dedup must keep the keeper's own row"
+        # house_sources: the loser's source row re-pointed onto the keeper — both now on 900001
+        # (plus the backfill row for the keeper's own avito ext_house_id). The loser's row must
+        # be present and re-pointed; no row may dangle on the deleted loser.
+        hs_on_keeper = db.execute(
+            _t(
+                "SELECT count(*) FROM house_sources "
+                "WHERE house_id = 900001 AND ext_id IN ('SRC-KEEP','SRC-LOSE')"
+            )
+        ).scalar()
+        assert hs_on_keeper == 2, "both source rows must be re-pointed onto the keeper"
+        hs_dangling = db.execute(
+            _t("SELECT count(*) FROM house_sources WHERE house_id = 900002")
+        ).scalar()
+        assert hs_dangling == 0, "no source row may remain on the deleted loser"
+
+        # ── idempotent second run = no-op ──
+        again = hdm.merge_duplicate_houses(db, dry_run=False)
+        assert again["losers_deleted"] == 0
+    finally:
+        # cleanup: drop listings first (house_id_fk has no CASCADE), then the houses — which
+        # CASCADE-deletes house_sources / price_dynamics / aliases / etc. Also sweep any
+        # backfill house_sources rows the merge may have inserted for the keeper.
+        db.rollback()
+        db.execute(_t("DELETE FROM listings WHERE id IN (910001,910002)"))
+        db.execute(
+            _t("DELETE FROM house_sources WHERE ext_id IN ('SRC-KEEP','SRC-LOSE','EXT-KEEP')")
+        )
+        db.execute(
+            _t(
+                "DELETE FROM house_address_aliases "
+                "WHERE normalized_address = 'тестдом 1772, 1'"
+            )
+        )
+        db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)"))
+        db.commit()
+        db.close()