fix(tradein/imv): «временная» ошибка снова временная — 1390 домов возвращаются в очередь

Очередь домовой IMV-оценки выбирает РОВНО ОДИН imv_status за прогон
(only_status, дефолт 'pending'). Повтор существовал только как значение этого
параметра — и за всю историю (41 прогон, 26.06–11.08) его не передал ни один
вызов. Дома, упавшие по инфраструктурной причине, которой больше нет (503/500
от браузерного сайдкара, #2698; за 7 суток до 12.08 в его access-логе 108 из
108 POST /fetch-json = 200), оставались в 'transient_error' навсегда: 1390 штук
на 12.08.2026, у всех есть и адрес, и параметры.

Что сделано:

1. Отдельная retry-очередь на половине пакета. Не OR к основной выборке: у
   pending last_imv_attempt_at всегда NULL, поэтому при общем
   ORDER BY ... NULLS FIRST повтор ждал бы опустошения pending (~5.7 тыс. домов
   ≈ год). Недобор retry-очереди возвращается pending — пакет не простаивает.
   Явный only_status отключает автоповтор: ручной запрос оператора обрабатывает
   ровно тот статус, который попросили, включая исчерпавшие лимит дома.

2. Выход из повтора — счётчик imv_transient_attempts (миграция 257), растёт
   только на transient_error, обнуляется успехом, порог 3. Число из замера:
   после починки сайдкара доля отказов на попытку 2/27 и 3/25 (прогоны
   3708/3467) ≈ 10%, то есть порог стоит максимум ~115 слотов ВСЕГО. Статус не
   переименовывается: «застряли окончательно» = transient_error AND
   imv_transient_attempts >= 3.

3. Половина пакета отдана повтору не за счёт pending: дома, по которым запрос к
   площадке невозможен, теперь помечаются no_params одним запросом до пакета
   (тот же вердикт и те же строки причин, что у поштучного пути). На проде это
   2276 домов из 6533 в очередях — 45 прогонов расписания (~136 суток) уходило
   на то, чтобы поштучно выяснить то, что видно в SQL. До правки до площадки
   доходили 17 домов из 50 (замер головы очереди 12.08), после — 50 из 50.

Пропускная способность (пакет 50 / 3 суток): pending 3218 + повтор 1039 после
премарка опустошаются за ~88 тактов (~264 суток) против 103 тактов (~309 суток)
на одном pending сейчас и «никогда» для 1390 застрявших.

Refs #2674

tradein-mvp/backend/app/api/v1/admin.py

