fix(tradein/deactivate): TTL не снимает объявления по порогу ниже собственного цикла обхода

Гейт #2710 отвечает «источник собирается?». Он не отвечает на второй вопрос
#2659 — «а достаточно ли ttl_days, чтобы молчание означало снятие?». Пока свип
возвращается к строке реже, чем раз в ttl_days, TTL меряет нашу выборку, а не
жизнь объявления, — и источник при этом полностью здоров, так что гейт молчит.

Замер на проде 2026-08-09 (read-only): с деплоя гейта 06.08 TTL снял 1 028 строк,
127 из них (12.4%) уже снова активны — свип нашёл их живыми через 1-3 суток и
вернул сам. В Екатеринбурге, единственном городе с настоящим покрытием, доля
ложных снятий 100% (cian 103/103, yandex 24/24); возраст на момент снятия у всех
127 — 29.9..30.3 суток при TTL=30, то есть срабатывание ровно на границе.

Корень не в конкретном числе, а в том, что число подобрано руками ниже хвоста
обхода. Разрывы переобхода против текущих TTL (listing_source_snapshots, 40 сут):
domklik 3.1 при TTL=14 (4.5x запас, сплошное суточное покрытие — контрольная
группа), cian 26.6 при 30, yandex 43.0 при 30, avito 42.1 при 10. Поэтому второе
подобранное руками число проблему не решает.

Вместо константы меряем факт: какой самый большой возраст, при котором свип за
последнее окно ДОКАЗАЛ строку живой. Эффективный TTL = max(ttl_days, этот пол),
по тому же срезу source+segments и по той же колонке свежести, что и UPDATE.
Пол только поднимает порог. Побочный эффект намеренный: после провала сбора хвост
разрывов распухает, пол растёт, деактивация замирает сама — то, чего issue просил
от «гейта по банам», но выраженное через результат, а не через причину.

Тот же запрос на проде даёт cian/vtorichka 34.0, yandex/vtorichka 74.3,
avito 69.7, domklik NULL (сбор стоит, гейт его и так пропускает) — 235 мс,
раз в сутки. Квантиль 0.99 — калибровочная ручка в default_params расписания,
подобран по требованию «пол обязан накрыть возраст 30.3 доказанно ложных снятий».

Видимое пользователю: ближайший прогон Яндекса перестаёт снимать 43 активные
строки vtorichka; на проде это единственные строки, которых TTL сейчас касается.

Refs #2659

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

