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

@@ -92,6 +92,30 @@ BACKFILL (reduces recurrence):
     (same as 108) so the matching pipeline's Tier-1/Tier-2 finds the keeper next scrape and
     does not immediately re-split it.
 
+MERGE JOURNAL — the merge is REVERSIBLE (#2690, migration 230):
+    Every loser gets a row in `house_merge_log` written in the SAME transaction as the merge:
+    the full jsonb snapshot of the deleted row, the keeper's snapshot BEFORE the identity
+    carry-over, the ids of every child row whose FK moved, the full snapshots of every child row
+    a UNIQUE collision destroyed, plus the grounds — which pass, which cluster-key VALUE fired,
+    whether the geo guard was on, and the keeper↔loser distance in metres.
+
+    This exists because the merge used to leave no restorable trace: losers were hard-deleted
+    with their children and the only record of «what went into what» was a log line, in a
+    container whose logs rotate faster than a day. A day after a run nobody could even NAME the
+    pairs, and the only rollback was restoring the whole database.
+
+    Undo: `SELECT * FROM house_merge_undo(batch_id)` inside a transaction — restores the loser
+    rows, points the children back, re-inserts the destroyed children, and un-does the identity
+    carry-over on the keeper, reporting per record what it could and could not restore.
+
+    NOTE the journal is deliberately NEUTRAL to the merge rule: it changes no cluster key, no
+    keeper rule and no guard. It only makes whatever the pass decides reversible — which is the
+    precondition for revisiting those decisions at all (#2690, #1772).
+
+    distance_m is recorded on BOTH passes, including the fias pass whose geo guard is off. That
+    asymmetry — merge allowed without a proximity check — was invisible in data before; now
+    «how many merges happened beyond N metres, on which key» is one query.
+
 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.
@@ -105,8 +129,10 @@ psycopg v3: all SQL uses CAST(:x AS type), never the colon-colon bound-param cas
 
 from __future__ import annotations
 
+import json
 import logging
 import time
+import uuid
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -260,7 +286,16 @@ def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> str:
     -- DIFFERENT house_fias_id — provably different buildings the cluster key collapsed (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 carries a fias.
-    SELECT id AS loser_id, keeper_id, norm_address
+    --
+    -- cluster_key / distance_m are carried out of the mapping for the MERGE JOURNAL (#2690):
+    -- cluster_key records WHICH key value fired, distance_m how far apart the two rows were.
+    -- distance_m is computed even when the geo guard is OFF for this pass — that is precisely
+    -- the case where nothing else records the distance, and #2690 had no way to ask
+    -- «how many merges happened at distances the guard would have blocked» from data.
+    SELECT id AS loser_id, keeper_id, norm_address, cluster_key,
+           CASE WHEN keeper_geom IS NOT NULL AND loser_geom IS NOT NULL
+                THEN ST_DistanceSphere(loser_geom, keeper_geom)
+                END AS distance_m
     FROM ranked
     WHERE rn > 1
       AND id <> keeper_id{geo_guard}
@@ -289,6 +324,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id_fk = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE l.house_id_fk = m.loser_id
+        RETURNING m.loser_id, l.id AS child_id
         """,
     ),
     (
@@ -298,6 +334,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hph.house_id = m.loser_id
+        RETURNING m.loser_id, hph.id AS child_id
         """,
     ),
     (
@@ -307,6 +344,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hr.house_id = m.loser_id
+        RETURNING m.loser_id, hr.id AS child_id
         """,
     ),
     (
@@ -316,6 +354,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hrc.house_id = m.loser_id
+        RETURNING m.loser_id, hrc.id AS child_id
         """,
     ),
     (
@@ -325,6 +364,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE ev.house_id = m.loser_id
+        RETURNING m.loser_id, ev.id AS child_id
         """,
     ),
     # ── UNIQUE(ext_source, ext_id): delete colliding losers, re-point rest ─────
@@ -340,6 +380,7 @@ _STEPS: list[tuple[str, str]] = [
                 AND hs2.ext_source = hs.ext_source
                 AND hs2.ext_id = hs.ext_id
           )
+        RETURNING hs.house_id AS loser_id, to_jsonb(hs.*) AS row_snapshot
         """,
     ),
     (
@@ -349,6 +390,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hs.house_id = m.loser_id
+        RETURNING m.loser_id, hs.id AS child_id
         """,
     ),
     # ── UNIQUE(normalized_address): delete colliding losers, re-point rest ─────
@@ -363,6 +405,7 @@ _STEPS: list[tuple[str, str]] = [
               WHERE haa2.house_id = m.keeper_id
                 AND haa2.normalized_address = haa.normalized_address
           )
+        RETURNING haa.house_id AS loser_id, to_jsonb(haa.*) AS row_snapshot
         """,
     ),
     (
@@ -372,6 +415,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE haa.house_id = m.loser_id
+        RETURNING m.loser_id, haa.id AS child_id
         """,
     ),
     # ── UNIQUE(house_id, source, room_count, prices_type, period, month_date) ──
@@ -393,6 +437,7 @@ _STEPS: list[tuple[str, str]] = [
             LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
         ) d
         WHERE t.id = d.id AND d.rn > 1
+        RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
         """,
     ),
     (
@@ -402,6 +447,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hpd.house_id = m.loser_id
+        RETURNING m.loser_id, hpd.id AS child_id
         """,
     ),
     # ── UNIQUE(house_id): one evaluation per keeper ───────────────────────────
@@ -419,6 +465,7 @@ _STEPS: list[tuple[str, str]] = [
             LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
         ) d
         WHERE t.id = d.id AND d.rn > 1
+        RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
         """,
     ),
     (
@@ -428,6 +475,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hie.house_id = m.loser_id
+        RETURNING m.loser_id, hie.id AS child_id
         """,
     ),
     # ── UNIQUE(house_id, ext_item_id) ─────────────────────────────────────────
@@ -445,6 +493,7 @@ _STEPS: list[tuple[str, str]] = [
             LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
         ) d
         WHERE t.id = d.id AND d.rn > 1
+        RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
         """,
     ),
     (
@@ -454,6 +503,7 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE hs.house_id = m.loser_id
+        RETURNING m.loser_id, hs.id AS child_id
         """,
     ),
     # ── UNIQUE(house_id, audit_batch) ─────────────────────────────────────────
@@ -471,6 +521,7 @@ _STEPS: list[tuple[str, str]] = [
             LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
         ) d
         WHERE t.id = d.id AND d.rn > 1
+        RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
         """,
     ),
     (
@@ -480,10 +531,98 @@ _STEPS: list[tuple[str, str]] = [
         SET house_id = m.keeper_id
         FROM _1772_dup_mapping m
         WHERE ama.house_id = m.loser_id
+        RETURNING m.loser_id, ama.id AS child_id
         """,
     ),
 ]
 
+# ── MERGE JOURNAL (#2690) ─────────────────────────────────────────────────────
+#
+# Every child of houses(id) except `listings` references it through a column named house_id;
+# listings uses house_id_fk. The undo function reads the column name back out of the journal
+# key ("таблица.колонка"), so this mapping is what makes the reverse UPDATE possible.
+_FK_COLUMN = {"listings": "house_id_fk"}
+
+# The (table, column) pairs the _STEPS pipeline actually handles, derived FROM the steps so the
+# set cannot drift away from them. Compared against pg_catalog before every merge — see
+# _assert_all_fk_children_handled.
+_HANDLED_CHILDREN: frozenset[tuple[str, str]] = frozenset(
+    (tbl, _FK_COLUMN.get(tbl, "house_id")) for tbl in {label.split("(")[0] for label, _ in _STEPS}
+)
+
+# Live FK children of houses(id), read from the catalog rather than trusted from a comment.
+_FK_CHILDREN_SQL = text(
+    """
+    SELECT CAST(CAST(c.conrelid AS regclass) AS text) AS child_table,
+           a.attname AS fk_column
+    FROM pg_constraint c
+    JOIN unnest(c.conkey) AS k(attnum) ON true
+    JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = k.attnum
+    WHERE c.confrelid = CAST('houses' AS regclass)
+      AND c.contype = 'f'
+    """
+)
+
+# One journal row per loser, written from the mapping BEFORE anything is mutated — so loser_row
+# is the row as it stood, and keeper_before precedes the identity carry-over.
+_JOURNAL_INSERT_SQL = text(
+    """
+    INSERT INTO house_merge_log (
+        batch_id, run_id, initiator, merge_pass, cluster_key, geo_guard, distance_m,
+        norm_address, loser_id, keeper_id, loser_row, keeper_before
+    )
+    SELECT
+        CAST(:batch_id AS uuid),
+        CAST(:run_id AS bigint),
+        CAST(:initiator AS text),
+        CAST(:merge_pass AS text),
+        m.cluster_key,
+        CAST(:geo_guard AS boolean),
+        m.distance_m,
+        m.norm_address,
+        m.loser_id,
+        m.keeper_id,
+        to_jsonb(l.*),
+        to_jsonb(k.*)
+    FROM _1772_dup_mapping m
+    JOIN houses l ON l.id = m.loser_id
+    JOIN houses k ON k.id = m.keeper_id
+    """
+)
+
+# Child bookkeeping lands after the steps ran — only then is it known which rows moved and which
+# were destroyed by a UNIQUE collision.
+_JOURNAL_CHILDREN_SQL = text(
+    """
+    UPDATE house_merge_log
+    SET children_repointed = CAST(:children_repointed AS jsonb),
+        children_deleted   = CAST(:children_deleted AS jsonb)
+    WHERE batch_id = CAST(:batch_id AS uuid)
+      AND loser_id = CAST(:loser_id AS bigint)
+    """
+)
+
+
+def _assert_all_fk_children_handled(db: Session) -> None:
+    """Fail the merge if houses(id) gained an FK child the _STEPS pipeline does not handle.
+
+    This is what makes the journal's promise true rather than merely documented. An unhandled
+    child is not a cosmetic gap: 9 of the 11 FKs are ON DELETE CASCADE, so `DELETE FROM houses`
+    would destroy its rows silently — no re-point step touches them, no RETURNING records them,
+    and the journal would claim a complete snapshot it does not have. Migration 133 already
+    broke on prod for exactly this (a missed child); there the failure was loud. Here it would
+    be silent, which is worse. Aborting the transaction costs one skipped merge cycle.
+    """
+    live = {(r.child_table, r.fk_column) for r in db.execute(_FK_CHILDREN_SQL).all()}
+    unhandled = live - _HANDLED_CHILDREN
+    if unhandled:
+        raise RuntimeError(
+            "merge_duplicate_houses: houses(id) has FK children the merge does not handle: "
+            f"{sorted(unhandled)}. Their rows would be CASCADE-deleted without a journal entry. "
+            "Add a re-point step to _STEPS (and its RETURNING) before merging again."
+        )
+
+
 # Delete the loser houses — all FK children are re-pointed or CASCADE by now.
 _DELETE_LOSERS_SQL = text(
     """
@@ -607,14 +746,20 @@ def _run_merge_pass(
     *,
     build_sql: Any,
     pass_label: str,
+    geo_guard: bool,
+    batch_id: str,
+    run_id: int | None,
+    initiator: str,
     result: DedupMergeResult,
 ) -> None:
     """Run ONE merge pass (fias- or canon-key) inside the caller's open transaction.
 
-    Builds a fresh loser→keeper mapping for this pass's cluster key, re-points every FK child
+    Builds a fresh loser→keeper mapping for this pass's cluster key, writes the MERGE JOURNAL
-    (UNIQUE-collision-safe), carries identity/enrichment onto the keeper, deletes the losers and
+    (#2690), re-points every FK child (UNIQUE-collision-safe), carries identity/enrichment onto
-    backfills sources/aliases. Accumulates counters onto `result`. NEVER commits/rolls back — the
+    the keeper, deletes the losers and backfills sources/aliases. Accumulates counters onto
-    caller owns the single transaction wrapping both passes.
+    `result`. NEVER commits/rolls back — the caller owns the single transaction wrapping both
+    passes, which is also what makes the journal atomic with the merge: there is no ordering in
+    which the rows vanish but the journal entry does not land (and dry_run rolls back both).
     """
     # 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.
@@ -623,8 +768,8 @@ def _run_merge_pass(
 
     mapping = db.execute(
         text(
-            "SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping "
+            "SELECT loser_id, keeper_id, norm_address, cluster_key, distance_m "
-            "ORDER BY keeper_id, loser_id"
+            "FROM _1772_dup_mapping ORDER BY keeper_id, loser_id"
         )
     ).all()
     if not mapping:
@@ -634,32 +779,73 @@ def _run_merge_pass(
     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.
+    # JOURNAL, phase 1 — snapshot loser + keeper BEFORE any statement mutates them.
+    db.execute(
+        _JOURNAL_INSERT_SQL,
+        {
+            "batch_id": batch_id,
+            "run_id": run_id,
+            "initiator": initiator,
+            "merge_pass": pass_label,
+            "geo_guard": geo_guard,
+        },
+    )
+
+    # Container logs rotate faster than a day (#2690), so this line is a convenience, not the
+    # record — house_merge_log is. Distance is logged too: it is the one number that says
+    # whether a merge would have survived the geo guard.
     for row in mapping:
         logger.info(
-            "merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r",
+            "merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r "
+            "distance_m=%s batch=%s",
             pass_label,
             row.loser_id,
             row.keeper_id,
             row.norm_address,
+            "n/a" if row.distance_m is None else f"{row.distance_m:.0f}",
+            batch_id,
         )
 
+    # Per-loser child bookkeeping, collected from each step's RETURNING: survivors by id (the
+    # rows are intact, only their FK moved), destroyed rows by full snapshot (nothing else is
+    # left of them).
+    repointed: dict[int, dict[str, list[int]]] = {}
+    deleted: dict[int, dict[str, list[Any]]] = {}
+
     for label, sql in _STEPS:
-        res = db.execute(text(sql))
+        rows = db.execute(text(sql)).all()
-        rowcount = res.rowcount or 0
+        rowcount = len(rows)
-        if label == "listings":
+        table = label.split("(")[0]
-            result.listings_repointed += rowcount
+        if label.endswith("(collision-delete)") or label.endswith("(dedup)"):
-        elif label.endswith("(collision-delete)") or label.endswith("(dedup)"):
             result.children_deleted += rowcount
-        elif label.endswith("(re-point)") or label in (
+            for r in rows:
-            "house_placement_history",
+                deleted.setdefault(r.loser_id, {}).setdefault(table, []).append(r.row_snapshot)
-            "house_reviews",
+        else:
-            "house_reliability_checks",
+            key = f"{table}.{_FK_COLUMN.get(table, 'house_id')}"
-            "external_valuations",
+            for r in rows:
-        ):
+                repointed.setdefault(r.loser_id, {}).setdefault(key, []).append(r.child_id)
-            result.children_repointed += rowcount
+            if label == "listings":
+                result.listings_repointed += rowcount
+            else:
+                result.children_repointed += rowcount
         logger.debug("merge_duplicate_houses: pass=%s step=%s rows=%d", pass_label, label, rowcount)
 
+    # JOURNAL, phase 2 — attach the child bookkeeping to the rows written in phase 1.
+    touched = sorted(set(repointed) | set(deleted))
+    if touched:
+        db.execute(
+            _JOURNAL_CHILDREN_SQL,
+            [
+                {
+                    "batch_id": batch_id,
+                    "loser_id": loser_id,
+                    "children_repointed": json.dumps(repointed.get(loser_id, {})),
+                    "children_deleted": json.dumps(deleted.get(loser_id, {}), default=str),
+                }
+                for loser_id in touched
+            ],
+        )
+
     # Carry identity/enrichment onto the keeper BEFORE the losers vanish, then delete + backfill.
     db.execute(_CARRY_OVER_IDENTITY_SQL)
     db.execute(_DELETE_LOSERS_SQL)
@@ -667,7 +853,13 @@ def _run_merge_pass(
     db.execute(_BACKFILL_ALIASES_SQL)
 
 
-def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
+def merge_duplicate_houses(
+    db: Session,
+    *,
+    dry_run: bool = False,
+    run_id: int | None = None,
+    initiator: str = "manual",
+) -> 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:
@@ -680,16 +872,41 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i
     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.
 
+    Every deleted row is journaled to house_merge_log in the SAME transaction (#2690), so a
+    merge is reversible via house_merge_undo(batch_id); the batch_id is returned in the log line
+    and stored on every journal row of this call.
+
     Returns the counter dict (DedupMergeResult.to_counters()).
     """
     start = time.monotonic()
     result = DedupMergeResult(dry_run=dry_run)
+    batch_id = str(uuid.uuid4())
 
     try:
+        # Refuse to merge at all if some FK child would be CASCADE-destroyed unjournaled.
+        _assert_all_fk_children_handled(db)
         # 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)
+        _run_merge_pass(
+            db,
+            build_sql=_BUILD_MAPPING_SQL_FIAS,
+            pass_label="fias",
+            geo_guard=False,
+            batch_id=batch_id,
+            run_id=run_id,
+            initiator=initiator,
+            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)
+        _run_merge_pass(
+            db,
+            build_sql=_BUILD_MAPPING_SQL,
+            pass_label="canon",
+            geo_guard=True,
+            batch_id=batch_id,
+            run_id=run_id,
+            initiator=initiator,
+            result=result,
+        )
 
         if result.losers_deleted == 0:
             # Clean table — both passes empty. Roll back (we only opened temp tables).
@@ -716,12 +933,15 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i
             db.commit()
             logger.info(
                 "merge_duplicate_houses: COMMITTED clusters=%d losers=%d "
-                "listings_repointed=%d children_deleted=%d children_repointed=%d",
+                "listings_repointed=%d children_deleted=%d children_repointed=%d "
+                "batch_id=%s (undo: SELECT * FROM house_merge_undo('%s'))",
                 result.clusters_merged,
                 result.losers_deleted,
                 result.listings_repointed,
                 result.children_deleted,
                 result.children_repointed,
+                batch_id,
+                batch_id,
             )
     except Exception:
         logger.exception("merge_duplicate_houses: FAILED — rolling back")
@@ -761,7 +981,7 @@ def run_house_dedup_merge(db: Session, *, run_id: int, params: dict) -> dict[str
     }
     try:
         runs_mod.update_heartbeat(db, run_id, counters)
-        counters = merge_duplicate_houses(db, dry_run=dry_run)
+        counters = merge_duplicate_houses(db, dry_run=dry_run, run_id=run_id, initiator="schedule")
         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",

tradein-mvp/backend/data/sql/230_house_merge_log.sql

@@ -0,0 +1,262 @@
+-- 230_house_merge_log.sql
+-- Журнал слияний домов + обратная операция (#2690).
+--
+-- WHY:
+--   `house_dedup_merge` — НЕ спящая идея, а живой деструктивный проход: расписание
+--   `house_dedup_merge` на проде enabled=true, dry_run=false, такт 7 дней. Шесть прогонов
+--   с 2026-06-27 уже удалили 119 строк `houses` (счётчики losers_deleted в scrape_runs:
+--   2/39/31/9/6/32). Единственным следом «кто в кого» была строка `logger.info` в контейнере,
+--   а логи ротируются быстрее суток. То есть **уже сегодня** нельзя назвать, какой дом в какой
+--   свернули 1 августа, — не говоря о том, чтобы вернуть.
+--
+--   Пока этого журнала нет, любой разговор о расширении ключа схлопывания (#2690, #1772)
+--   ведётся без права на ошибку: единственный откат — restore всей БД на момент до прогона,
+--   т.е. выброс недели сбора. Журнал снимает это условие: слияние становится обратимым,
+--   и вопрос о ключе можно пересматривать, а не решать «навсегда».
+--
+--   Правку НЕ следует читать как одобрение текущего ключа/победителя/гео-стража. Она к ним
+--   НЕЙТРАЛЬНА: ни ключ, ни правило выбора победителя, ни гео-страж здесь не меняются.
+--   Меняется только одно — теперь есть что откатить.
+--
+-- WHAT (одна строка = один проигравший дом):
+--   merge_pass / cluster_key / geo_guard / distance_m — ОСНОВАНИЕ слияния. Это не косметика:
+--       ровно этих полей не хватило в #2690, чтобы ответить на вопрос «сколько слияний прошло
+--       на расстояниях, которые гео-страж заблокировал бы» по ДАННЫМ, а не по ревью. distance_m
+--       пишется всегда, даже когда страж для прохода выключен (fias-проход) — тогда он и есть
+--       единственная запись о том, насколько далеко разъехались объединённые дома.
+--   loser_row     — ПОЛНЫЙ jsonb-снимок удаляемой строки (`to_jsonb(h.*)`, все 86 колонок).
+--       Ссылка на удалённую строку бесполезна, поэтому хранится содержимое. Снимок целиком,
+--       а не список полей: проверено, что `jsonb_populate_record(NULL::houses, loser_row)`
+--       восстанавливает строку побайтово, включая PostGIS-geom (to_jsonb отдаёт её GeoJSON'ом,
+--       populate_record разбирает обратно входной функцией типа). Побочная выгода: новая
+--       колонка в `houses` попадает в снимок и в откат САМА, без правки этой миграции.
+--   keeper_before — снимок ПОБЕДИТЕЛЯ до переноса метаданных. Нужен, потому что слияние не
+--       только удаляет проигравшего: `_CARRY_OVER_IDENTITY_SQL` дозаполняет победителю NULL-поля
+--       идентичности (fias/кадастр/ГАР/DaData) значениями проигравшего. Без этого снимка откат
+--       вернул бы дом, но оставил бы его ФИАС на победителе — и следующий же fias-проход слил
+--       бы их обратно.
+--   children_repointed — {"таблица.колонка": [id, ...]}. Дочерние строки ПЕРЕЖИЛИ слияние,
+--       у них сменилась только ссылка, поэтому хранятся id, а не содержимое (иначе одни
+--       listings с их raw-payload'ом дали бы ~7 КБ на строку вместо ~8 байт на id).
+--   children_deleted   — {"таблица": [{строка целиком}, ...]}. Дочерние строки, которые проход
+--       УДАЛИЛ из-за коллизии по UNIQUE. Их содержимое уничтожено, id недостаточно — только
+--       полный снимок. Таких таблиц шесть (см. _STEPS), строки мелкие.
+--   batch_id — один вызов merge_duplicate_houses() (оба прохода). Единица отката.
+--   run_id / initiator — кто инициировал: scrape_runs.id для расписания, NULL для ручного.
+--
+-- НАМЕРЕННО БЕЗ ВНЕШНИХ КЛЮЧЕЙ на houses(id) и scrape_runs(id):
+--   журнал обязан ПЕРЕЖИВАТЬ строки, которые описывает. loser_id указывает на заведомо
+--   удалённый дом. keeper_id — на дом, который сам может быть слит следующим прогоном; FK
+--   с CASCADE стёр бы историю ровно тогда, когда она нужнее всего, а FK без CASCADE
+--   заблокировал бы слияние. То же с run_id: чистка scrape_runs не должна трогать журнал.
+--
+-- ОБЪЁМ (замерено на проде 2026-08-06):
+--   9 571 дом, средняя строка houses в jsonb 2 581 Б. Строка журнала ≈ loser_row 2.5 КБ +
+--   keeper_before 2.5 КБ + списки id (в среднем 27.9 дочерних строк на дом × ~8 Б) ≈ 5.3 КБ.
+--   Наблюдаемый темп — 20 слияний в неделю (119 за 6 прогонов) ≈ 106 КБ/нед ≈ 5.5 МБ/год.
+--   Ближайший прогон (замер тем же выражением, что и код): 93 проигравших ≈ 0.5 МБ.
+--   Абсолютный потолок, если схлопнуть вообще все дома: 9 571 × 5.3 КБ ≈ 50 МБ против 23 МБ
+--   самой таблицы houses.
+--
+-- RETENTION: НЕ НУЖЕН, сознательно. Потолок роста — двузначные мегабайты, то есть дешевле
+--   любой процедуры чистки; а журнал слияний — это ровно то, что удалять не хочется: его
+--   ценность в том, что он отвечает на вопрос «что было год назад», когда логов давно нет.
+--   Если объём когда-нибудь станет проблемой, удалять надо не строки, а тяжёлые снимки
+--   (loser_row/keeper_before → NULL) у записей старше N лет, сохранив соответствие
+--   loser→keeper: оно весит байты и именно оно нужно дольше всего.
+--
+-- Dependencies: 002_core_tables.sql (houses), 135_scrape_schedules_seed_house_dedup_merge.sql
+-- Пишется в ТОЙ ЖЕ транзакции, что и слияние (см. house_dedup_merge._run_merge_pass) —
+-- разрыв «слияние прошло, запись не легла» невозможен по построению; dry_run откатывает и то,
+-- и другое вместе.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS house_merge_log (
+    id                 bigserial PRIMARY KEY,
+    merged_at          timestamptz NOT NULL DEFAULT now(),
+    batch_id           uuid        NOT NULL,
+    run_id             bigint,
+    initiator          text        NOT NULL,
+    merge_pass         text        NOT NULL,
+    cluster_key        text        NOT NULL,
+    geo_guard          boolean     NOT NULL,
+    distance_m         double precision,
+    norm_address       text,
+    loser_id           bigint      NOT NULL,
+    keeper_id          bigint      NOT NULL,
+    loser_row          jsonb       NOT NULL,
+    keeper_before      jsonb       NOT NULL,
+    children_repointed jsonb       NOT NULL DEFAULT '{}'::jsonb,
+    children_deleted   jsonb       NOT NULL DEFAULT '{}'::jsonb
+);
+
+CREATE INDEX IF NOT EXISTS idx_house_merge_log_loser  ON house_merge_log (loser_id);
+CREATE INDEX IF NOT EXISTS idx_house_merge_log_keeper ON house_merge_log (keeper_id);
+CREATE INDEX IF NOT EXISTS idx_house_merge_log_batch  ON house_merge_log (batch_id);
+
+COMMENT ON TABLE house_merge_log IS
+    'Журнал слияний домов (#2690): одна строка = один проигравший дом, удалённый проходом '
+    'house_dedup_merge. Пишется в ТОЙ ЖЕ транзакции, что и слияние. Содержит полный снимок '
+    'удалённой строки и перечень перенесённых/удалённых дочерних строк — достаточно, чтобы '
+    'назвать поимённо, что во что свернули, и вернуть обратно (house_merge_undo). Намеренно '
+    'БЕЗ FK на houses/scrape_runs: журнал переживает строки, которые описывает. Retention нет.';
+
+COMMENT ON COLUMN house_merge_log.cluster_key IS
+    'ЗНАЧЕНИЕ ключа, по которому дома попали в один кластер («addr:вайнера66» / «fias:<uuid>»), '
+    'а не имя ключа — по нему видно, какое именно совпадение сработало.';
+COMMENT ON COLUMN house_merge_log.geo_guard IS
+    'Был ли для этого прохода включён гео-страж 250 м. false = слияние разрешено БЕЗ проверки '
+    'близости; вместе с distance_m это и есть аудит основания (#2690).';
+COMMENT ON COLUMN house_merge_log.distance_m IS
+    'ST_DistanceSphere между победителем и проигравшим на момент слияния; NULL = у одной из '
+    'сторон не было geom. Пишется ВСЕГДА, в том числе когда гео-страж выключен.';
+COMMENT ON COLUMN house_merge_log.loser_row IS
+    'to_jsonb() удалённой строки houses целиком. Восстановление: '
+    'INSERT INTO houses SELECT r.* FROM jsonb_populate_record(NULL::houses, loser_row) r.';
+COMMENT ON COLUMN house_merge_log.keeper_before IS
+    'Снимок победителя ДО переноса метаданных с проигравшего (COALESCE-дозаполнение полей '
+    'идентичности). Без него откат вернул бы дом, но оставил его ФИАС/кадастр на победителе.';
+COMMENT ON COLUMN house_merge_log.children_repointed IS
+    '{"таблица.колонка": [id, ...]} — дочерние строки, у которых слияние сменило ссылку '
+    'loser→keeper. Строки целы, поэтому хранятся id: откат возвращает ссылку обратно.';
+COMMENT ON COLUMN house_merge_log.children_deleted IS
+    '{"таблица": [{строка целиком}, ...]} — дочерние строки, УДАЛЁННЫЕ проходом из-за коллизии '
+    'по UNIQUE с победителем. Содержимое уничтожено, поэтому хранится снимок, а не id.';
+
+-- ── Обратная операция ────────────────────────────────────────────────────────
+--
+-- Откат одного батча (или его части) по журналу. Транзакционен: вызывающий сам решает
+-- COMMIT/ROLLBACK, увидев отчёт. Возвращает СТРОКУ НА КАЖДУЮ запись журнала со статусом —
+-- в том числе «не смог», потому что молчаливо-успешный откат хуже отсутствующего.
+--
+-- Порядок внутри одной записи важен: сначала воскресить дом (на него ссылаются дети), потом
+-- вернуть ссылки детей, потом вернуть удалённых детей, потом снять перенос метаданных с
+-- победителя. Записи батча обходятся в обратном порядке (id DESC) — если дом A слили в B,
+-- а B потом в C, разматывать надо с конца.
+--
+-- ИЗВЕСТНЫЕ ГРАНИЦЫ (сознательные, отражены в статусе):
+--   * дочерняя строка, удалённая по коллизии, может не вернуться: место в UNIQUE-ключе занято
+--     строкой победителя. ON CONFLICT DO NOTHING + счётчик в статусе, а не тихая потеря;
+--   * backfill-строки house_sources/house_address_aliases, которые проход дописал победителю,
+--     НЕ удаляются: они собраны из собственных полей победителя и остались бы верны и без
+--     слияния;
+--   * если id проигравшего уже занят — запись пропускается со статусом, откат не гадает.
+CREATE OR REPLACE FUNCTION house_merge_undo(
+    p_batch       uuid,
+    p_only_losers bigint[] DEFAULT NULL
+)
+RETURNS TABLE (
+    out_log_id   bigint,
+    out_loser_id bigint,
+    out_keeper_id bigint,
+    out_status   text
+)
+LANGUAGE plpgsql
+AS $$
+DECLARE
+    rec           record;
+    v_table       text;
+    v_column      text;
+    v_ids         bigint[];
+    v_rows        jsonb;
+    v_field       text;
+    v_repointed   int;
+    v_restored    int;
+    v_lost        int;
+    v_n           int;
+    -- Список полей ДОЛЖЕН совпадать с SET в house_dedup_merge._CARRY_OVER_IDENTITY_SQL;
+    -- за расхождением следит тест test_undo_carryover_fields_match_merge_carryover.
+    c_carry_fields constant text[] := ARRAY[
+        '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'
+    ];
+BEGIN
+    FOR rec IN
+        SELECT *
+        FROM house_merge_log l
+        WHERE l.batch_id = p_batch
+          AND (p_only_losers IS NULL OR l.loser_id = ANY (p_only_losers))
+        ORDER BY l.id DESC
+    LOOP
+        out_log_id := rec.id;
+        out_loser_id := rec.loser_id;
+        out_keeper_id := rec.keeper_id;
+
+        IF EXISTS (SELECT 1 FROM houses h WHERE h.id = rec.loser_id) THEN
+            out_status := 'skipped: houses.id ' || rec.loser_id || ' занят — уже откачено?';
+            RETURN NEXT;
+            CONTINUE;
+        END IF;
+
+        -- 1. Воскресить проигравшего целиком из снимка (все колонки, включая geom).
+        INSERT INTO houses
+        SELECT r.* FROM jsonb_populate_record(NULL::houses, rec.loser_row) r;
+
+        -- 2. Вернуть ссылки уцелевших детей. Условие «сейчас указывает на победителя»
+        --    защищает от затирания строк, которые после слияния перепривязали чем-то ещё.
+        v_repointed := 0;
+        FOR v_table, v_column, v_ids IN
+            SELECT split_part(e.key, '.', 1),
+                   split_part(e.key, '.', 2),
+                   ARRAY(SELECT jsonb_array_elements_text(e.value)::bigint)
+            FROM jsonb_each(rec.children_repointed) AS e
+        LOOP
+            EXECUTE format(
+                'UPDATE %I SET %I = $1 WHERE id = ANY ($2) AND %I = $3',
+                v_table, v_column, v_column
+            ) USING rec.loser_id, v_ids, rec.keeper_id;
+            GET DIAGNOSTICS v_n = ROW_COUNT;
+            v_repointed := v_repointed + v_n;
+        END LOOP;
+
+        -- 3. Вернуть детей, удалённых по коллизии UNIQUE. Место могло остаться занятым
+        --    строкой победителя — тогда DO NOTHING, и это попадёт в отчёт как «не вернулось».
+        v_restored := 0;
+        v_lost := 0;
+        FOR v_table, v_rows IN
+            SELECT e.key, e.value FROM jsonb_each(rec.children_deleted) AS e
+        LOOP
+            EXECUTE format(
+                'INSERT INTO %I SELECT r.* FROM jsonb_array_elements($1) AS el, '
+                'LATERAL jsonb_populate_record(NULL::%I, el) r ON CONFLICT DO NOTHING',
+                v_table, v_table
+            ) USING v_rows;
+            GET DIAGNOSTICS v_n = ROW_COUNT;
+            v_restored := v_restored + v_n;
+            v_lost := v_lost + (jsonb_array_length(v_rows) - v_n);
+        END LOOP;
+
+        -- 4. Снять перенос метаданных с победителя. Только там, где до слияния было NULL И
+        --    текущее значение всё ещё РОВНО то, что принёс этот проигравший: если поле успел
+        --    заполнить загрузчик (или донором был другой проигравший кластера) — не трогаем.
+        --    Сравнение в jsonb-пространстве, чтобы один цикл покрыл text/int/timestamptz.
+        FOREACH v_field IN ARRAY c_carry_fields LOOP
+            IF rec.keeper_before ->> v_field IS NULL THEN
+                EXECUTE format(
+                    'UPDATE houses SET %I = NULL WHERE id = $1 AND to_jsonb(%I) = $2',
+                    v_field, v_field
+                ) USING rec.keeper_id, rec.loser_row -> v_field;
+            END IF;
+        END LOOP;
+
+        out_status := format(
+            'restored: дом %s вернулся, ссылок возвращено %s, дочерних строк восстановлено %s'
+            || CASE WHEN v_lost > 0 THEN ', НЕ ВЕРНУЛОСЬ ' || v_lost || ' (место занято)'
+                    ELSE '' END,
+            rec.loser_id, v_repointed, v_restored
+        );
+        RETURN NEXT;
+    END LOOP;
+END;
+$$;
+
+COMMENT ON FUNCTION house_merge_undo(uuid, bigint[]) IS
+    'Откат слияния домов по журналу house_merge_log (#2690). Аргументы: batch_id (единица '
+    'отката = один вызов merge_duplicate_houses) и опциональный список loser_id для частичного '
+    'отката. Возвращает строку-статус на КАЖДУЮ запись журнала, включая неудачные. '
+    'Транзакции не открывает и не закрывает — вызывающий смотрит отчёт и решает COMMIT/ROLLBACK: '
+    '  BEGIN; SELECT * FROM house_merge_undo(''<batch_id>''); -- прочитать статусы -- COMMIT;';
+
+COMMIT;

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -284,9 +284,18 @@ def test_fias_pass_drops_geo_guard_canon_pass_keeps_it() -> None:
     assert "keeper_geom IS NOT NULL" in canon
     assert "loser_geom IS NOT NULL" in canon
     # fias pass drops the distance guard AND the NULL-geom exclusions entirely.
-    assert "ST_DistanceSphere" not in fias
+    #
-    assert "loser_geom IS NOT NULL" not in fias
+    # Asserted on the guard PREDICATE, not on the bare function name: since #2690 the mapping also
-    assert "keeper_geom IS NOT NULL" not in fias
+    # MEASURES the keeper↔loser distance into `distance_m` for the merge journal, on both passes.
+    # Measuring is the opposite of guarding — the fias pass is precisely where nothing else records
+    # how far apart the merged rows were — so the name alone can no longer stand in for the guard.
+    assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" not in fias
+    guard = (
+        "AND keeper_geom IS NOT NULL AND loser_geom IS NOT NULL "
+        "AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250"
+    )
+    assert guard in canon
+    assert guard not in fias
     # the cross-fias anti-over-merge guard is untouched in the canon pass.
     assert "lower(loser_fias) <> lower(keeper_fias)" in canon
 
@@ -294,12 +303,15 @@ def test_fias_pass_drops_geo_guard_canon_pass_keeps_it() -> None:
 def test_mapping_sql_geo_guard_param_toggles_only_distance_filter() -> None:
     """_mapping_sql(apply_geo_guard=...) toggles ONLY the 250 m distance filter; the cross-fias
     guard is emitted regardless, and the default is True (canon-safe)."""
+    guard = "ST_DistanceSphere(loser_geom, keeper_geom) <= 250"
     with_guard = _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR, apply_geo_guard=True))
     without_guard = _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR, apply_geo_guard=False))
-    assert "ST_DistanceSphere" in with_guard
+    assert guard in with_guard
-    assert "ST_DistanceSphere" not in without_guard
+    assert guard not in without_guard
     # default = True (the canon pass must never lose its guard by omission).
-    assert "ST_DistanceSphere" in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR))
+    assert guard in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR))
+    # ...while the journal's distance MEASUREMENT is emitted either way (#2690).
+    assert "AS distance_m" in with_guard and "AS distance_m" in without_guard
     # cross-fias guard present in BOTH renderings (independent of the geo guard).
     assert "lower(loser_fias) <> lower(keeper_fias)" in with_guard
     assert "lower(loser_fias) <> lower(keeper_fias)" in without_guard
@@ -327,7 +339,9 @@ def test_both_passes_share_one_pipeline_no_copy_paste() -> None:
         assert token in canon and token in fias
     # the 250 m distance guard is CANON-ONLY (#2187) — fias identity outranks proximity.
     assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" in canon
-    assert "ST_DistanceSphere" not in fias
+    assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" not in fias
+    # ...but the journal's distance MEASUREMENT is on both — measuring is not guarding.
+    assert "AS distance_m" in canon and "AS distance_m" in fias
 
 
 def test_cross_fias_guard_blocks_slash_collapse_over_merge() -> None:
@@ -433,28 +447,62 @@ class _FakeResult:
 
 
 class _Row:
-    def __init__(self, loser_id: int, keeper_id: int, norm_address: str):
+    def __init__(
+        self,
+        loser_id: int,
+        keeper_id: int,
+        norm_address: str,
+        cluster_key: str = "addr:тест",
+        distance_m: float | None = 12.0,
+    ):
         self.loser_id = loser_id
         self.keeper_id = keeper_id
         self.norm_address = norm_address
+        # journal grounds (#2690): which key value fired, and how far apart the rows were.
+        self.cluster_key = cluster_key
+        self.distance_m = distance_m
+
+
+class _ChildRow:
+    """What a step's RETURNING yields: an id for a survivor, a snapshot for a destroyed row."""
+
+    def __init__(self, loser_id: int, child_id: int = 1):
+        self.loser_id = loser_id
+        self.child_id = child_id
+        self.row_snapshot = {"id": child_id, "house_id": loser_id}
+
+
+class _FKChild:
+    def __init__(self, child_table: str, fk_column: str):
+        self.child_table = child_table
+        self.fk_column = fk_column
 
 
 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):
+    def __init__(
+        self,
+        mapping_rows: list[_Row],
+        step_rowcount: int = 1,
+        fk_children: dict[str, str] | None = None,
+    ):
         self._mapping_rows = mapping_rows
         self._step_rowcount = step_rowcount
         self._mapping_served = False
+        # The catalog the FK-child guard reads; defaults to the real live set.
+        self._fk_children = _FK_CHILDREN if fk_children is None else fk_children
         self.commits = 0
         self.rollbacks = 0
         self.executed: list[str] = []
 
-    def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
+    def execute(self, clause: Any, params: Any = None) -> _FakeResult:
         sql = str(getattr(clause, "text", clause))
         self.executed.append(sql)
         if "CREATE TEMP TABLE" in sql:
             return _FakeResult()
+        if "FROM pg_constraint" in sql:
+            return _FakeResult(rows=[_FKChild(t, c) for t, c in self._fk_children.items()])
         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.
@@ -462,7 +510,13 @@ class _FakeDB:
                 return _FakeResult(rows=[])
             self._mapping_served = True
             return _FakeResult(rows=list(self._mapping_rows))
-        # any UPDATE/DELETE/INSERT step (incl. DROP TABLE, carry-over, delete, backfill)
+        # Steps now RETURN the rows they touched (journal, #2690) — one per scripted rowcount,
+        # attributed to the first loser so the per-loser bookkeeping has something to bucket.
+        if "RETURNING" in sql:
+            loser = self._mapping_rows[0].loser_id if self._mapping_rows else 0
+            rows = [_ChildRow(loser, child_id=i + 1) for i in range(self._step_rowcount)]
+            return _FakeResult(rowcount=self._step_rowcount, rows=rows)
+        # any other UPDATE/DELETE/INSERT (DROP TABLE, journal, carry-over, delete, backfill)
         return _FakeResult(rowcount=self._step_rowcount)
 
     def commit(self) -> None:
@@ -524,7 +578,11 @@ def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -
     monkeypatch.setattr(
         hdm,
         "merge_duplicate_houses",
-        lambda _db, dry_run=False: {"clusters_merged": 3, "losers_deleted": 5, "dry_run": 0},
+        lambda _db, dry_run=False, run_id=None, initiator="manual": {
+            "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]
@@ -541,13 +599,20 @@ def test_run_wrapper_passes_dry_run_param(monkeypatch: pytest.MonkeyPatch) -> No
     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]:
+    def _fake_merge(
+        _db: Any, dry_run: bool = False, run_id: int | None = None, initiator: str = "manual"
+    ) -> dict[str, int]:
         captured["dry_run"] = dry_run
+        captured["run_id"] = run_id
+        captured["initiator"] = initiator
         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
+    # the journal must be able to say WHICH run did it, and that it was not a human (#2690)
+    assert captured["run_id"] == 1
+    assert captured["initiator"] == "schedule"
 
 
 def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -562,7 +627,9 @@ def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> N
         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]:
+    def _boom(
+        _db: Any, dry_run: bool = False, run_id: int | None = None, initiator: str = "manual"
+    ) -> dict[str, int]:
         raise RuntimeError("merge exploded")
 
     monkeypatch.setattr(hdm, "merge_duplicate_houses", _boom)
@@ -642,12 +709,16 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
     db = _live_session()
     assert db is not None
     try:
-        # Two houses at the SAME address. Keeper (geom present) should win.
+        # Two houses at the SAME address, ~10 m apart (the #2187 canon geo guard needs geom
+        # on BOTH sides). Keeper = min(id) once geom and listing counts tie.
         db.execute(
             _t(
-                "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES "
+                # url is NOT NULL in houses (002_core_tables); nothing here asserts on it,
-                "(900001, 'avito', 'EXT-KEEP', 'тестдом 1772, 1', 56.84, 60.60),"
+                # so 'u' is a placeholder. These live-DB fixtures self-skip in CI, which is
-                "(900002, 'cian',  'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)"
+                # how they silently drifted out of sync with the schema in the first place.
+                "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon) VALUES "
+                "(900001, 'avito', 'EXT-KEEP','u', 'тестдом 1772, 1', 56.84, 60.60),"
+                "(900002, 'cian',  'EXT-LOSE','u', 'тестдом 1772, 1', 56.84009, 60.60)"
             )
         )
         # listings pointing at BOTH (the loser's must be re-pointed). source_url, dedup_hash,
@@ -755,6 +826,9 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
         db.execute(
             _t("DELETE FROM house_address_aliases WHERE normalized_address = 'тестдом 1772, 1'")
         )
+        # journal rows have no FK and are never cascaded away — sweep them explicitly,
+        # or a re-run accumulates them (all live fixtures live in the 9000xx id range).
+        db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299"))
         db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)"))
         db.commit()
         db.close()
@@ -781,16 +855,16 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None:
     try:
         db.execute(
             _t(
-                "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES "
+                "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon) VALUES "
                 # A — ул/улица spelling variants of the SAME building, ~10 m apart → MERGE
-                "(900010, 'avito', 'EXT-T-VK', 'улица Тестовая1772, 66', 56.84000, 60.60000),"
+                "(900010, 'avito', 'EXT-T-VK','u', 'улица Тестовая1772, 66', 56.84000, 60.60000),"
-                "(900011, 'cian',  'EXT-T-VL', 'ул. Тестовая1772, 66',   56.84009, 60.60000),"
+                "(900011, 'cian',  'EXT-T-VL','u', 'ул. Тестовая1772, 66',   56.84009, 60.60000),"
                 # B — same canon (ленина-like) but ~5 km apart → geo guard BLOCKS the merge
-                "(900012, 'avito', 'EXT-T-L1', 'улица Тестовая1772, 5',  56.84000, 60.60000),"
+                "(900012, 'avito', 'EXT-T-L1','u', 'улица Тестовая1772, 5',  56.84000, 60.60000),"
-                "(900013, 'cian',  'EXT-T-L2', 'улица Тестовая1772, 5',  56.88500, 60.60000),"
+                "(900013, 'cian',  'EXT-T-L2','u', 'улица Тестовая1772, 5',  56.88500, 60.60000),"
                 # C — different корпус → different canon, ~10 m apart → NOT merged
-                "(900014, 'avito', 'EXT-T-M2', 'Тестовая1772, 34к2', 56.84000, 60.60000),"
+                "(900014, 'avito', 'EXT-T-M2','u', 'Тестовая1772, 34к2', 56.84000, 60.60000),"
-                "(900015, 'cian',  'EXT-T-M4', 'Тестовая1772, 34к4', 56.84009, 60.60000)"
+                "(900015, 'cian',  'EXT-T-M4','u', 'Тестовая1772, 34к4', 56.84009, 60.60000)"
             )
         )
         db.execute(
@@ -840,7 +914,7 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None:
         db.execute(
             _t(
                 "DELETE FROM house_sources WHERE ext_id IN "
-                "('EXT-T-VK','EXT-T-VL','EXT-T-L1','EXT-T-L2','EXT-T-M2','EXT-T-M4')"
+                "('EXT-T-VK','u','EXT-T-VL','u','EXT-T-L1','u','EXT-T-L2','u','EXT-T-M2','u','EXT-T-M4')"
             )
         )
         db.execute(
@@ -850,6 +924,9 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None:
                 "'тестовая1772, 34к2','тестовая1772, 34к4')"
             )
         )
+        # journal rows have no FK and are never cascaded away — sweep them explicitly,
+        # or a re-run accumulates them (all live fixtures live in the 9000xx id range).
+        db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299"))
         db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900010 AND 900015"))
         db.commit()
         db.close()
@@ -877,21 +954,21 @@ def test_real_fias_pass_cross_guard_and_identity_carryover() -> None:
         db.execute(
             _t(
                 "INSERT INTO houses "
-                "(id, source, ext_house_id, address, lat, lon, house_fias_id, gar_house_guid, "
+                "(id, source, ext_house_id, url, 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,"
+                "(900020,'avito','EXT-F-K','u','ФиасОдин1772, 10',   56.84000,60.60000,"
                 " 'F-SAME-1772',NULL,NULL),"
-                "(900021,'cian', 'EXT-F-L','СовсемДругая1772, 77',56.84009,60.60000,"
+                "(900021,'cian', 'EXT-F-L','u','СовсемДругая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,"
+                "(900022,'avito','EXT-B-1','u','Клара1772, 32',56.84000,60.60000,"
                 " 'F-B1-1772',NULL,NULL),"
-                "(900023,'cian', 'EXT-B-2','Клара1772, 3/2',56.84009,60.60000,"
+                "(900023,'cian', 'EXT-B-2','u','Клара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),"
+                "(900024,'avito','EXT-C-K','u','Донбасс1772, 8',56.84000,60.60000,NULL,NULL,NULL),"
-                "(900025,'cian', 'EXT-C-L','Донбасс1772, 8',56.84009,60.60000,"
+                "(900025,'cian', 'EXT-C-L','u','Донбасс1772, 8',56.84009,60.60000,"
                 " 'F-CARRY-1772','G-CARRY-1772',NOW())"
             )
         )
@@ -944,7 +1021,7 @@ def test_real_fias_pass_cross_guard_and_identity_carryover() -> None:
         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')"
+                "('EXT-F-K','u','EXT-F-L','u','EXT-B-1','u','EXT-B-2','u','EXT-C-K','u','EXT-C-L')"
             )
         )
         db.execute(
@@ -954,6 +1031,9 @@ def test_real_fias_pass_cross_guard_and_identity_carryover() -> None:
                 "'донбасс1772, 8')"
             )
         )
+        # journal rows have no FK and are never cascaded away — sweep them explicitly,
+        # or a re-run accumulates them (all live fixtures live in the 9000xx id range).
+        db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299"))
         db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900020 AND 900025"))
         db.commit()
         db.close()
@@ -982,16 +1062,16 @@ def test_real_fias_pass_ignores_geo_guard() -> None:
         db.execute(
             _t(
                 "INSERT INTO houses "
-                "(id, source, ext_house_id, address, lat, lon, house_fias_id) VALUES "
+                "(id, source, ext_house_id, url, address, lat, lon, house_fias_id) VALUES "
                 # A — same fias, loser NULL geom → fias pass merges despite the missing coordinate
-                "(900030,'avito','EXT-2187-A-K','ФиасГеоA2187, 1', 56.84000,60.60000,'F-A-2187'),"
+                "(900030,'avito','EXT-2187-A-K','u','ФиасГеоA2187, 1',56.84,60.6,'F-A-2187'),"
-                "(900031,'cian', 'EXT-2187-A-L','ФиасГеоAL2187, 2',NULL,    NULL,    'F-A-2187'),"
+                "(900031,'cian', 'EXT-2187-A-L','u','ФиасГеоAL2187, 2',NULL,NULL,'F-A-2187'),"
                 # B — same fias, ~5 km apart (>250 m) → fias pass merges despite the distance
-                "(900032,'avito','EXT-2187-B-K','ФиасГеоB2187, 3', 56.84000,60.60000,'F-B-2187'),"
+                "(900032,'avito','EXT-2187-B-K','u','ФиасГеоB2187, 3',56.84,60.6,'F-B-2187'),"
-                "(900033,'cian', 'EXT-2187-B-L','ФиасГеоBL2187, 4',56.88500,60.60000,'F-B-2187'),"
+                "(900033,'cian', 'EXT-2187-B-L','u','ФиасГеоBL2187, 4',56.885,60.6,'F-B-2187'),"
                 # C — same canon, NO fias, ~5 km apart → canon pass STILL blocks (guard unchanged)
-                "(900034,'avito','EXT-2187-C-1','КанонГео2187, 5', 56.84000,60.60000,NULL),"
+                "(900034,'avito','EXT-2187-C-1','u','КанонГео2187, 5', 56.84000,60.60000,NULL),"
-                "(900035,'cian', 'EXT-2187-C-2','КанонГео2187, 5', 56.88500,60.60000,NULL)"
+                "(900035,'cian', 'EXT-2187-C-2','u','КанонГео2187, 5', 56.88500,60.60000,NULL)"
             )
         )
         # A loser gets a listing so we prove the re-point still fires with a NULL-geom loser.
@@ -1028,6 +1108,199 @@ def test_real_fias_pass_ignores_geo_guard() -> None:
         db.execute(_t("DELETE FROM listings WHERE id = 910031"))
         db.execute(_t("DELETE FROM house_sources WHERE house_id BETWEEN 900030 AND 900035"))
         db.execute(_t("DELETE FROM house_address_aliases WHERE house_id BETWEEN 900030 AND 900035"))
+        # journal rows have no FK and are never cascaded away — sweep them explicitly,
+        # or a re-run accumulates them (all live fixtures live in the 9000xx id range).
+        db.execute(_t("DELETE FROM house_merge_log WHERE loser_id BETWEEN 900000 AND 900299"))
         db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900030 AND 900035"))
         db.commit()
         db.close()
+
+
+# ── Merge journal: reversibility (#2690) ──────────────────────────────────────
+
+
+def test_undo_carryover_fields_match_merge_carryover() -> None:
+    """Static drift guard: migration 230's undo must un-set EXACTLY the fields the merge carries.
+
+    The undo NULLs the keeper's identity fields that the merge COALESCE-filled from a loser.
+    If _CARRY_OVER_IDENTITY_SQL ever gains a field and the migration's array does not, the undo
+    silently leaves that field on the keeper — the restored loser and the keeper would then both
+    claim the same ФИАС, and the next fias pass would merge them straight back.
+    """
+    migration = (_SQL_DIR / "230_house_merge_log.sql").read_text(encoding="utf-8")
+    carried = set(re.findall(r"^\s+(\w+)\s*=\s*COALESCE\(k\.", _CARRY_SQL, re.M))
+    # slice the ARRAY[...] literal itself — the declaration's own `text[]` also holds a «]»
+    block = migration[migration.index("c_carry_fields") :]
+    undone = set(re.findall(r"'(\w+)'", block[block.index("ARRAY[") : block.index("];")]))
+    assert carried, "could not parse carried fields out of _CARRY_OVER_IDENTITY_SQL"
+    assert carried == undone, f"carry-over/undo field drift: merge={carried} undo={undone}"
+
+
+def test_journal_written_in_the_same_transaction_as_the_merge() -> None:
+    """The journal INSERT must sit between mapping and delete, with no commit in between.
+
+    Requirement from #2690: a merge that commits without its journal row is exactly the failure
+    the journal exists to prevent, so the two must share one transaction.
+    """
+    src = inspect.getsource(hdm._run_merge_pass)
+    assert "_JOURNAL_INSERT_SQL" in src
+    assert "db.commit()" not in src, "the pass must not commit — the caller owns the txn"
+    # phase 1 (snapshots) strictly before the steps mutate anything, delete strictly after.
+    assert src.index("_JOURNAL_INSERT_SQL") < src.index("for label, sql in _STEPS")
+    assert src.index("for label, sql in _STEPS") < src.index("_DELETE_LOSERS_SQL")
+
+
+def test_every_step_returns_what_it_touched() -> None:
+    """Each step must RETURN its rows: ids for survivors, full snapshots for destroyed rows."""
+    for label, sql in hdm._STEPS:
+        assert "RETURNING" in sql, f"{label}: no RETURNING — its rows would go unjournaled"
+        if label.endswith("(collision-delete)") or label.endswith("(dedup)"):
+            assert "to_jsonb(" in sql, f"{label}: destroys rows, must snapshot them, not ids"
+        else:
+            assert "AS child_id" in sql, f"{label}: re-points rows, must return their ids"
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_merge_is_reversible_via_journal() -> None:
+    """End-to-end on a real DB: merge → journal is sufficient → undo restores the ORIGINAL state.
+
+    The comparison is over `to_jsonb(row.*)` for every row that existed before the merge — all
+    columns, not a chosen pair — for houses and for every FK child touched.
+    """
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    ids = "(900201, 900202)"
+    try:
+        # Keeper 900201 and loser 900202: same canon address, ~12 m apart (inside the 250 m
+        # guard), keeper has the listings so the keeper rule picks it.
+        db.execute(
+            _t(
+                "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon, geom, "
+                "year_built, house_fias_id, gar_flat_count, raw_payload) VALUES "
+                "(900201,'avito','K-2690','http://t/2690/k','улица Журнальная, 7', "
+                "  56.8400, 60.6000, ST_SetSRID(ST_MakePoint(60.6000,56.8400),4326), "
+                "  1979, NULL, NULL, '{\"k\":[1,2]}'), "
+                "(900202,'cian','L-2690','http://t/2690/l','ул. Журнальная,7', "
+                "  56.8401, 60.6000, ST_SetSRID(ST_MakePoint(60.6000,56.8401),4326), "
+                # NB: no «:word» inside the literal — SQLAlchemy text() would read it as a bind.
+                "  NULL, 'fias-2690-uuid', 144, '{\"l\":{\"deep\":[3,4]}}')"
+            )
+        )
+        db.execute(
+            _t(
+                "INSERT INTO listings (id, source, source_url, source_id, dedup_hash, price_rub, "
+                "house_id_fk) VALUES "
+                "(910201,'avito','http://t/2690/1','L1','dh-2690-1',5000000,900201),"
+                "(910202,'avito','http://t/2690/2','L2','dh-2690-2',5100000,900201),"
+                "(910203,'cian','http://t/2690/3','L3','dh-2690-3',6000000,900202)"
+            )
+        )
+        db.execute(
+            _t(
+                "INSERT INTO house_sources (house_id, ext_source, ext_id, confidence, "
+                "matched_method) VALUES (900201,'avito','S-2690-K',1.0,'t'),"
+                "(900202,'cian','S-2690-L',1.0,'t')"
+            )
+        )
+        # Colliding child: identical 6-col UNIQUE key on both → the loser's row is DESTROYED by
+        # the dedup step. Only a full snapshot can bring it back.
+        db.execute(
+            _t(
+                "INSERT INTO houses_price_dynamics (house_id, month_date, source, room_count, "
+                "prices_type, period, price_per_sqm) VALUES "
+                "(900201, DATE '2026-02-01','cian','all','priceSqm','allTime',100000),"
+                "(900202, DATE '2026-02-01','cian','all','priceSqm','allTime',999999)"
+            )
+        )
+        db.commit()
+
+        def snapshot() -> dict[tuple[str, int], Any]:
+            """to_jsonb of every seeded row, keyed by (table, id) — the full-fidelity state."""
+            out: dict[tuple[str, int], Any] = {}
+            for tbl, col in (
+                ("houses", "id"),
+                ("listings", "house_id_fk"),
+                ("house_sources", "house_id"),
+                ("houses_price_dynamics", "house_id"),
+            ):
+                where = f"id IN {ids}" if tbl == "houses" else f"{col} IN {ids}"
+                for r in db.execute(
+                    _t(f"SELECT id, to_jsonb(t.*) AS j FROM {tbl} t WHERE {where}")
+                ):
+                    out[(tbl, r.id)] = r.j
+            return out
+
+        before = snapshot()
+        assert len(before) == 9, f"fixture should seed 9 rows, got {sorted(before)}"
+
+        # ── merge ──
+        out = hdm.merge_duplicate_houses(db, dry_run=False, initiator="test")
+        assert out["losers_deleted"] == 1
+        assert db.execute(_t(f"SELECT count(*) FROM houses WHERE id IN {ids}")).scalar() == 1
+
+        # ── the journal alone must be able to NAME what went into what ──
+        row = db.execute(
+            _t("SELECT * FROM house_merge_log WHERE loser_id = 900202 ORDER BY id DESC LIMIT 1")
+        ).one()
+        assert (row.loser_id, row.keeper_id) == (900202, 900201)
+        assert row.merge_pass == "canon" and row.geo_guard is True
+        assert row.cluster_key.startswith("addr:")
+        assert 0 < row.distance_m < 250, "distance to the keeper must be recorded, in metres"
+        assert row.initiator == "test"
+        # full snapshot of the deleted row, not a reference to it
+        assert row.loser_row == before[("houses", 900202)]
+        # keeper as it stood BEFORE the identity carry-over (fias still empty there, filled now)
+        assert row.keeper_before["house_fias_id"] is None
+        assert (
+            db.execute(_t("SELECT house_fias_id FROM houses WHERE id = 900201")).scalar()
+            == "fias-2690-uuid"
+        ), "carry-over should have moved the loser's fias up"
+        # children: the loser's listing moved by id, the destroyed price row by content
+        assert row.children_repointed["listings.house_id_fk"] == [910203]
+        assert [r["price_per_sqm"] for r in row.children_deleted["houses_price_dynamics"]] == [
+            999999
+        ]
+
+        # ── undo ──
+        report = db.execute(
+            _t("SELECT * FROM house_merge_undo(CAST(:b AS uuid))"), {"b": str(row.batch_id)}
+        ).all()
+        assert len(report) == 1 and report[0].out_status.startswith("restored:"), report
+        db.commit()
+
+        after = snapshot()
+        # every row that existed before is back, byte-identical, on every column
+        assert {k: v for k, v in after.items() if k in before} == before
+        # the ONLY residue is the house_sources row the merge backfilled for the keeper.
+        # migration 230 documents this: it is built from the keeper's OWN ext_house_id, so
+        # it would have been true without the merge too. Asserted, not assumed.
+        residue = [v for k, v in after.items() if k not in before]
+        assert all(v["matched_method"] == "backfill_dedup_merge" for v in residue), residue
+    finally:
+        db.rollback()
+        db.execute(_t(f"DELETE FROM listings WHERE house_id_fk IN {ids}"))
+        db.execute(_t("DELETE FROM listings WHERE id IN (910201,910202,910203)"))
+        db.execute(_t("DELETE FROM house_merge_log WHERE loser_id = 900202"))
+        db.execute(_t("DELETE FROM house_address_aliases WHERE house_id IN (900201,900202)"))
+        db.execute(_t(f"DELETE FROM houses WHERE id IN {ids}"))
+        db.commit()
+        db.close()
+
+
+def test_merge_refuses_when_an_fk_child_is_unhandled() -> None:
+    """A new FK child on houses(id) must ABORT the merge, not be CASCADE-deleted unjournaled.
+
+    9 of the 11 FKs are ON DELETE CASCADE. A child the _STEPS pipeline does not know about is
+    therefore destroyed by `DELETE FROM houses` — no re-point step touches it, no RETURNING
+    records it, and the journal would claim a complete snapshot it does not have. Migration 133
+    already broke on prod over a missed child; there it failed loudly, here it would be silent.
+    """
+    db = _FakeDB(
+        mapping_rows=[_Row(2, 1, "ул. ленина, 5")],
+        fk_children={**_FK_CHILDREN, "house_brand_new_child": "house_id"},
+    )
+    with pytest.raises(RuntimeError, match="house_brand_new_child"):
+        hdm.merge_duplicate_houses(db, dry_run=False)  # type: ignore[arg-type]
+    assert db.commits == 0, "an unhandled child must abort before anything is committed"