@@ -2230,7 +2230,10 @@ class HouseIMVBackfillRequest(BaseModel):
     )
     only_status: str = Field(
         default="pending",
-        description="Обрабатывать дома с этим imv_status. 'transient_error' — retry.",
+        description=(
+            "Обрабатывать дома с этим imv_status. По умолчанию 'pending' + автоповтор "
+            "'transient_error' на половине пакета; явное значение = только этот статус."
+        ),
     )
     house_id: int | None = Field(
         default=None,
@@ -2267,7 +2270,12 @@ async def scrape_house_imv_backfill(
     batch_size: сколько домов обработать за запуск (default 50).
     request_delay_sec: пауза между IMV-вызовами (default 5s). ВАЖНО: Avito IMV
         реагирует на частые запросы с datacenter-IP. Не снижать < 3s.
-    only_status: по умолчанию 'pending'. Для retry failed — 'transient_error'.
+    only_status: по умолчанию 'pending' — и тогда половина пакета сама уходит на
+        повтор домов в 'transient_error' с непотраченным лимитом попыток (#2674:
+        раньше повтор существовал только как этот параметр, и за 41 прогон его
+        не передали ни разу — 1390 домов застряли навсегда). Явное значение
+        отключает автоповтор и обрабатывает РОВНО указанный статус, включая
+        дома, исчерпавшие лимит (imv_transient_attempts >= 3).
     house_id: обработать один дом (debug).
 
     Примечание по прокси: Avito IMV использует собственную curl_cffi-сессию.

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

@@ -411,7 +411,8 @@ def save_imv_result(db: Session, house_id: int, params: dict, result: IMVEvaluat
             UPDATE houses
             SET imv_status = 'ok',
                 last_imv_attempt_at = NOW(),
-                imv_error_reason = NULL
+                imv_error_reason = NULL,
+                imv_transient_attempts = 0
             WHERE id = :hid
         """),
         {"hid": house_id},
@@ -424,12 +425,19 @@ def _mark_status(
     status: str,
     reason: str | None = None,
 ) -> None:
+    # #2674: счётчик растёт ТОЛЬКО на transient_error — это «сколько раз подряд дом
+    # падал по временной причине», а не «сколько раз его трогали». no_params /
+    # no_address / not_found счётчик не двигают: они не занимают retry-слот.
     db.execute(
         text("""
             UPDATE houses
             SET imv_status = :s,
                 last_imv_attempt_at = NOW(),
-                imv_error_reason = :r
+                imv_error_reason = :r,
+                imv_transient_attempts = CASE
+                    WHEN :s = 'transient_error' THEN imv_transient_attempts + 1
+                    ELSE imv_transient_attempts
+                END
             WHERE id = :hid
         """),
         {"hid": house_id, "s": status, "r": reason},
@@ -443,6 +451,104 @@ _IMVStatus = Literal[
     "ok", "no_params", "no_address", "not_found", "auth_error", "transient", "error"
 ]
 
+# #2674: сколько раз подряд дом может упасть в transient_error, прежде чем
+# перестанет занимать retry-слот. Число из замера: после починки сайдкара (04.08)
+# доля отказов на попытку — 2/27 и 3/25 (прогоны 3708/3467), т.е. ~10%. На 1039
+# застрявших это ~104 повторных отказа на первом проходе, ~10 на втором, ~1 на
+# третьем. Порог 3 стоит максимум ~115 слотов ВСЕГО (≈2 прогона) и гарантирует,
+# что дом со СВОЕЙ (не инфраструктурной) причиной не крутится в пакете вечно.
+# Исчерпавшие лимит не исчезают из наблюдаемости: они остаются imv_status=
+# 'transient_error' и считаются как
+#   WHERE imv_status='transient_error' AND imv_transient_attempts >= 3.
+_MAX_TRANSIENT_ATTEMPTS = 3
+
+# Доля пакета под повтор transient_error. Половина — потому что остальные слоты
+# после #2674 достаются ТОЛЬКО домам, по которым реально будет запрос к площадке
+# (см. _premark_unusable): раньше из 50 слотов до площадки доходили 17 (замер
+# головы очереди на 12.08), так что pending на половине пакета всё равно идёт
+# быстрее, чем на целом до правки.
+_RETRY_SLOTS_SHARE = 0.5
+
+# Дом без пригодных параметров backfill всё равно пометит no_params — но только
+# заплатив слотом пакета и паузой request_delay_sec. Тот же вердикт берётся одним
+# запросом: нет ни одного объявления с rooms+area (pick_lot_params вернёт {}) ИЛИ
+# не из чего взять house_type (_map_house_type вернёт None → «unknown house_type»).
+# Причины пишем ТЕМИ ЖЕ строками, что и поштучный путь, — старые разрезы по
+# imv_error_reason продолжают работать.
+# Условие сознательно УЖЕ питоновского: normalize_house_type схлопывает в None ещё
+# и нераспознанный вокабуляр ('other', 'wireframe'), который тут остаётся текстом.
+# Промахнуться можно только в безопасную сторону — пометить меньше, чем пометил бы
+# поштучный путь.
+_PREMARK_UNUSABLE_SQL = text("""
+    WITH unusable AS (
+        SELECT h.id,
+               CASE WHEN NOT EXISTS (
+                        SELECT 1 FROM listings l
+                        WHERE l.house_id_fk = h.id
+                          AND l.rooms IS NOT NULL
+                          AND l.area_m2 IS NOT NULL)
+                    THEN 'no listings with rooms+area'
+                    ELSE 'unknown house_type'
+               END AS reason
+        FROM houses h
+        WHERE h.imv_status = ANY(CAST(:statuses AS text[]))
+          AND h.lat IS NOT NULL
+          AND h.lon IS NOT NULL
+          AND h.address IS NOT NULL
+          AND (
+            NOT EXISTS (
+                SELECT 1 FROM listings l
+                WHERE l.house_id_fk = h.id
+                  AND l.rooms IS NOT NULL
+                  AND l.area_m2 IS NOT NULL)
+            OR COALESCE(
+                NULLIF(TRIM((
+                    SELECT mode() WITHIN GROUP (ORDER BY l.house_type)
+                    FROM listings l
+                    WHERE l.house_id_fk = h.id
+                      AND l.rooms IS NOT NULL
+                      AND l.area_m2 IS NOT NULL)), ''),
+                NULLIF(TRIM(h.house_type), '')
+            ) IS NULL
+          )
+    )
+    UPDATE houses
+    SET imv_status = 'no_params',
+        last_imv_attempt_at = NOW(),
+        imv_error_reason = unusable.reason
+    FROM unusable
+    WHERE houses.id = unusable.id
+""")
+
+# Основная очередь: один статус, как и было (only_status — публичный параметр
+# admin-API, семантику не трогаем).
+_QUEUE_SQL = text("""
+    SELECT id, address, full_address, lat, lon
+    FROM houses
+    WHERE imv_status = :status
+      AND lat IS NOT NULL
+      AND lon IS NOT NULL
+      AND address IS NOT NULL
+    ORDER BY last_imv_attempt_at NULLS FIRST, id
+    LIMIT :batch
+""")
+
+# Retry-очередь (#2674). Отдельный запрос, а не OR к основной: у pending
+# last_imv_attempt_at всегда NULL, поэтому при общем ORDER BY ... NULLS FIRST
+# transient_error не попал бы в пакет, пока не кончится pending (по замеру
+# 12.08 — 5747 домов ≈ год). Отдельная квота = отдельный проход.
+_RETRY_QUEUE_SQL = text("""
+    SELECT id, address, full_address, lat, lon
+    FROM houses
+    WHERE imv_status = 'transient_error'
+      AND imv_transient_attempts < :max_attempts
+      AND lat IS NOT NULL
+      AND lon IS NOT NULL
+      AND address IS NOT NULL
+    ORDER BY last_imv_attempt_at NULLS FIRST, id
+    LIMIT :batch
+""")
+
 
 @dataclass
 class HouseIMVBackfillResult:
@@ -452,6 +558,23 @@ class HouseIMVBackfillResult:
     errors: int = 0
     duration_sec: float = field(default=0.0)
     status_counts: dict[str, int] = field(default_factory=dict)
+    # #2674: сколько домов пакета пришло из retry-очереди transient_error и
+    # сколько помечено no_params до пакета (без запроса к площадке).
+    retried: int = 0
+    premarked: int = 0
+
+
+def _premark_unusable(db: Session, statuses: list[str]) -> int:
+    """Пометить no_params дома, по которым запрос к площадке невозможен. → сколько.
+
+    Не новое поведение, а тот же вердикт _process_one_house одним запросом: на
+    12.08 в очереди 1925 таких домов из 5143 (113 без объявлений с rooms+area,
+    1812 без house_type) — каждый занимал слот пакета и паузу, чтобы получить
+    ответ, который виден в SQL.
+    """
+    res = db.execute(_PREMARK_UNUSABLE_SQL, {"statuses": statuses})
+    db.commit()
+    return int(res.rowcount or 0)
 
 
 def _beat(heartbeat: Callable[[], None] | None) -> None:
@@ -480,7 +603,11 @@ async def backfill_house_imv(
         batch_size: max houses to process (ignored when house_id given).
         request_delay_sec: sleep between Avito API calls (default 5s — anti-bot).
         only_status: process houses with this imv_status (default 'pending').
-            Use 'transient_error' to retry failures.
+            Use 'transient_error' to retry failures. При значении по умолчанию
+            часть пакета (_RETRY_SLOTS_SHARE) автоматически уходит на повтор
+            transient_error с непотраченным лимитом попыток (#2674) — явно
+            переданный only_status этот проход отключает, оператор получает
+            ровно то, что попросил, включая исчерпавшие лимит дома.
         house_id: process a single specific house (debug).
         heartbeat: optional callback дёргается каждые _HEARTBEAT_EVERY_N_HOUSES
             домов — caller обновляет scrape_runs.heartbeat_at, чтобы reap_zombies
@@ -509,23 +636,45 @@ async def backfill_house_imv(
             .all()
         )
     else:
-        rows = (
+        # Повторный проход только на расписании (only_status по умолчанию): явный
+        # only_status от оператора — это ручной запрос ровно одного статуса.
+        retry_lane = only_status == "pending"
+
+        statuses = [only_status] + (["transient_error"] if retry_lane else [])
+        result.premarked = _premark_unusable(db, statuses)
+        if result.premarked:
+            logger.info(
+                "house_imv_backfill: %d домов помечены no_params до пакета (нет rooms+area "
+                "или house_type) — слоты пакета не потрачены",
+                result.premarked,
+            )
+
+        retry_rows: list = []
+        if retry_lane:
+            retry_rows = (
                 db.execute(
-                text("""
+                    _RETRY_QUEUE_SQL,
-                    SELECT id, address, full_address, lat, lon
+                    {
-                    FROM houses
+                        "max_attempts": _MAX_TRANSIENT_ATTEMPTS,
-                    WHERE imv_status = :status
+                        "batch": int(batch_size * _RETRY_SLOTS_SHARE),
-                      AND lat IS NOT NULL
+                    },
-                      AND lon IS NOT NULL
-                      AND address IS NOT NULL
-                    ORDER BY last_imv_attempt_at NULLS FIRST, id
-                    LIMIT :batch
-                """),
-                {"status": only_status, "batch": batch_size},
                 )
                 .mappings()
                 .all()
             )
+        result.retried = len(retry_rows)
+
+        # Недобор retry-очереди (она кончится раньше pending: 1039 против 3218 на
+        # 12.08) возвращается pending — пакет не простаивает.
+        fresh_rows = (
+            db.execute(
+                _QUEUE_SQL,
+                {"status": only_status, "batch": max(batch_size - result.retried, 0)},
+            )
+            .mappings()
+            .all()
+        )
+        rows = list(fresh_rows) + list(retry_rows)
 
     result.checked = len(rows)
     if not rows:
@@ -534,9 +683,11 @@ async def backfill_house_imv(
         return result
 
     logger.info(
-        "house_imv_backfill: %d houses (status=%r delay=%.1fs)",
+        "house_imv_backfill: %d houses (status=%r retry=%d premarked=%d delay=%.1fs)",
         result.checked,
         only_status,
+        result.retried,
+        result.premarked,
         request_delay_sec,
     )
 
@@ -610,11 +761,14 @@ async def backfill_house_imv(
 
     result.duration_sec = time.time() - t0
     logger.info(
-        "house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
+        "house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d "
+        "retried=%d premarked=%d %.1fs %s",
         result.checked,
         result.saved,
         result.skipped,
         result.errors,
+        result.retried,
+        result.premarked,
         result.duration_sec,
         result.status_counts,
     )

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

@@ -447,6 +447,12 @@ async def _job_house_imv_backfill(
             # алерт — но пустая очередь при ежедневном расписании это и правда сигнал.
             "total_seen": result.checked,
             "new_count": result.saved,
+            # #2674: из скольких слотов пакета взяты дома на ПОВТОР (transient_error)
+            # и сколько домов ушло в no_params до пакета одним запросом. Без этих
+            # двух счётчиков в scrape_runs.counters проверить, что застрявшие
+            # действительно возвращаются в очередь, можно только по houses.
+            "retried": result.retried,
+            "premarked": result.premarked,
         }
         # Честный статус (#2674, тот же класс, что #2670/#2657): успех — это
         # «сделали то, что собирались», а не «не поймали известное исключение».

tradein-mvp/backend/data/sql/258_houses_imv_transient_attempts.sql

@@ -0,0 +1,41 @@
+-- 258_houses_imv_transient_attempts.sql
+-- Счётчик подряд идущих временных отказов домовой оценки Авито (эпик #2674).
+--
+-- ЗАЧЕM. imv_status='transient_error' был состоянием БЕЗ ВЫХОДА: очередь
+-- backfill'а выбирает ровно один статус за прогон (only_status, по умолчанию
+-- 'pending'), и за всю историю (41 прогон, 26.06–11.08) ни один не был запущен
+-- с другим значением. На 12.08.2026 в этом статусе лежали 1390 домов, 1337 из
+-- них — с причиной «503/500 от tradein-browser:3000/fetch-json» или «All
+-- connection attempts failed», то есть с ИНФРАСТРУКТУРНОЙ причиной, которой
+-- больше нет (сайдкар починен #2698; за 7 суток до 12.08 в его access-логе
+-- 108 из 108 POST /fetch-json = 200).
+--
+-- Сервис теперь отдаёт часть пакета на повтор transient_error автоматически
+-- (house_imv_backfill._RETRY_QUEUE_SQL). Этот счётчик — условие ВЫХОДА из
+-- повтора: дом, падающий по своей причине, а не по инфраструктурной, перестаёт
+-- занимать слот пакета после _MAX_TRANSIENT_ATTEMPTS (3) подряд.
+--
+-- Наблюдаемость НЕ переименовывается: статус остаётся 'transient_error',
+-- прежние разрезы по imv_status/imv_error_reason работают как работали, а
+-- «застряли окончательно» — это
+--   SELECT count(*) FROM houses
+--    WHERE imv_status='transient_error' AND imv_transient_attempts >= 3;
+--
+-- Индекс не добавляем: houses_imv_status_idx (064) уже частичный по
+-- imv_status IN ('pending','transient_error') с сортировкой по
+-- last_imv_attempt_at — фильтр по счётчику остаётся остаточным условием на
+-- выборке в тысячи строк.
+
+BEGIN;
+
+SET LOCAL lock_timeout = '5s';
+
+ALTER TABLE houses
+  ADD COLUMN IF NOT EXISTS imv_transient_attempts smallint NOT NULL DEFAULT 0;
+
+COMMENT ON COLUMN houses.imv_transient_attempts IS
+  'Сколько раз подряд домовая IMV-оценка падала в transient_error. '
+  'Растёт только на transient_error, обнуляется успехом. '
+  '>= 3 — дом больше не берётся в автоматический повтор (эпик #2674).';
+
+COMMIT;

tradein-mvp/backend/data/sql/_manifest_applied.txt

@@ -246,3 +246,4 @@
 251_listings_drop_ceiling_height.sql
 254_listings_backfill_avito_rating_glued_address.sql
 257_listings_backfill_yandex_source_url.sql
+258_houses_imv_transient_attempts.sql

tradein-mvp/backend/tests/test_backfill_wave2.py

@@ -500,7 +500,11 @@ async def test_backfill_house_imv_ok_path():
         patch("app.services.house_imv_backfill.save_imv_result") as mock_save,
     ):
         mock_mappings = MagicMock()
-        mock_mappings.all.return_value = houses
+        # #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
+        # затем основную по only_status. Фейк отвечает одним и тем же списком на любой
+        # SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
+        # придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
+        mock_mappings.all.side_effect = [[], houses]
         mock_db.execute.return_value.mappings.return_value = mock_mappings
 
         result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
@@ -539,7 +543,11 @@ async def test_backfill_house_imv_no_params():
         patch("app.services.house_imv_backfill._mark_status") as mock_mark,
     ):
         mock_mappings = MagicMock()
-        mock_mappings.all.return_value = houses
+        # #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
+        # затем основную по only_status. Фейк отвечает одним и тем же списком на любой
+        # SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
+        # придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
+        mock_mappings.all.side_effect = [[], houses]
         mock_db.execute.return_value.mappings.return_value = mock_mappings
 
         result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
@@ -593,7 +601,11 @@ async def test_backfill_house_imv_not_found():
         patch("app.services.house_imv_backfill._mark_status") as mock_mark,
     ):
         mock_mappings = MagicMock()
-        mock_mappings.all.return_value = houses
+        # #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
+        # затем основную по only_status. Фейк отвечает одним и тем же списком на любой
+        # SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
+        # придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
+        mock_mappings.all.side_effect = [[], houses]
         mock_db.execute.return_value.mappings.return_value = mock_mappings
 
         result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)

tradein-mvp/backend/tests/test_house_imv_params_honesty.py

@@ -138,7 +138,11 @@ async def test_unknown_house_type_skips_request_and_marks_house() -> None:
         patch.object(hib, "evaluate_via_imv", new_callable=AsyncMock) as mock_eval,
         patch.object(hib, "_mark_status") as mock_mark,
     ):
-        db.execute.return_value.mappings.return_value.all.return_value = houses
+        # #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
+        # затем основную по only_status. Фейк отвечает одним и тем же списком на любой
+        # SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
+        # придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
+        db.execute.return_value.mappings.return_value.all.side_effect = [[], houses]
         result = await hib.backfill_house_imv(db, batch_size=10, request_delay_sec=0.0)
 
     mock_eval.assert_not_called()

tradein-mvp/backend/tests/test_house_imv_retry_stuck.py

@@ -0,0 +1,267 @@
+"""#2674: «временная» ошибка домовой IMV-оценки снова означает временную.
+
+Что было. Очередь backfill'а выбирает РОВНО ОДИН imv_status за прогон
+(`only_status`, дефолт 'pending'). Возможность повтора существовала и была
+задокументирована в admin-API, но за всю историю (41 прогон, 26.06–11.08.2026)
+её не передал никто: все прогоны ушли с 'pending'. Дома, упавшие по причине,
+которой больше нет (503/500 от браузерного сайдкара, #2698), лежали в
+'transient_error' без единого шанса вернуться — 1390 штук на 12.08.2026.
+
+Что проверяем — по существу, а не по настройке:
+  * дом в 'transient_error' ПОПАДАЕТ в пакет прогона с дефолтными параметрами
+    (тот же вызов, что делает расписание) — на старом коде он не попадает;
+  * pending при этом из пакета НЕ вытесняется;
+  * дом, исчерпавший лимит попыток, в автоматический повтор не берётся, а
+    ЯВНЫЙ only_status='transient_error' (ручной запрос оператора) берёт и его;
+  * дом без пригодных параметров помечается no_params БЕЗ запроса к площадке.
+
+Живой Postgres обязателен: проверяется поведение SQL-выборки, а не то, какие
+строки вернул мок. Без БД тест самоскипается (в CI БД есть — ci-tradein.yml
+поднимает Postgres и накатывает схему).
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import house_imv_backfill as hib
+
+# Фикстуры живут в собственном диапазоне id (ср. tests/test_house_dedup_merge.py).
+_H_STUCK = 900601  # transient_error, попыток 0 → обязан вернуться в очередь
+_H_EXHAUSTED = 900602  # transient_error, попытки исчерпаны → в автоповтор не берём
+_H_PENDING = 900603  # pending → не должен быть вытеснен retry-очередью
+_H_NO_PARAMS = 900604  # pending без объявлений с rooms+area → no_params без запроса
+_IDS = (_H_STUCK, _H_EXHAUSTED, _H_PENDING, _H_NO_PARAMS)
+
+
+def _live_session() -> Any | None:
+    """Session к достижимому не-заглушечному Postgres, иначе None."""
+    try:
+        from sqlalchemy import create_engine, text
+        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()
+        conn.execute(text("SELECT 1"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+def _seed(db: Any) -> None:
+    from sqlalchemy import text as _t
+
+    # url/source/ext_house_id — NOT NULL в houses (002_core_tables).
+    db.execute(
+        _t(
+            "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon, "
+            "                    imv_status, imv_transient_attempts, last_imv_attempt_at) VALUES "
+            "(:s,  'avito', 'EXT-2674-S', 'u', 'тестдом 2674, 1', 56.84, 60.60, "
+            " 'transient_error', 0, NOW() - INTERVAL '10 days'),"
+            "(:x,  'avito', 'EXT-2674-X', 'u', 'тестдом 2674, 2', 56.84, 60.60, "
+            " 'transient_error', :cap, NOW() - INTERVAL '10 days'),"
+            "(:p,  'avito', 'EXT-2674-P', 'u', 'тестдом 2674, 3', 56.84, 60.60, "
+            " 'pending', 0, NULL),"
+            "(:n,  'avito', 'EXT-2674-N', 'u', 'тестдом 2674, 4', 56.84, 60.60, "
+            " 'pending', 0, NULL)"
+        ),
+        {
+            "s": _H_STUCK,
+            "x": _H_EXHAUSTED,
+            "p": _H_PENDING,
+            "n": _H_NO_PARAMS,
+            # Литерал, а не hib._MAX_TRANSIENT_ATTEMPTS: подготовка теста обязана
+            # работать и на СТАРОМ коде, иначе красный прогон получится про
+            # отсутствующий атрибут, а не про застрявший дом. Синхронность
+            # литерала и константы стережёт test_transient_attempts_counter_*.
+            "cap": 3,
+        },
+    )
+    # Объявления с rooms+area+house_type — у всех, КРОМЕ _H_NO_PARAMS.
+    # source_url / dedup_hash / price_rub — NOT NULL, dedup_hash UNIQUE.
+    db.execute(
+        _t(
+            "INSERT INTO listings (id, source, source_url, source_id, dedup_hash, price_rub, "
+            "                      house_id_fk, rooms, area_m2, floor, total_floors, house_type) "
+            "VALUES "
+            "(:l1, 'avito', 'http://t/2674/1', 'L-2674-1', 'dh-2674-1', 5000000, :s, "
+            " 2, 52.0, 3, 9, 'brick'),"
+            "(:l2, 'avito', 'http://t/2674/2', 'L-2674-2', 'dh-2674-2', 5000000, :x, "
+            " 2, 52.0, 3, 9, 'brick'),"
+            "(:l3, 'avito', 'http://t/2674/3', 'L-2674-3', 'dh-2674-3', 5000000, :p, "
+            " 2, 52.0, 3, 9, 'brick')"
+        ),
+        {
+            "l1": 910601,
+            "l2": 910602,
+            "l3": 910603,
+            "s": _H_STUCK,
+            "x": _H_EXHAUSTED,
+            "p": _H_PENDING,
+        },
+    )
+    db.commit()
+
+
+def _cleanup(db: Any) -> None:
+    from sqlalchemy import text as _t
+
+    db.rollback()
+    db.execute(_t("DELETE FROM listings WHERE id IN (910601, 910602, 910603)"))
+    db.execute(_t("DELETE FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"), {"ids": list(_IDS)})
+    db.commit()
+    db.close()
+
+
+async def _run(db: Any, **kwargs: Any) -> Any:
+    """Прогнать backfill с замоканным вызовом площадки.
+
+    Мокаем ровно сетевую границу: evaluate_via_imv поднимает «адрес не найден»,
+    поэтому дом уходит в not_found и путь сохранения (три таблицы) не нужен —
+    проверяем именно ВЫБОРКУ. Флаг браузерного фетчера гасим, чтобы не поднимать
+    сайдкар.
+    """
+    err = hib.IMVAddressNotFoundError("address not found (test)")
+    with (
+        patch.object(hib.settings, "avito_imv_use_browser_fetcher", False),
+        patch.object(hib, "evaluate_via_imv", AsyncMock(side_effect=err)),
+    ):
+        return await hib.backfill_house_imv(db, request_delay_sec=0.0, **kwargs)
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+async def test_stuck_transient_house_returns_to_the_queue_by_itself() -> None:
+    """Прогон с дефолтными параметрами (как у расписания) забирает застрявший дом.
+
+    На старом коде выборка — `WHERE imv_status = :status` с единственным
+    значением 'pending', поэтому дом остаётся в 'transient_error' навсегда.
+    """
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    if db.execute(_t("SELECT count(*) FROM houses")).scalar():
+        pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
+    try:
+        _seed(db)
+
+        # Ровно то, что вызывает расписание: batch_size из default_params, статус по умолчанию.
+        result = await _run(db, batch_size=50)
+
+        picked = {
+            r.id: r.imv_status
+            for r in db.execute(
+                _t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
+                {"ids": list(_IDS)},
+            )
+        }
+        assert picked[_H_STUCK] != "transient_error", (
+            "дом с временной ошибкой и непотраченным лимитом обязан попасть в пакет "
+            f"обычного прогона; фактически статус {picked[_H_STUCK]!r} "
+            f"(в пакет прогона попало домов: {result.checked})"
+        )
+        # pending не вытеснен повтором.
+        assert picked[_H_PENDING] != "pending", "pending-дом обязан остаться в том же пакете"
+        # Исчерпавший лимит в автоповтор не берётся — иначе вечный цикл.
+        assert picked[_H_EXHAUSTED] == "transient_error"
+        assert result.retried == 1
+        # Дом без объявлений с rooms+area помечен ДО пакета, слот не потрачен.
+        assert picked[_H_NO_PARAMS] == "no_params"
+        assert result.premarked == 1
+    finally:
+        _cleanup(db)
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+async def test_explicit_only_status_still_takes_exhausted_houses() -> None:
+    """Явный only_status — ручной запрос оператора: лимит попыток его не режет."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    if db.execute(_t("SELECT count(*) FROM houses")).scalar():
+        pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
+    try:
+        _seed(db)
+
+        result = await _run(db, batch_size=50, only_status="transient_error")
+
+        statuses = {
+            r.id: r.imv_status
+            for r in db.execute(
+                _t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
+                {"ids": list(_IDS)},
+            )
+        }
+        assert (
+            statuses[_H_EXHAUSTED] != "transient_error"
+        ), "явно запрошенный статус обрабатывается целиком, включая исчерпавшие лимит"
+        # Автоповтора поверх явного запроса нет: pending не тронут.
+        assert statuses[_H_PENDING] == "pending"
+        assert result.retried == 0
+    finally:
+        _cleanup(db)
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+async def test_transient_attempts_counter_only_counts_transient() -> None:
+    """Счётчик — это «сколько раз подряд упал по временной причине», и он обнуляем.
+
+    Без верхней границы у повтора не было бы выхода; без обнуления успехом дом,
+    однажды поймавший 3 сетевых сбоя, навсегда терял бы право на повтор.
+    """
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    if db.execute(_t("SELECT count(*) FROM houses")).scalar():
+        pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
+    try:
+        _seed(db)
+        assert hib._MAX_TRANSIENT_ATTEMPTS == 3, "литерал 'cap' в _seed рассинхронизирован"
+
+        def _attempts() -> int:
+            return int(
+                db.execute(
+                    _t("SELECT imv_transient_attempts FROM houses WHERE id = :i"),
+                    {"i": _H_STUCK},
+                ).scalar()
+            )
+
+        hib._mark_status(db, _H_STUCK, "transient_error", "boom")
+        assert _attempts() == 1
+        hib._mark_status(db, _H_STUCK, "not_found", "адрес не найден")
+        assert _attempts() == 1, "нетранзиентные исходы счётчик не двигают"
+        db.execute(
+            _t("UPDATE houses SET imv_status='ok', imv_transient_attempts=0 WHERE id = :i"),
+            {"i": _H_STUCK},
+        )
+        db.commit()
+        assert _attempts() == 0
+    finally:
+        _cleanup(db)
+
+
+def test_retry_queue_is_not_bound_to_only_status() -> None:
+    """Статика (бежит и без БД): retry-выборка не зависит от параметра only_status.
+
+    Ровно этот дефект и застрял на проде: единственный путь к 'transient_error'
+    шёл через значение, которое никто не передавал.
+    """
+    retry_sql = str(hib._RETRY_QUEUE_SQL.text)
+    assert "imv_status = 'transient_error'" in retry_sql
+    assert ":status" not in retry_sql, "retry-очередь обязана быть независимой от only_status"
+    assert "imv_transient_attempts < :max_attempts" in retry_sql, "у повтора должен быть выход"
+    # psycopg v3: только CAST(:x AS type), никогда :x::type.
+    for sql in (retry_sql, str(hib._QUEUE_SQL.text), str(hib._PREMARK_UNUSABLE_SQL.text)):
+        assert "::" not in sql