@@ -218,6 +218,7 @@ async def _job_deactivate_stale(
     from app.core.config import settings as _settings
     from app.tasks.deactivate_stale_avito import (
         DEFAULT_MIN_CONFIRMATIONS,
+        DEFAULT_REVISIT_FLOOR_QUANTILE,
         deactivate_stale_listings,
     )
 
@@ -229,6 +230,12 @@ async def _job_deactivate_stale(
     # получает страховочный порог, а не «деактивируй вслепую». Посчитанные по
     # источнику пороги приходят из default_params (миграция 219).
     min_confirmations: int = params.get("min_confirmations", DEFAULT_MIN_CONFIRMATIONS)
+    # Пол TTL по измеренному циклу переобхода (#2659) — тоже включён по умолчанию:
+    # незасеянное расписание не должно снимать объявления по порогу ниже собственного
+    # хвоста обхода. Снять ручку вручную: revisit_floor_quantile = 0.
+    revisit_floor_quantile: float = params.get(
+        "revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE
+    )
 
     loop = asyncio.get_event_loop()
     await loop.run_in_executor(
@@ -241,6 +248,7 @@ async def _job_deactivate_stale(
             segments=segments,
             staleness_column=staleness_column,
             min_confirmations=min_confirmations,
+            revisit_floor_quantile=revisit_floor_quantile,
         ),
     )
 

tradein-mvp/backend/app/tasks/deactivate_stale_avito.py

@@ -24,6 +24,7 @@ TTL для avito берётся из settings.avito_stale_ttl_days (env AVITO_ST
 from __future__ import annotations
 
 import logging
+from math import ceil
 from typing import Any
 
 from sqlalchemy import text
@@ -127,6 +128,107 @@ DEFAULT_MIN_CONFIRMATIONS = 500
 _CONFIRMATIONS_SEGMENT_FILTER = "\n          AND listing_segment = ANY(CAST(:segments AS text[]))"
 
 
+# ── Пол TTL по измеренному циклу переобхода (#2659) ───────────────────────────
+# Гейт выше отвечает на вопрос «источник вообще собирается?». Он НЕ отвечает на
+# вопрос, из-за которого заведён #2659: «а достаточно ли ttl_days, чтобы молчание
+# означало снятие?». Пока свип возвращается к строке реже, чем раз в ttl_days,
+# TTL меряет НАШУ выборку, а не жизнь объявления, — и источник при этом полностью
+# здоров, так что гейт молчит.
+#
+# ЗАМЕР НА ПРОДЕ 2026-08-09, из-за которого этот пол существует.
+# С момента деплоя гейта (06.08) TTL снял 1 028 строк; 127 из них (12.4%) УЖЕ снова
+# активны — свип нашёл их живыми через 1-3 суток и вернул сам (upsert в
+# scraper_kit/base.py ставит is_active = true). В единственном городе с настоящим
+# покрытием доля ложных снятий 100%:
+#   cian   Екатеринбург  103 снято → 103 снова активны
+#   yandex Екатеринбург   24 снято →  24 снова активны
+#   cian/yandex без города 901 снято → 0 вернулись (их свип не обходит вовсе)
+# Возраст на момент снятия у всех 127: 29.9..30.3 суток при TTL=30 — то есть TTL
+# срабатывал ровно на границе, а свип возвращался к строке на 31-34-е сутки.
+#
+# ПОЧЕМУ ЭТО НЕ ЛЕЧИТСЯ НОВОЙ КОНСТАНТОЙ. Разрывы переобхода, суток
+# (listing_source_snapshots, 40 суток, посчитано по срезу TTL-джобы):
+#   источник/сегмент      p90    p99   TTL сейчас   TTL/p99
+#   domklik  vtorichka     1.9    3.1      14         4.5   ← сплошное суточное покрытие
+#   cian     vtorichka    10.9   26.6      30         1.1
+#   yandex   vtorichka     5.7   43.0      30         0.7
+#   avito    vtorichka    29.1   42.1      10         0.24  ← отсюда 9 033 строки
+# Домклик — контрольная группа: при почти полном суточном обходе TTL=14 лежит в
+# 4.5 раза выше хвоста, и снятие у него действительно означает снятие. У остальных
+# трёх порог ниже собственного хвоста обхода — руками подобранное число и есть
+# корень #2659, поэтому чинить его вторым руками подобранным числом бессмысленно.
+#
+# ЧТО МЕРЯЕМ ВМЕСТО КОНСТАНТЫ: факт, а не оценку. «Какой самый большой возраст, при
+# котором свип за последнее окно ДОКАЗАЛ, что объявление живо» — то есть насколько
+# старую строку он только что нашёл на площадке. Если свип буквально вчера вернул к
+# жизни строку, молчавшую 40 суток, то 30 суток молчания не доказывают ничего.
+# Пол = квантиль этого распределения, эффективный TTL = max(ttl_days, пол).
+#
+# Считается по ТОМУ ЖЕ срезу (source + segments) и по ТОЙ ЖЕ колонке свежести, что
+# и UPDATE. Предыдущее наблюдение берётся из listing_source_snapshots — единственной
+# истории свежести, что у нас есть; расхождение listings.<col> и
+# listing_sources.last_seen_at замерено на проде и не превышает 0.5 суток в среднем
+# (максимум 0), что на шкале 30-70 суток шум.
+#
+# КВАНТИЛЬ — калибровочная ручка, не догма. 0.99 подобран по требованию «пол обязан
+# накрыть 127 доказанных ложных снятий», у которых возраст был 29.9..30.3: замер
+# того же запроса на проде даёт 34.0 для cian/vtorichka и 74.3 для yandex/vtorichka.
+# Ниже 0.99 опускать нельзя без нового замера. Ручка живёт в default_params
+# расписания (revisit_floor_quantile), 0 -> пол выключен.
+#
+# ПОБОЧНЫЙ ЭФФЕКТ, КОТОРЫЙ ЗДЕСЬ НАМЕРЕННЫЙ: после провала сбора хвост разрывов
+# распухает (свип разгребает завал и находит очень старые строки), пол поднимается,
+# и деактивация замирает сама — без отдельного детектора банов. Когда завал разобран,
+# хвост схлопывается и пол опускается обратно. Это ровно то поведение, которого
+# issue просил от «гейта по банам», но выраженное через результат, а не через причину.
+#
+# ПОТОЛОК: пол не может превысить глубину истории снимков. Если снимок за нужную
+# дату не писался (дыры на проде есть — 30.07, 01.08), берётся ближайший более
+# ранний; при полном отсутствии снимков пол не считается и TTL остаётся как задан.
+DEFAULT_REVISIT_FLOOR_QUANTILE = 0.99
+
+_REVISIT_FLOOR_SEGMENT_FILTER = "\n          AND l.listing_segment = ANY(CAST(:segments AS text[]))"
+
+
+def _build_revisit_floor_sql(staleness_column: str, *, with_segments: bool) -> Any:
+    """Квантиль возраста, при котором свип за окно ДОКАЗАЛ, что строка жива.
+
+    Пара «предыдущее наблюдение (снимок) → текущее наблюдение (listings)» даёт
+    разрыв переобхода в сутках; берём его квантиль по срезу source+segments.
+    Только строки, у которых свежесть реально сдвинулась, — то есть выжившие,
+    а не «мы к ним не приходили».
+
+    staleness_column уже прошёл whitelist-проверку в deactivate_stale_listings.
+    Значения — param-binding, psycopg v3 safe (CAST(... AS ...), никаких :param::type).
+    """
+    segment_filter = _REVISIT_FLOOR_SEGMENT_FILTER if with_segments else ""
+    return text(
+        f"""
+        SELECT percentile_disc(CAST(:revisit_quantile AS double precision))
+               WITHIN GROUP (
+                   ORDER BY EXTRACT(epoch FROM (l.{staleness_column} - prev.last_seen_at))
+                            / 86400.0
+               )
+        FROM listings l
+        JOIN listing_sources ls
+          ON ls.listing_id = l.id
+         AND ls.ext_source = l.source
+        JOIN listing_source_snapshots prev
+          ON prev.listing_source_id = ls.id
+         AND prev.snapshot_date = (
+                 SELECT max(snapshot_date)
+                 FROM listing_source_snapshots
+                 WHERE snapshot_date
+                       <= CURRENT_DATE - CAST(:health_window_days AS integer)
+             )
+        WHERE l.source = :listing_source
+          AND l.{staleness_column}
+              > NOW() - CAST(:health_window_days || ' days' AS interval)
+          AND l.{staleness_column} > prev.last_seen_at{segment_filter}
+        """
+    )
+
+
 def _build_confirmations_sql(staleness_column: str, *, with_segments: bool) -> Any:
     """SELECT count(*) подтверждённых за окно строк — тот же срез, что и у UPDATE.
 
@@ -209,6 +311,7 @@ def deactivate_stale_listings(
     staleness_column: str = "last_seen_at",
     min_confirmations: int = 0,
     health_window_days: int = _HEALTH_WINDOW_DAYS,
+    revisit_floor_quantile: float = 0.0,
 ) -> dict[str, int]:
     """Пометить is_active=false объявления, чья свежесть старше ttl_days дней.
 
@@ -229,6 +332,11 @@ def deactivate_stale_listings(
             вызывают старые тесты и совместимая обёртка); реальные значения приходят
             из default_params расписания, см. миграцию 219 и комментарий выше.
         health_window_days: окно подтверждений для гейта, суток. Дефолт 3.
+        revisit_floor_quantile: пол TTL по измеренному циклу переобхода (#2659).
+            Квантиль возраста, при котором свип за окно ДОКАЗАЛ строку живой;
+            эффективный TTL = max(ttl_days, этот пол). 0 -> пол выключен (так
+            вызывают старые тесты и совместимая обёртка), рабочее значение —
+            DEFAULT_REVISIT_FLOOR_QUANTILE, см. комментарий выше.
 
     Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
     Один statement в транзакции: UPDATE флага + снимок 'stale' в listings_snapshots
@@ -236,7 +344,8 @@ def deactivate_stale_listings(
 
     Returns {"deactivated": N} -- количество обновлённых строк (1:1 со снимками).
     Если гейт не пропустил прогон: {"deactivated": 0, "confirmations": N,
-    "skipped_unhealthy": 1} и НИ ОДНА строка не тронута.
+    "skipped_unhealthy": 1} и НИ ОДНА строка не тронута. Если пол переобхода поднял
+    TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}.
 
     Raises:
         ValueError: если staleness_column не входит в whitelist (проверка ДО SQL,
@@ -292,6 +401,44 @@ def deactivate_stale_listings(
                 )
                 return counters
 
+        # Пол TTL по измеренному циклу переобхода (#2659) — тоже ДО UPDATE и по тому же
+        # срезу. Поднимает порог, никогда не опускает: max(), а не замена.
+        effective_ttl_days = ttl_days
+        if revisit_floor_quantile > 0:
+            floor_params: dict[str, Any] = {
+                "listing_source": listing_source,
+                "health_window_days": health_window_days,
+                "revisit_quantile": revisit_floor_quantile,
+            }
+            if segments is not None:
+                floor_params["segments"] = segments
+            floor_days = db.execute(
+                _build_revisit_floor_sql(staleness_column, with_segments=segments is not None),
+                floor_params,
+            ).scalar()
+            # NULL = истории снимков за окно нет вовсе (свежая БД, дыра в снимках).
+            # Тогда пола нет и TTL остаётся как задан: выдумывать пол не из чего.
+            if floor_days is not None:
+                counters["revisit_floor_days"] = ceil(float(floor_days))
+                effective_ttl_days = max(ttl_days, counters["revisit_floor_days"])
+                counters["ttl_days_effective"] = effective_ttl_days
+                if effective_ttl_days > ttl_days:
+                    logger.warning(
+                        "deactivate_stale source=%s run_id=%d TTL поднят с %d до %d сут: "
+                        "свип за %d сут доказал живой строку, молчавшую %d сут "
+                        "(квантиль %.3f, segments=%r) — при ttl_days=%d снятие означало бы "
+                        "«мы не дошли», а не «объявление снято»",
+                        listing_source,
+                        run_id,
+                        ttl_days,
+                        effective_ttl_days,
+                        health_window_days,
+                        counters["revisit_floor_days"],
+                        revisit_floor_quantile,
+                        segments,
+                        ttl_days,
+                    )
+
         # segments is None -> все сегменты (поведение avito). segments=[...] -> только
         # перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список []
         # означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0),
@@ -299,7 +446,7 @@ def deactivate_stale_listings(
         if segments is not None:
             params: dict[str, Any] = {
                 "listing_source": listing_source,
-                "ttl_days": ttl_days,
+                "ttl_days": effective_ttl_days,
                 "segments": segments,
                 "run_id": run_id,
             }
@@ -307,7 +454,7 @@ def deactivate_stale_listings(
         else:
             params = {
                 "listing_source": listing_source,
-                "ttl_days": ttl_days,
+                "ttl_days": effective_ttl_days,
                 "run_id": run_id,
             }
             result = db.execute(_build_all_segments_sql(staleness_column), params)
@@ -318,10 +465,11 @@ def deactivate_stale_listings(
         runs_mod.mark_done(db, run_id, counters)
         logger.info(
             "deactivate_stale source=%s run_id=%d done: deactivated=%d "
-            "(ttl_days=%d, segments=%r, staleness_column=%s)",
+            "(ttl_days=%d эффективный, задан %d, segments=%r, staleness_column=%s)",
             listing_source,
             run_id,
             counters["deactivated"],
+            effective_ttl_days,
             ttl_days,
             segments,
             staleness_column,

tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py

@@ -0,0 +1,334 @@
+"""Пол TTL по измеренному циклу переобхода (#2659).
+
+Гейт здоровья (#2710) отвечает «источник собирается?». Этот пол отвечает на второй
+вопрос issue — «а достаточно ли ttl_days, чтобы молчание означало снятие?». Пока свип
+возвращается к строке реже, чем раз в ttl_days, TTL меряет нашу выборку, а не жизнь
+объявления, и источник при этом ЗДОРОВ — гейт молчит.
+
+Ключевой тест — test_effective_ttl_covers_every_proven_false_kill: он проигрывает
+РЕАЛЬНЫЙ прод-случай (127 строк, снятых на возрасте 29.9..30.3 суток при TTL=30 и
+доказанно вернувшихся живыми) и требует, чтобы эффективный TTL накрыл каждую. На
+старом коде — без пола — эффективный TTL остаётся 30, и тест падает на всех срезах.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.tasks import deactivate_stale_avito as task_mod
+
+# ── Прод-замер 2026-08-09 (read-only), из-за которого пол существует ───────────
+# С деплоя гейта 06.08 TTL снял 1 028 строк, 127 уже снова активны (12.4%).
+# В Екатеринбурге — единственном городе с настоящим покрытием — доля ложных 100%.
+_FALSE_KILLS_BY_CITY: dict[str, tuple[int, int]] = {  # срез -> (снято, снова активны)
+    "cian/Екатеринбург": (103, 103),
+    "yandex/Екатеринбург": (24, 24),
+    "cian/без города": (560, 0),
+    "yandex/без города": (341, 0),
+}
+# Возраст строки на момент снятия у всех 127 доказанно ложных снятий, суток.
+_FALSE_KILL_AGE_MIN = 29.9
+_FALSE_KILL_AGE_MAX = 30.3
+
+# Пол, который отдаёт ТОТ ЖЕ запрос на проде (percentile_disc 0.99, окно 3 суток,
+# срез = срез TTL-джобы). Ключ -> (listing_source, segments, ttl_days сейчас, пол).
+_PROD_FLOORS: dict[str, tuple[str, list[str] | None, int, float]] = {
+    "cian/vtorichka": ("cian", ["vtorichka"], 30, 34.0),
+    "yandex/vtorichka": ("yandex", ["vtorichka"], 30, 74.3),
+    "avito/все сегменты": ("avito", None, 10, 69.7),
+}
+
+# Разрывы переобхода по срезу TTL-джобы (listing_source_snapshots, 40 суток):
+# источник -> (p90, p99, ttl_days сейчас). Домклик — контрольная группа: почти
+# полное суточное покрытие, TTL лежит в 4.5 раза выше хвоста, и снятие у него
+# действительно означает снятие.
+_REVISIT_TAIL: dict[str, tuple[float, float, int]] = {
+    "domklik/vtorichka": (1.9, 3.1, 14),
+    "cian/vtorichka": (10.9, 26.6, 30),
+    "yandex/vtorichka": (5.7, 43.0, 30),
+    "avito/vtorichka": (29.1, 42.1, 10),
+}
+
+
+# ── Фейковая сессия ───────────────────────────────────────────────────────────
+
+
+class _FakeResult:
+    def __init__(self, rowcount: int = 0, scalar_value: Any = None) -> None:
+        self.rowcount = rowcount
+        self._scalar = scalar_value
+
+    def scalar(self) -> Any:
+        return self._scalar
+
+
+class _FakeDB:
+    """Session-заглушка: percentile_disc -> пол, count(*) -> подтверждения, UPDATE -> rowcount."""
+
+    def __init__(
+        self,
+        *,
+        floor_days: float | None,
+        confirmations: int = 10_000,
+        rowcount: int = 137,
+    ) -> None:
+        self._floor = floor_days
+        self._confirmations = confirmations
+        self._rowcount = rowcount
+        self.executed: list[tuple[str, dict[str, Any] | None]] = []
+        self.committed = False
+        self.rolled_back = False
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        sql = str(stmt.text)
+        self.executed.append((sql, params))
+        if "percentile_disc" in sql:
+            return _FakeResult(scalar_value=self._floor)
+        if "SELECT count(*)" in sql:
+            return _FakeResult(scalar_value=self._confirmations)
+        return _FakeResult(rowcount=self._rowcount)
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+    @property
+    def floor_query(self) -> tuple[str, dict[str, Any] | None]:
+        return next((e for e in self.executed if "percentile_disc" in e[0]), ("", None))
+
+    @property
+    def update_query(self) -> tuple[str, dict[str, Any] | None]:
+        return next((e for e in self.executed if "UPDATE listings" in e[0]), ("", None))
+
+
+def _run(db: _FakeDB, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> dict[str, int]:
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    return task_mod.deactivate_stale_listings(
+        db,  # type: ignore[arg-type]
+        1,
+        listing_source=kwargs.pop("listing_source", "cian"),
+        ttl_days=kwargs.pop("ttl_days", 30),
+        **kwargs,
+    )
+
+
+# ── Исторический случай: 127 доказанных ложных снятий ─────────────────────────
+
+
+def test_effective_ttl_covers_every_proven_false_kill(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ни одно из 127 доказанно ложных снятий не должно повториться.
+
+    Все они произошли на возрасте 29.9..30.3 суток. Эффективный TTL обязан быть
+    строго выше этого возраста на КАЖДОМ прод-срезе — иначе следующий прогон
+    снимет ту же строку снова.
+    """
+    for slice_name, (source, segments, ttl_days, floor) in _PROD_FLOORS.items():
+        db = _FakeDB(floor_days=floor)
+        out = _run(
+            db,
+            monkeypatch,
+            listing_source=source,
+            ttl_days=ttl_days,
+            segments=segments,
+            revisit_floor_quantile=task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE,
+        )
+        effective = out["ttl_days_effective"]
+        assert effective > _FALSE_KILL_AGE_MAX, (
+            f"{slice_name}: эффективный TTL {effective} не накрывает возраст ложного "
+            f"снятия {_FALSE_KILL_AGE_MAX} — 127 строк снимутся снова"
+        )
+        _, update_params = db.update_query
+        assert update_params is not None
+        assert (
+            update_params["ttl_days"] == effective
+        ), f"{slice_name}: UPDATE получил не поднятый TTL — пол посчитан и выброшен"
+
+
+def test_false_kill_ages_sit_inside_the_old_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Замер согласован сам с собой: снимали ровно на границе TTL=30, не раньше."""
+    assert _FALSE_KILL_AGE_MIN < 30.0 <= _FALSE_KILL_AGE_MAX
+    for source, _segments, ttl_days, _floor in _PROD_FLOORS.values():
+        if source in ("cian", "yandex"):
+            assert ttl_days == 30, f"{source}: прод-TTL разошёлся с замером"
+
+
+def test_false_kill_rate_is_total_where_coverage_is_real() -> None:
+    """В городе с настоящим покрытием ложны ВСЕ снятия — это и есть корень #2659."""
+    for slice_name in ("cian/Екатеринбург", "yandex/Екатеринбург"):
+        killed, returned = _FALSE_KILLS_BY_CITY[slice_name]
+        assert returned == killed, f"{slice_name}: замер разошёлся"
+    total_killed = sum(k for k, _ in _FALSE_KILLS_BY_CITY.values())
+    total_returned = sum(r for _, r in _FALSE_KILLS_BY_CITY.values())
+    assert total_killed == 1028
+    assert total_returned == 127
+
+
+def test_control_group_domklik_needs_no_floor() -> None:
+    """Домклик — контроль: при почти полном суточном обходе TTL с запасом выше хвоста.
+
+    Если бы пол был нужен всем подряд, он был бы нужен и источнику со сплошным
+    покрытием. Ему не нужен — значит меряем именно покрытие, а не «TTL маловат».
+
+    Разделяет источники ЗАПАС над хвостом, а не сам факт превышения: у домклика
+    TTL/p99 = 4.5, у остальных трёх 0.24..1.13, то есть порог сидит вплотную к
+    хвосту или внутри него. Именно у самого «благополучного» из трёх (cian, 1.13)
+    и случились 103 доказанно ложных снятия: p99 запаса не даёт, снимает 1% живых.
+    """
+    _p90, p99, ttl = _REVISIT_TAIL["domklik/vtorichka"]
+    assert ttl / p99 > 4, "домклик перестал быть контрольной группой — перемерить"
+    for name, (_p90, p99, ttl) in _REVISIT_TAIL.items():
+        if name.startswith("domklik"):
+            continue
+        assert ttl / p99 < 1.5, f"{name}: TTL отошёл от хвоста обхода, замер устарел"
+
+
+# ── Контракт пола ─────────────────────────────────────────────────────────────
+
+
+def test_floor_never_lowers_configured_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пол поднимает порог и только. Здоровый источник с коротким циклом не даёт
+    поводу снять больше, чем задано оператором."""
+    db = _FakeDB(floor_days=3.1)
+    out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 30
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 30
+
+
+def test_floor_is_rounded_up_not_down(monkeypatch: pytest.MonkeyPatch) -> None:
+    """34.0 суток разрыва -> TTL 34, а 34.2 -> 35: округление в сторону осторожности."""
+    db = _FakeDB(floor_days=34.2)
+    out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99)
+    assert out["revisit_floor_days"] == 35
+    assert out["ttl_days_effective"] == 35
+
+
+def test_floor_disabled_keeps_old_behaviour(monkeypatch: pytest.MonkeyPatch) -> None:
+    """quantile=0 -> ни одного лишнего запроса, поведение как до правки."""
+    db = _FakeDB(floor_days=74.3)
+    out = _run(db, monkeypatch, ttl_days=30)
+    assert out == {"deactivated": 137}
+    assert len(db.executed) == 1
+    assert "percentile_disc" not in db.executed[0][0]
+
+
+def test_missing_snapshot_history_leaves_ttl_as_configured(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """NULL (истории снимков за окно нет) -> пол не выдумывается, TTL как задан."""
+    db = _FakeDB(floor_days=None)
+    out = _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99)
+    assert "revisit_floor_days" not in out
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 30
+
+
+def test_floor_runs_before_any_write(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пол считается ДО UPDATE: снятое объявление возвращает только повторный сбор."""
+    db = _FakeDB(floor_days=74.3)
+    _run(db, monkeypatch, ttl_days=30, revisit_floor_quantile=0.99)
+    kinds = ["floor" if "percentile_disc" in sql else "update" for sql, _ in db.executed]
+    assert kinds.index("floor") < kinds.index("update")
+
+
+def test_floor_measures_same_slice_as_update(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Срез пола совпадает со срезом UPDATE: тот же source и те же сегменты."""
+    db = _FakeDB(floor_days=74.3)
+    _run(
+        db,
+        monkeypatch,
+        listing_source="yandex",
+        segments=["vtorichka"],
+        revisit_floor_quantile=0.99,
+    )
+    floor_sql, floor_params = db.floor_query
+    assert "ANY(CAST(:segments AS text[]))" in floor_sql
+    assert floor_params is not None
+    assert floor_params["segments"] == ["vtorichka"]
+    assert floor_params["listing_source"] == "yandex"
+
+
+def test_floor_uses_same_staleness_column_as_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
+    """domklik считает свежесть по scraped_at (#2204) — пол обязан мерить ту же колонку,
+    иначе bulk-touch по last_seen_at показал бы цикл обхода там, где сбора нет."""
+    db = _FakeDB(floor_days=20.0)
+    _run(
+        db,
+        monkeypatch,
+        listing_source="domklik",
+        ttl_days=14,
+        staleness_column="scraped_at",
+        revisit_floor_quantile=0.99,
+    )
+    floor_sql, _ = db.floor_query
+    assert "l.scraped_at" in floor_sql
+    assert "l.last_seen_at" not in floor_sql
+
+
+def test_floor_rejects_invalid_staleness_column(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Whitelist колонки работает и на пути пола — интерполяции чужого имени нет."""
+    db = _FakeDB(floor_days=20.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, staleness_column="is_active", revisit_floor_quantile=0.99)
+    assert db.executed == []
+
+
+def test_health_gate_still_wins_over_floor(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Нездоровый источник блокируется гейтом ДО того, как считается пол:
+    лишний тяжёлый запрос по мёртвому источнику не нужен."""
+    db = _FakeDB(floor_days=74.3, confirmations=10)
+    out = _run(
+        db,
+        monkeypatch,
+        min_confirmations=500,
+        revisit_floor_quantile=0.99,
+    )
+    assert out["skipped_unhealthy"] == 1
+    assert db.floor_query[0] == ""
+    assert db.update_query[0] == ""
+
+
+def test_floor_sql_is_psycopg_v3_safe() -> None:
+    sql = str(task_mod._build_revisit_floor_sql("last_seen_at", with_segments=True).text)
+    assert "CAST(:revisit_quantile AS double precision)" in sql
+    assert "CAST(:health_window_days AS integer)" in sql
+    assert not re.search(r":\w+::", sql)
+    assert "UPDATE" not in sql.upper()
+    assert "DELETE" not in sql.upper()
+
+
+def test_floor_only_counts_rows_the_sweep_actually_refound() -> None:
+    """Пол меряет выживших, а не «мы к ним не приходили»: свежесть обязана СДВИНУТЬСЯ
+    относительно предыдущего снимка, иначе разрыв нулевой и хвост схлопнется в ноль."""
+    sql = str(task_mod._build_revisit_floor_sql("last_seen_at", with_segments=False).text)
+    assert "l.last_seen_at > prev.last_seen_at" in sql
+
+
+def test_default_quantile_is_high_enough_for_the_prod_case() -> None:
+    """Ниже 0.99 опускать нельзя без нового замера: именно на 0.99 прод-запрос даёт
+    34.0 для cian/vtorichka, что накрывает возраст ложных снятий 30.3."""
+    assert task_mod.DEFAULT_REVISIT_FLOOR_QUANTILE >= 0.99
+    assert _PROD_FLOORS["cian/vtorichka"][3] > _FALSE_KILL_AGE_MAX
+
+
+def test_handler_wires_revisit_floor_from_schedule_params() -> None:
+    """Читаем исходник файлом: product_handlers тянет scraper_kit, которого в
+    юнит-окружении может не быть, а проверяем мы проводку, а не импорт."""
+    handlers = Path(__file__).resolve().parents[1] / "app" / "services" / "product_handlers.py"
+    src = handlers.read_text("utf-8")
+    job = src.split("async def _job_deactivate_stale")[1].split("\nasync def ")[0]
+    flat = " ".join(job.split())
+    assert 'params.get( "revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE )' in flat
+    assert "revisit_floor_quantile=revisit_floor_quantile" in job