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

@@ -217,6 +217,7 @@ async def _job_deactivate_stale(
 ) -> None:
     from app.core.config import settings as _settings
     from app.tasks.deactivate_stale_avito import (
+        CAP_MULT,
         DEFAULT_MIN_CONFIRMATIONS,
         DEFAULT_REVISIT_FLOOR_QUANTILE,
         deactivate_stale_listings,
@@ -236,6 +237,11 @@ async def _job_deactivate_stale(
     revisit_floor_quantile: float = params.get(
         "revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE
     )
+    # Потолок эффективного TTL (см. CAP_MULT в deactivate_stale_avito.py) — множитель,
+    # а не голая константа: источник с непропорционально длинным хвостом переобхода
+    # относительно своего ttl_days переопределяет его через default_params (ключ
+    # "cap_mult"), не трогая дефолт для остальных источников.
+    cap_mult: float = params.get("cap_mult", CAP_MULT)
 
     loop = asyncio.get_event_loop()
     await loop.run_in_executor(
@@ -249,6 +255,7 @@ async def _job_deactivate_stale(
             staleness_column=staleness_column,
             min_confirmations=min_confirmations,
             revisit_floor_quantile=revisit_floor_quantile,
+            cap_mult=cap_mult,
         ),
     )
 

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

@@ -7,6 +7,16 @@
 - Cian/Yandex не поддерживают full-coverage sweep -> паушальный TTL сломает живой
   инвентарь. DECISION: для yandex/cian деактивировать ТОЛЬКО listing_segment='vtorichka',
   TTL=30. novostroyki (9659 активных первичных строк) и NULL-сегмент не трогаем.
+  ИЗВЕСТНЫЙ ПРОБЕЛ (ревью TTL-CAP круг 2, 2026-08-15): этот скоуп уже, чем множество
+  реально протухших строк -- живой замер на проде даёт cian/novostroyki 9 483,
+  cian/NULL-сегмент 211, yandex/NULL-сегмент 523 активных строки старше 60 суток, ни
+  одна из них не деактивируется НИ ОДНОЙ джобой (внутри скоупа cian/vtorichka и
+  yandex/vtorichka таких строк 0). Потолок cap_mult (см. CAP_MULT ниже) этот пробел
+  не закрывает и закрыть не может -- он сжимает пул ВНУТРИ скоупа джобы, а не
+  расширяет сам скоуп. Расширение скоупа -- отдельная задача (нужно сперва выяснить,
+  поддерживают ли cian/yandex full-coverage sweep для novostroyki/NULL-сегмента
+  СЕЙЧАС, иначе паушальный TTL повторит инцидент, ради которого этот DECISION и
+  принят) и намеренно НЕ входит в TTL-CAP.
 - avito: все сегменты (segments=None), TTL=10 дней -- поведение без изменений.
 - Строки НЕ удаляются -- история нужна для бэктеста (#667).
 - #2674: деактивация в той же транзакции пишет снимок listings_snapshots со статусом
@@ -190,6 +200,66 @@ DEFAULT_REVISIT_FLOOR_QUANTILE = 0.99
 _REVISIT_FLOOR_SEGMENT_FILTER = "\n          AND l.listing_segment = ANY(CAST(:segments AS text[]))"
 
 
+# ── Потолок эффективного TTL (положительная обратная связь пола, найдено 2026-08-15) ──
+# У пола выше нет верхней границы: max(ttl_days, пол) может расти неограниченно.
+# ЗАМЕР НА ПРОДЕ (уточнён 2026-08-15 после разбора): у yandex counters держали
+# ttl_days_effective 75/75/75/39/52/54 шесть прогонов подряд при deactivated=0 —
+# пол реально разгонялся без верхней границы, и потолок закрывает именно это.
+# ЧЕГО ПОТОЛОК НЕ ДЕЛАЕТ: он НЕ сжимает пул «активных». Замер показал 0
+# деактивируемых строк на всех четырёх джобах и до, и после калибровки. Цифра
+# «23 687 из 44 744 не подтверждались >7 суток» относится ко ВСЕМ источникам
+# сразу, и две трети её — новостройки, которых оценщик не берёт. У avito
+# просроченных ноль. Раздутый пул, влияющий на оценку, лежит в строках с ПУСТЫМ
+# сегментом и чинится отдельной джобой, не этим потолком.
+#
+# МЕХАНИЗМ ПЕТЛИ: медленный обход поднимает пол (он же квантиль разрывов переобхода)
+# -> высокий пол продлевает жизнь снятым лотам дольше, чем к ним успевает вернуться
+# свежий обход -> пул «активных» раздувается «протухшими» строками -> следующий замер
+# пола на том же раздутом пуле оказывается ещё выше. Без верхней границы это не
+# самокорректирующийся пол, а положительная обратная связь.
+#
+# CAP_MULT = 2 -- эффективный TTL не может превысить удвоенный заданный оператором
+# ttl_days. Пол по-прежнему может его поднять (ради #2659 -- см. комментарий выше:
+# ложные снятия при неполном покрытии обхода), но не бесконечно. Почему именно 2, а
+# не 3 или 1.5: вдвое — это ещё «мы искренне не уверены, что молчание значит
+# снятие», не «источник вообще умер». Дальнейший рост пола сигнализирует не о
+# медленном, но живом обходе, а о мёртвом источнике -- для ЭТОГО случая уже есть
+# отдельный гейт по здоровью (min_confirmations) выше в этой же функции, который
+# выключает деактивацию целиком, а не растягивает TTL до бесконечности. Калибровочная
+# ручка, не догма -- при новом замере можно пересмотреть, как и revisit_floor_quantile.
+#
+# ПОЧЕМУ MULT, А НЕ ФИКСИРОВАННОЕ ЧИСЛО СУТОК -- И ГДЕ ЭТА ФОРМА ЛОМАЕТСЯ. Множитель
+# от ttl_days даёт разный АБСОЛЮТНЫЙ потолок на разных источниках: cian/yandex
+# (ttl=30) -> 60 суток, avito (ttl=10) -> 20 суток, domklik (ttl=14) -> 28 суток. Это
+# ломается ровно там, где абсолютный хвост переобхода источника НЕ пропорционален его
+# ttl_days. Замер (_REVISIT_TAIL, 40 суток): avito p99 = 42.1 сут -- ВЫШЕ его же
+# потолка 20. То есть для avito дефолтный CAP_MULT=2 может резать ttl ниже
+# собственного хвоста обхода -- ровно тот false-kill, ради которого пол вообще
+# заведён (см. комментарий выше). domklik (потолок 28 при хвосте 3.1) разрыва не
+# имеет -- множитель 2 для него калиброван верно.
+#
+# YANDEX -- ТА ЖЕ ДЫРА, НАЙДЕНА ПОЗЖЕ (ревью круга 3, 2026-08-15). Строка выше до
+# этой правки утверждала, что cian/yandex с потолком 60 тоже в порядке -- это было
+# верно для cian (live-пол сейчас 31.1), но НЕ для yandex: ЖИВЫЕ полы из
+# scrape_runs.counters (deactivate_stale_yandex, 2026-08-10..08-15) -- 75/75/75/39/
+# 52/54, а прямой live-замер той же percentile_disc(0.99)-формулы сегодня даёт 79.2
+# (n=1961 подтверждений за 3 суток). И то, и другое ВЫШЕ потолка 60 -- тот же
+# false-kill класс, что у avito, статический p99=43.0 (_REVISIT_TAIL) для yandex
+# устарел и вводит в заблуждение. cap_mult для yandex откалиброван отдельной
+# миграцией (265_deactivate_stale_yandex_cap_mult.sql, cap_mult=3 -> потолок 90) --
+# см. её комментарий про то, почему это НЕ меняет число деактивированных строк
+# следующим прогоном (0 активных строк источника старше 39 суток на момент замера).
+#
+# ПОЭТОМУ cap_mult -- параметр функции (как revisit_floor_quantile, min_confirmations),
+# не голая константа: default = CAP_MULT для источников, где 2x достаточно (cian,
+# domklik), но расписание может переопределить через default_params (JSON-колонка
+# scrape_schedules, ключ "cap_mult") для источника с непропорционально длинным
+# хвостом -- см. миграции для avito (cap_mult=6, потолок 60, с запасом выше
+# статического p99=42.1 и живого прод-пика 52, замеренного 2026-08-10..12) и yandex
+# (cap_mult=3, потолок 90, с запасом выше живого пола 79.2, замеренного 2026-08-15).
+CAP_MULT = 2
+
+
 def _build_revisit_floor_sql(staleness_column: str, *, with_segments: bool) -> Any:
     """Квантиль возраста, при котором свип за окно ДОКАЗАЛ, что строка жива.
 
@@ -312,6 +382,7 @@ def deactivate_stale_listings(
     min_confirmations: int = 0,
     health_window_days: int = _HEALTH_WINDOW_DAYS,
     revisit_floor_quantile: float = 0.0,
+    cap_mult: float = CAP_MULT,
 ) -> dict[str, int]:
     """Пометить is_active=false объявления, чья свежесть старше ttl_days дней.
 
@@ -334,9 +405,17 @@ def deactivate_stale_listings(
         health_window_days: окно подтверждений для гейта, суток. Дефолт 3.
         revisit_floor_quantile: пол TTL по измеренному циклу переобхода (#2659).
             Квантиль возраста, при котором свип за окно ДОКАЗАЛ строку живой;
-            эффективный TTL = max(ttl_days, этот пол). 0 -> пол выключен (так
+            эффективный TTL = min(max(ttl_days, этот пол), ttl_days * cap_mult) --
+            пол поднимает TTL, но не выше потолка. 0 -> пол выключен (так
             вызывают старые тесты и совместимая обёртка), рабочее значение —
             DEFAULT_REVISIT_FLOOR_QUANTILE, см. комментарий выше.
+        cap_mult: множитель потолка эффективного TTL (см. комментарий у модульной
+            константы CAP_MULT). Дефолт -- сама CAP_MULT=2, но параметр, а НЕ голая
+            константа: источник с непропорционально длинным хвостом переобхода
+            относительно своего ttl_days (avito: p99=42.1 при ttl=10 -> дефолтный
+            потолок 20 режет ниже хвоста) может переопределить его через
+            default_params расписания (ключ "cap_mult"), не трогая остальные
+            источники. Итоговый потолок = ttl_days * cap_mult.
 
     Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
     Один statement в транзакции: UPDATE флага + снимок 'stale' в listings_snapshots
@@ -345,14 +424,56 @@ def deactivate_stale_listings(
     Returns {"deactivated": N} -- количество обновлённых строк (1:1 со снимками).
     Если гейт не пропустил прогон: {"deactivated": 0, "confirmations": N,
     "skipped_unhealthy": 1} и НИ ОДНА строка не тронута. Если пол переобхода поднял
-    TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}.
+    TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}. Если пол
+    упёрся в потолок cap_mult: дополнительно {"ttl_floor_capped": 1,
+    "ttl_days_floor_raw": N} -- N это то, во что пол поднял бы TTL БЕЗ потолка.
 
     Raises:
-        ValueError: если staleness_column не входит в whitelist (проверка ДО SQL,
+        ValueError: если staleness_column не входит в whitelist, ИЛИ ttl_days <= 0
-            никакой интерполяции пользовательского ввода в запрос).
+            (проверка ДО SQL, никакой интерполяции пользовательского ввода в запрос;
+            ttl_days<=0 в WHERE-условии last_seen_at < NOW() - INTERVAL 'N days'
+            матчит практически весь активный пул -- без явного guard'а потолок
+            (ttl_days * cap_mult <= 0) к тому же перебивал бы пол в формуле min(),
+            снимая защиту, которую max(ttl_days, floor) давал раньше), ИЛИ cap_mult < 1
+            (тот же класс дыры, но со стороны потолка, а не пола: cap_mult приходит из
+            jsonb default_params расписания -- ЕДИНСТВЕННЫЙ запланированный способ его
+            задать, т.е. именно там опечатка 0 / 0.5 вместо 6 доходит до прода. cap_mult=0
+            даёт capped=0 -> effective_ttl_days=0 -> UPDATE снимает практически весь
+            активный пул источника; cap_mult<1 (например 0.5) опускает потолок НИЖЕ
+            заданного оператором ttl_days -- прямое нарушение инварианта «потолок не
+            может понизить TTL ниже настроенного», который проверяет
+            test_cap_never_lowers_ttl_below_configured_value), ИЛИ ttl_days/cap_mult --
+            bool (найдено ревью круга 3, 2026-08-15: `cap_mult < 1` пропускает `True` --
+            `bool` наследует `int`, `True < 1` ложно, а `ttl_days * True` == `ttl_days`,
+            то есть потолок = сам ttl_days и пол молча отключается, никакого ValueError.
+            jsonb `true`/`false` вместо числа -- ровно та опечатка в расписании, ради
+            которой оба guard'а вообще написаны, поэтому bool отклоняется явной
+            type-проверкой ДО числового сравнения для обоих параметров).
     """
     counters: dict[str, int] = {"deactivated": 0}
     try:
+        # bool -- подкласс int в Python, поэтому `True < 1` (False) и `False <= 0`
+        # (True) НЕ ловят опечатку `"ttl_days": true` / `"cap_mult": true` в jsonb:
+        # `ttl_days * True` == `ttl_days`, `cap_mult=True` даёт потолок == ttl_days и
+        # молча отключает пол (см. Raises выше). Проверка типа -- ДО числового
+        # сравнения, иначе bool проскакивает мимо него необнаруженным.
+        if isinstance(ttl_days, bool):
+            raise ValueError(f"ttl_days must be a number, not bool: {ttl_days!r}")
+        if ttl_days <= 0:
+            raise ValueError(f"ttl_days must be positive, got {ttl_days!r}")
+
+        # Тот же класс дыры, что и ttl_days<=0 выше, только со стороны потолка:
+        # cap_mult < 1 может опустить потолок (ttl_days * cap_mult) НИЖЕ заданного
+        # ttl_days, а cap_mult <= 0 -- сделать капнутый потолок <= 0 и победить пол
+        # в min() молча (ровно та дыра, ради которой заведён guard выше). Единственный
+        # запланированный способ задать cap_mult -- вписать его руками в jsonb
+        # default_params расписания (см. миграцию для avito), т.е. именно там опечатка
+        # 0 / 0.5 вместо 6 -- реальный риск, а не гипотетика.
+        if isinstance(cap_mult, bool):
+            raise ValueError(f"cap_mult must be a number, not bool: {cap_mult!r}")
+        if cap_mult < 1:
+            raise ValueError(f"cap_mult must be >= 1, got {cap_mult!r}")
+
         # Whitelist-проверка ДО построения/выполнения SQL: только после неё имя колонки
         # интерполируется f-string'ом. Значения по-прежнему идут через param-binding.
         # Внутри try -> невалидная колонка финализирует run как failed (mark_failed),
@@ -402,7 +523,9 @@ def deactivate_stale_listings(
                 return counters
 
         # Пол TTL по измеренному циклу переобхода (#2659) — тоже ДО UPDATE и по тому же
-        # срезу. Поднимает порог, никогда не опускает: max(), а не замена.
+        # срезу. Поднимает порог (max), но не выше потолка cap_mult * ttl_days (min) —
+        # см. комментарий у CAP_MULT про петлю с положительной обратной связью и про
+        # то, почему cap_mult -- параметр, а не голая константа.
         effective_ttl_days = ttl_days
         if revisit_floor_quantile > 0:
             floor_params: dict[str, Any] = {
@@ -420,9 +543,39 @@ def deactivate_stale_listings(
             # Тогда пола нет и 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"])
+                # Пол поднимает TTL (max), потолок cap_mult его не пускает выше
+                # ttl_days * cap_mult (min) — без этого пол растёт без ограничения
+                # (см. комментарий у CAP_MULT). capped_ttl_days может быть float,
+                # если cap_mult переопределён нецелым значением из default_params —
+                # effective_ttl_days приводим к int (UPDATE ждёт целые сутки).
+                raw_effective_ttl_days = max(ttl_days, counters["revisit_floor_days"])
+                capped_ttl_days = ttl_days * cap_mult
+                effective_ttl_days = int(min(raw_effective_ttl_days, capped_ttl_days))
                 counters["ttl_days_effective"] = effective_ttl_days
-                if effective_ttl_days > ttl_days:
+
+                if raw_effective_ttl_days > capped_ttl_days:
+                    # Пол упёрся в потолок -- оба числа в counters (не только в логе),
+                    # чтобы это было видно в витрине прогонов, а не только в логах.
+                    # 1, а не True -- counters типизирован dict[str, int] (тот же
+                    # идиом, что skipped_unhealthy выше).
+                    counters["ttl_floor_capped"] = 1
+                    counters["ttl_days_floor_raw"] = raw_effective_ttl_days
+                    logger.warning(
+                        "deactivate_stale source=%s run_id=%d TTL пол упёрся в потолок "
+                        "cap_mult=%s: пол поднял бы TTL до %d сут, потолок ограничивает "
+                        "заданные %d сут значением %d (квантиль %.3f, segments=%r) — "
+                        "растущий без ограничения пол это петля с положительной обратной "
+                        "связью, см. комментарий у CAP_MULT",
+                        listing_source,
+                        run_id,
+                        cap_mult,
+                        raw_effective_ttl_days,
+                        ttl_days,
+                        effective_ttl_days,
+                        revisit_floor_quantile,
+                        segments,
+                    )
+                elif effective_ttl_days > ttl_days:
                     logger.warning(
                         "deactivate_stale source=%s run_id=%d TTL поднят с %d до %d сут: "
                         "свип за %d сут доказал живой строку, молчавшую %d сут "

tradein-mvp/backend/data/sql/264_deactivate_stale_avito_cap_mult.sql

@@ -0,0 +1,83 @@
+-- 264_deactivate_stale_avito_cap_mult.sql
+-- Калибрует потолок эффективного TTL (cap_mult) для avito (#TTL-CAP, 2026-08-15).
+--
+-- ЗАЧЕМ. Пол TTL по измеренному циклу переобхода (#2659, deactivate_stale_avito.py)
+-- поднимает эффективный TTL через max(ttl_days, пол) без верхней границы -- на проде
+-- это оказалось петлёй с положительной обратной связью: медленный обход поднимает
+-- пол, высокий пол продлевает жизнь снятым лотам дольше, чем к ним успевает
+-- вернуться свежий обход, пул «активных» раздувается протухшими строками. ВАЖНАЯ
+-- ОГОВОРКА (перепроверено 2026-08-15): цифра «23 687 из 44 744» -- это ВСЕ источники
+-- вместе, и две трети её -- новостройки, которые оценщик не берёт вообще. У самого
+-- avito просроченных строк НОЛЬ (8 663 активных, максимальный возраст 10 суток) --
+-- его деактивация работает исправно. Этот потолок существует не ради сжатия пула
+-- (он деактивирует 0 строк, замерено), а как защита от опечатки в расписании и от
+-- будущего разгона пола. Потолок cap_mult ограничивает пол сверху: эффективный TTL не
+-- может превысить ttl_days * cap_mult (код -- app/tasks/deactivate_stale_avito.py,
+-- CAP_MULT).
+--
+-- ПОЧЕМУ ИМЕННО AVITO. Дефолт CAP_MULT=2 даёт разный АБСОЛЮТНЫЙ потолок на разных
+-- источниках (множитель от ttl_days), и ломается там, где хвост переобхода
+-- источника НЕ пропорционален его ttl_days. Таблица ниже -- ЖИВЫЕ полы из
+-- scrape_runs.counters (ttl_days_effective/revisit_floor_days по каждой job'е за
+-- 2026-08-10..08-15, ПЕРЕСЧИТАНО ревью круга 3 2026-08-15 -- прежняя версия таблицы
+-- брала статический p99 из _REVISIT_TAIL (40-суточный замер на более раннюю дату)
+-- и по нему ошибочно утверждала «yandex 43.0 -> потолок 60, запас есть»; live-полы
+-- показывают обратное, см. ниже), а не по статической константе:
+--   источник/сегмент      живой пол (6 прогонов)    ttl_days   потолок cap_mult=2
+--   domklik  vtorichka    23/24/25/skip/skip/skip       14        28   (запас есть)
+--   cian     vtorichka    34/34/37/27/27/32             30        60   (запас есть)
+--   yandex   vtorichka    75/75/75/39/52/54             30        60   (ХВОСТ ВЫШЕ)
+--   avito    все сегменты 52/52/52/7/8/9                10        20   (ХВОСТ ВЫШЕ)
+-- У avito p99=42.1 суток (_REVISIT_TAIL) и живой пик 52 -- ВЫШЕ его же дефолтного
+-- потолка 20: дефолтный cap_mult=2 может резать пол ниже собственного хвоста
+-- обхода, то есть ровно тот false-kill, ради которого пол вообще заведён.
+--
+-- YANDEX -- ТА ЖЕ ДЫРА, что и у avito, но найдена ПОЗЖЕ (при первой версии этой
+-- миграции статический p99=43.0 ошибочно считался достаточным запасом). Живой пол
+-- yandex/vtorichka держится 39-75 суток шесть прогонов подряд, а прямой live-замер
+-- 2026-08-15 (та же percentile_disc(0.99)-формула, что и в проде) даёт 79.2 суток
+-- (n=1961 подтверждений за 3 суток) -- выше потолка 60 при дефолтном cap_mult=2.
+-- Калибровка yandex вынесена в ОТДЕЛЬНУЮ миграцию
+-- (265_deactivate_stale_yandex_cap_mult.sql, cap_mult=3 -> потолок 90), не сюда --
+-- эта миграция специфична для avito по имени и назначению, смешивать источники в
+-- одном файле хуже для git-истории калибровок. cian и domklik разрыва не имеют,
+-- дефолт cap_mult=2 для них по-прежнему калиброван верно, эта миграция их не трогает.
+--
+-- ЧИСЛЕННЫЙ ЭФФЕКТ (обе миграции, 264+265, live-замер 2026-08-15): на пул активных
+-- строк не влияет ни у одного из четырёх источников -- next-run deactivated=0 что до,
+-- что после калибровки. У avito и cian живой пол (12/32 суток) уже ниже потолка --
+-- калибровка cap_mult просто не участвует в min(). У yandex 0 активных строк старше
+-- 39 суток вообще (весь "просроченный" хвост младше того возраста, где потолок
+-- 60 vs 90 может разойтись), поэтому даже БЕЗ калибровки (дефолт cap_mult=2,
+-- потолок 60 < живой пол 79.2) next-run deactivated тоже 0 -- калибровка убирает
+-- будущий риск (потолок бы капал ttl_days_effective 79->60 в counters и резал бы
+-- ниже собственного хвоста обхода, как только появятся строки в возрастной полосе
+-- 60-90 суток), а не текущее число. domklik заблокирован гейтом здоровья
+-- (confirmations 94 < min_confirmations 200) -- до потолка/пола дело не доходит.
+--
+-- ПОЧЕМУ 6. Потолок 60 = 10 * 6 -- тот же порядок, что у cian (60, дефолт cap_mult=2),
+-- с запасом выше и статического p99=42.1 (_REVISIT_TAIL, tests/test_deactivate_stale_revisit_floor.py),
+-- и живого прод-пика: floor=52 три прогона подряд 2026-08-10..08-12
+-- (scrape_runs.counters, status=done, confirmations 6934..7138, гейт здоровья
+-- пропустил). Без этой калибровки в проде остаётся дефолт cap_mult=2 (потолок 20)
+-- -- именно тот случай, для которого потолок и его собственная калибровочная ручка
+-- заведены, но не применены к единственному источнику, ради которого ручка сделана.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)), 219 (тот же
+-- приём -- UPDATE default_params через jsonb ?, min_confirmations).
+-- ТОЛЬКО данные (UPDATE default_params), DDL нет.
+-- Идемпотентность + уважение к ручной настройке: ключ проставляется лишь там, где
+-- его ещё нет, поэтому повторный прогон файла не затирает подкрученное оператором
+-- значение. Снять/поднять потолок вручную: cap_mult в default_params
+-- (deactivate_stale_avito), 1 -> потолок = сам ttl_days (см. guard cap_mult < 1
+-- в deactivate_stale_listings -- ниже 1 отклоняется до любого SQL).
+
+BEGIN;
+
+UPDATE scrape_schedules
+SET default_params = default_params || jsonb_build_object('cap_mult', 6),
+    updated_at = NOW()
+WHERE source = 'deactivate_stale_avito'
+  AND NOT default_params ? 'cap_mult';
+
+COMMIT;

tradein-mvp/backend/data/sql/265_deactivate_stale_yandex_cap_mult.sql

@@ -0,0 +1,57 @@
+-- 265_deactivate_stale_yandex_cap_mult.sql
+-- Калибрует потолок эффективного TTL (cap_mult) для yandex (#TTL-CAP круг 3, 2026-08-15).
+--
+-- ЗАЧЕМ. Та же дыра, что закрыта для avito миграцией
+-- 264_deactivate_stale_avito_cap_mult.sql (см. её комментарий про механизм петли),
+-- но обнаружена на yandex позже: первая версия 264 утверждала, что дефолтный
+-- CAP_MULT=2 (потолок 60 при ttl_days=30) для yandex "калиброван верно" на
+-- основании статического p99=43.0 (_REVISIT_TAIL, замер на более раннюю дату).
+--
+-- ЖИВОЙ ЗАМЕР, из-за которого миграция существует. scrape_runs.counters
+-- (deactivate_stale_yandex, 2026-08-10..08-15) держал ttl_days_effective 75/75/75/
+-- 39/52/54 шесть прогонов подряд при deactivated=0 -- то есть пол ВСЕ ЭТИ ДНИ был
+-- выше потолка 60. Прямой live-замер той же percentile_disc(0.99)-формулы, что и в
+-- коде (app/tasks/deactivate_stale_avito.py, _build_revisit_floor_sql), 2026-08-15
+-- даёт 79.2 суток (n=1961 подтверждений за окно 3 суток). Оба замера выше потолка
+-- 60 -- ровно тот false-kill, ради которого пол #2659 вообще заведён: без калибровки
+-- потолок капал бы ttl_days_effective yandex до 60 в counters уже сегодня и резал бы
+-- ниже собственного хвоста обхода, как только в пуле появятся строки возрастом
+-- 60-90 суток (сейчас таких 0 -- см. ЧИСЛЕННЫЙ ЭФФЕКТ ниже).
+--
+-- ПОЧЕМУ 3. Потолок 90 = 30 * 3 -- запас ~14% над живым пиком 79.2, той же
+-- пропорции, что и у avito (потолок 60 против пика 52 -- запас ~15%, см. 264).
+-- Меньший cap_mult=2 (потолок 60) уже сейчас ниже пика 79.2. Больший cap_mult
+-- намеренно не берём -- дальнейший рост пола означает не "медленный, но живой
+-- обход", а кандидата в mёртвый источник, для которого есть отдельный гейт
+-- здоровья (min_confirmations), а не растягивание потолка до бесконечности (см.
+-- комментарий у CAP_MULT в deactivate_stale_avito.py).
+--
+-- ЧИСЛЕННЫЙ ЭФФЕКТ (live-замер 2026-08-15): 0 активных строк yandex/vtorichka
+-- старше 39 суток вообще (запрос: count(*) FROM listings WHERE source='yandex' AND
+-- listing_segment='vtorichka' AND is_active=true AND last_seen_at < NOW() -
+-- INTERVAL 'N days', N=39/52/54/60/75/79 -- везде 0). Next-run deactivated=0 что
+-- при дефолтном cap_mult=2 (потолок 60, капает пол), что при cap_mult=3 из этой
+-- миграции (потолок 90, не капает) -- эта миграция убирает БУДУЩИЙ риск
+-- false-kill при появлении строк в полосе 60-90 суток, а не текущее число
+-- деактиваций. Ветка #TTL-CAP не сжимает пул ни у одного из четырёх источников --
+-- см. 264 для остальных трёх.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)), 219 (тот же
+-- приём -- UPDATE default_params через jsonb ?, min_confirmations), 264 (тот же
+-- приём для avito, cap_mult -- параметр deactivate_stale_listings).
+-- ТОЛЬКО данные (UPDATE default_params), DDL нет.
+-- Идемпотентность + уважение к ручной настройке: ключ проставляется лишь там, где
+-- его ещё нет, поэтому повторный прогон файла не затирает подкрученное оператором
+-- значение. Снять/поднять потолок вручную: cap_mult в default_params
+-- (deactivate_stale_yandex), 1 -> потолок = сам ttl_days (см. guard cap_mult < 1
+-- в deactivate_stale_listings -- ниже 1 отклоняется до любого SQL).
+
+BEGIN;
+
+UPDATE scrape_schedules
+SET default_params = default_params || jsonb_build_object('cap_mult', 3),
+    updated_at = NOW()
+WHERE source = 'deactivate_stale_yandex'
+  AND NOT default_params ? 'cap_mult';
+
+COMMIT;

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

@@ -252,3 +252,5 @@
 261_listings_search_mv_drop_placeholder_columns.sql
 262_scrape_schedules_seed_oblast_city_sweeps_wave2.sql
 263_scrape_schedules_wave2_cian_newbuilding_only_false.sql
+264_deactivate_stale_avito_cap_mult.sql
+265_deactivate_stale_yandex_cap_mult.sql

tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py

@@ -126,13 +126,21 @@ def _run(db: _FakeDB, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> dict[st
 
 
 def test_effective_ttl_covers_every_proven_false_kill(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Ни одно из 127 доказанно ложных снятий не должно повториться.
+    """Ни одно из 127 доказанно ложных снятий (cian/yandex) не должно повториться.
 
     Все они произошли на возрасте 29.9..30.3 суток. Эффективный TTL обязан быть
-    строго выше этого возраста на КАЖДОМ прод-срезе — иначе следующий прогон
+    строго выше этого возраста на cian/yandex-срезах — иначе следующий прогон
-    снимет ту же строку снова.
+    снимет ту же строку снова. avito из этого цикла исключён намеренно: 127
+    доказанных ложных снятий (_FALSE_KILLS_BY_CITY) измерены только по cian/yandex,
+    у avito другой сценарий и своя проверка ниже
+    (test_avito_prod_floor_is_capped_by_calibrated_cap_mult) -- калибровка cap_mult=6
+    для avito (миграция 264_deactivate_stale_avito_cap_mult.sql) пиннится ТАМ, а не
+    здесь, чтобы не смешивать два разных замера под одним порогом
+    _FALSE_KILL_AGE_MAX, который к avito не относится.
     """
     for slice_name, (source, segments, ttl_days, floor) in _PROD_FLOORS.items():
+        if source == "avito":
+            continue
         db = _FakeDB(floor_days=floor)
         out = _run(
             db,
@@ -154,6 +162,108 @@ def test_effective_ttl_covers_every_proven_false_kill(monkeypatch: pytest.Monkey
         ), f"{slice_name}: UPDATE получил не поднятый TTL — пол посчитан и выброшен"
 
 
+def _read_cap_mult_from_migration(filename: str, *, source: str) -> int:
+    """Читает cap_mult из UPDATE default_params миграции -- НЕ хардкодит дубль в тесте.
+
+    Найдено ревью круга 3 2026-08-15: раньше тест ниже принимал cap_mult=6 как
+    аргумент напрямую, захардкоженный прямо в теле теста. Мутация значения в
+    264_deactivate_stale_avito_cap_mult.sql (6 -> 2) НЕ трогала вход теста вовсе --
+    набор оставался зелёным при любом реальном значении в миграции, то есть
+    калибровка нигде не была пином, только упоминанием в комментарии. Здесь
+    значение читается ИЗ ФАЙЛА миграции regex'ом, а ожидаемый результат
+    (ttl_days_effective, ttl_floor_capped) остаётся зафиксированным числом в самом
+    тесте -- так дрейф калибровки в миграции ломает тест, как и задумано.
+    """
+    migration = Path(__file__).resolve().parents[1] / "data" / "sql" / filename
+    src = migration.read_text("utf-8")
+    # Порядок в файле -- jsonb_build_object('cap_mult', N) в SET, ЗАТЕМ WHERE source
+    # = '<source>' ниже (см. 264/265_*.sql). DOTALL матчит перевод строки между ними;
+    # source в regex -- страховка от чтения не того UPDATE, если файл когда-нибудь
+    # станет мульти-source (сейчас в каждом файле ровно один UPDATE).
+    match = re.search(
+        r"jsonb_build_object\('cap_mult',\s*(\d+)\).*?WHERE\s+source\s*=\s*'"
+        + re.escape(source)
+        + r"'",
+        src,
+        re.DOTALL,
+    )
+    assert match is not None, (
+        f"{filename} сменил формат UPDATE default_params для source={source!r} -- "
+        "обнови regex в _read_cap_mult_from_migration"
+    )
+    return int(match.group(1))
+
+
+def test_avito_prod_floor_is_capped_by_calibrated_cap_mult(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пиннит калибровку cap_mult=6 для avito (миграция
+    264_deactivate_stale_avito_cap_mult.sql) на измеренном прод-поле _PROD_FLOORS
+    ("avito/все сегменты" = 69.7, замер 2026-08-09).
+
+    cap_mult -- ВХОД теста, читается ИЗ ФАЙЛА миграции (regex), не хардкодится
+    здесь: дрейф калибровки в 264_*.sql (например 6 -> 2) меняет вход, но НЕ
+    ожидаемый результат ниже (60/70) -- эти числа пинят калибровку саму по себе,
+    поэтому дрейф ломает тест, как и задумано (см. _read_cap_mult_from_migration).
+
+    С дефолтным cap_mult=2 потолок avito (20 сут) РЕЖЕТ ниже собственного хвоста
+    переобхода p99=42.1 (_REVISIT_TAIL) -- ровно тот false-kill, ради которого пол
+    заведён. С калиброванным cap_mult=6 потолок 60 сут -- выше и p99=42.1, и живого
+    прод-пика 52 (замер 08-10..08-12), и этого гипотетического замера 69.7 (капается
+    ровно на 60, не пропускается как есть).
+    """
+    calibrated_cap_mult = _read_cap_mult_from_migration(
+        "264_deactivate_stale_avito_cap_mult.sql", source="deactivate_stale_avito"
+    )
+    source, segments, ttl_days, floor = _PROD_FLOORS["avito/все сегменты"]
+    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,
+        cap_mult=calibrated_cap_mult,
+    )
+    assert out["ttl_days_effective"] == 60, "cap_mult из миграции 264 обязан дать потолок 60"
+    assert out["ttl_floor_capped"] == 1
+    assert out["ttl_days_floor_raw"] == 70, "ceil(69.7) == 70 -- пол считается по real-числу"
+
+
+def test_yandex_prod_floor_is_not_capped_by_calibrated_cap_mult(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Пиннит калибровку cap_mult=3 для yandex (миграция
+    265_deactivate_stale_yandex_cap_mult.sql, найдено ревью круга 3 2026-08-15) на
+    измеренном прод-поле _PROD_FLOORS ("yandex/vtorichka" = 74.3).
+
+    cap_mult -- ВХОД теста, читается ИЗ ФАЙЛА миграции 265 (тот же приём, что и у
+    avito выше): дрейф калибровки в 265_*.sql ломает тест.
+
+    С дефолтным cap_mult=2 потолок yandex (60 сут) РЕЖЕТ живой пол (75-79 сут,
+    scrape_runs.counters 08-10..08-15 и live-замер 08-15) -- та же дыра, что у
+    avito, найдена позже (первая версия 264 ошибочно считала yandex безопасным по
+    устаревшему статическому p99=43.0). С калиброванным cap_mult=3 потолок 90 сут
+    выше живого пика 79.2 -- пол 74.3 из этого теста НЕ капается, эффективный TTL
+    равен сырому полу (75, ceil(74.3)).
+    """
+    calibrated_cap_mult = _read_cap_mult_from_migration(
+        "265_deactivate_stale_yandex_cap_mult.sql", source="deactivate_stale_yandex"
+    )
+    source, segments, ttl_days, floor = _PROD_FLOORS["yandex/vtorichka"]
+    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,
+        cap_mult=calibrated_cap_mult,
+    )
+    assert out["ttl_days_effective"] == 75, "ceil(74.3) == 75, потолок 90 не должен резать"
+    assert "ttl_floor_capped" not in out, "потолок 90 выше живого пола 74.3 -- капать нечего"
+
+
 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

tradein-mvp/backend/tests/test_deactivate_stale_ttl_cap.py

@@ -0,0 +1,443 @@
+"""Потолок эффективного TTL деактивации (найдено на проде 2026-08-15).
+
+Пол TTL по измеренному циклу переобхода (#2659, deactivate_stale_avito.py) поднимает
+эффективный TTL через max(ttl_days, пол) без верхней границы. На проде это оказалось
+петлёй с положительной обратной связью: медленный обход поднимает пол, высокий пол
+продлевает жизнь снятым лотам дольше, чем к ним успевает вернуться свежий обход, пул
+«активных» раздувается протухшими строками. У yandex ttl_days_effective держали
+75/75/75/39/52/54 шесть прогонов подряд при deactivated=0 -- это и есть разгон пола,
+ради которого потолок написан. Цифру «23 687 из 44 744» из исходного разбора сюда НЕ
+переносим: она про все источники сразу, две трети её -- новостройки вне выборки
+оценщика, а у самого avito просроченных строк ноль (уточнено 2026-08-15).
+
+Этот файл проверяет CAP_MULT -- потолок, не пускающий эффективный TTL выше
+ttl_days * CAP_MULT, независимо от того, насколько высоко посчитанный пол.
+"""
+
+from __future__ import annotations
+
+import os
+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
+
+# ── Фейковая сессия (тот же контракт, что в test_deactivate_stale_revisit_floor.py) ──
+
+
+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 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", "avito"),
+        ttl_days=kwargs.pop("ttl_days", 10),
+        **kwargs,
+    )
+
+
+# ── Контракт из задачи ─────────────────────────────────────────────────────────
+
+
+def test_high_floor_is_capped_at_double_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
+    """revisit_floor=75, ttl_days=10 -> итог 20 (потолок 2x), НЕ 75."""
+    db = _FakeDB(floor_days=75.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 20
+    assert out["ttl_floor_capped"] == 1
+    assert out["ttl_days_floor_raw"] == 75
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 20, "UPDATE обязан получить капнутый TTL, не сырой пол"
+
+
+def test_low_floor_leaves_ttl_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    """revisit_floor=5, ttl_days=10 -> итог 10 (пол ниже заданного TTL, max() его не поднимает)."""
+    db = _FakeDB(floor_days=5.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10
+    assert "ttl_floor_capped" not in out
+    assert "ttl_days_floor_raw" not in out
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 10
+
+
+# ── Контракт потолка ────────────────────────────────────────────────────────────
+
+
+def test_cap_mult_is_named_module_constant_equal_two() -> None:
+    assert task_mod.CAP_MULT == 2
+
+
+def test_floor_between_ttl_and_cap_is_not_flagged_capped(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пол поднял TTL, но не дотянулся до потолка -- capped-флаг НЕ выставляется."""
+    db = _FakeDB(floor_days=15.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 15
+    assert "ttl_floor_capped" not in out
+
+
+def test_floor_exactly_at_cap_boundary_is_not_flagged_capped(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Пол ровно на потолке (2x ttl) -- это ещё "поднят до потолка", не "срезан выше него".
+
+    Формула -- min(raw, cap): при raw == cap срезания не происходит (raw > cap ложно),
+    капнутый флаг предназначен сигналить именно "потолок реально что-то отрезал".
+    """
+    db = _FakeDB(floor_days=20.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 20
+    assert "ttl_floor_capped" not in out
+
+
+def test_cap_logs_warning_containing_both_numbers(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """WARNING при срезании содержит и сырой пол, и капнутый результат -- не только counters."""
+    db = _FakeDB(floor_days=75.0)
+    with caplog.at_level("WARNING", logger=task_mod.logger.name):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    messages = " ".join(r.getMessage() for r in caplog.records)
+    assert "75" in messages, "лог обязан называть сырой пол"
+    assert "20" in messages, "лог обязан называть итоговый (капнутый) TTL"
+
+
+def test_cap_never_lowers_ttl_below_configured_value(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Потолок -- верхняя граница, не альтернативный источник истины: заданный TTL
+    (10) остаётся нижней границей независимо от того, насколько низко ушёл пол."""
+    db = _FakeDB(floor_days=1.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10
+
+
+# ── avito self-descend (52 -> ... -> 10) не должен ломаться потолком ────────────
+
+
+def test_avito_high_transient_floor_is_capped_not_left_unbounded(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Наблюдённый на проде транзиентный пик avito (счётчики видели ttl_days_effective=52)
+    теперь капается на 2x ttl=20, а не пропускается в UPDATE как есть."""
+    db = _FakeDB(floor_days=52.0)
+    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 20
+    assert out["ttl_floor_capped"] == 1
+    assert out["ttl_days_floor_raw"] == 52
+
+
+def test_avito_recovered_low_floor_still_reaches_configured_ttl(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """После восстановления обхода (пол опустился ниже ttl_days=10, как на проде 52->10)
+    потолок не мешает нормальному пути -- эффективный TTL просто равен заданному."""
+    db = _FakeDB(floor_days=9.0)
+    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10
+    assert "ttl_floor_capped" not in out
+
+
+def test_avito_floor_above_ttl_but_under_cap_passes_through_uncapped(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Промежуточная точка того же самопонижения (пол между ttl и потолком, например 18)
+    поднимает TTL как раньше -- потолок не мешает нормальному постепенному пути."""
+    db = _FakeDB(floor_days=18.0)
+    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 18
+    assert "ttl_floor_capped" not in out
+
+
+# ── cap_mult конфигурируем per-source (найдено ревью 2026-08-15) ────────────────
+# Дефолтный CAP_MULT=2 даёт разный АБСОЛЮТНЫЙ потолок на разных источниках
+# (cian/yandex 60 сут, avito 20 сут), а хвост переобхода не пропорционален
+# ttl_days: avito p99=42.1 -- выше его же дефолтного потолка 20. cap_mult -- ручка
+# для конкретно такого источника, без изменения дефолта для остальных.
+
+
+def test_cap_mult_defaults_to_module_constant_when_not_overridden(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Без явного cap_mult поведение не меняется: потолок = ttl_days * CAP_MULT (2)."""
+    db = _FakeDB(floor_days=75.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10 * task_mod.CAP_MULT
+
+
+def test_cap_mult_override_raises_the_ceiling_for_a_long_tailed_source(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """avito p99=42.1: cap_mult=6 (потолок 60) больше не режет пол ниже хвоста обхода,
+    в отличие от дефолтного cap_mult=2 (потолок 20)."""
+    db = _FakeDB(floor_days=45.0)
+    out = _run(
+        db,
+        monkeypatch,
+        listing_source="avito",
+        ttl_days=10,
+        revisit_floor_quantile=0.99,
+        cap_mult=6,
+    )
+    assert out["ttl_days_effective"] == 45
+    assert "ttl_floor_capped" not in out
+
+
+def test_cap_mult_override_still_caps_when_floor_exceeds_the_wider_ceiling(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """cap_mult поднимает потолок, но не убирает его -- пол выше 60 всё равно срезается."""
+    db = _FakeDB(floor_days=90.0)
+    out = _run(
+        db,
+        monkeypatch,
+        listing_source="avito",
+        ttl_days=10,
+        revisit_floor_quantile=0.99,
+        cap_mult=6,
+    )
+    assert out["ttl_days_effective"] == 60
+    assert out["ttl_floor_capped"] == 1
+    assert out["ttl_days_floor_raw"] == 90
+
+
+def test_cap_mult_is_threaded_into_update_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Капнутый по override'нутому потолку TTL реально уходит в UPDATE, не только считается."""
+    db = _FakeDB(floor_days=90.0)
+    _run(
+        db,
+        monkeypatch,
+        listing_source="avito",
+        ttl_days=10,
+        revisit_floor_quantile=0.99,
+        cap_mult=6,
+    )
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 60
+
+
+# ── ttl_days <= 0 (LOW из ревью 2026-08-15) ──────────────────────────────────────
+# До потолка max(ttl_days, floor) прикрывал ttl_days<=0, если пол посчитан и
+# положителен. С потолком min(raw, ttl_days * cap_mult) при ttl_days<=0 капнутый
+# потолок тоже <= 0 и побеждает в min() -- защита пола пропадает молча. Явный guard
+# ловит это ДО любого SQL, тем же путём, что и невалидный staleness_column.
+
+
+def test_ttl_days_zero_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    db = _FakeDB(floor_days=75.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=0, revisit_floor_quantile=0.99)
+    assert db.executed == []
+
+
+def test_ttl_days_negative_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    db = _FakeDB(floor_days=75.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=-5, revisit_floor_quantile=0.99)
+    assert db.executed == []
+
+
+def test_ttl_days_zero_fails_the_run_via_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Тот же контракт, что и невалидный staleness_column: run помечается failed,
+    а не остаётся 'running'."""
+    marked_failed: list[Any] = []
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_failed",
+        lambda db, run_id, err, counters: marked_failed.append((run_id, err, counters)),
+    )
+    db = _FakeDB(floor_days=75.0)
+    with pytest.raises(ValueError):
+        task_mod.deactivate_stale_listings(
+            db,  # type: ignore[arg-type]
+            7,
+            listing_source="avito",
+            ttl_days=0,
+        )
+    assert len(marked_failed) == 1
+    assert marked_failed[0][0] == 7
+
+
+# ── cap_mult < 1 (HIGH из ревью круга 2, 2026-08-15) ─────────────────────────────
+# Тот же класс дыры, что и ttl_days<=0 выше, но со стороны потолка: cap_mult -- ЕДИНСТВЕННЫЙ
+# запланированный способ его задать -- руками вписать в jsonb default_params расписания
+# (см. миграцию для avito), т.е. именно там опечатка 0 / 0.5 вместо 6 доходит до прода.
+# cap_mult=0 -> capped=0 -> effective_ttl_days=0 -> UPDATE снимает весь активный пул
+# источника молча. cap_mult<1 (например 0.5) опускает потолок НИЖЕ заданного оператором
+# ttl_days -- прямое нарушение инварианта, который проверяет
+# test_cap_never_lowers_ttl_below_configured_value для пола, но не было проверено для
+# потолка при некорректном cap_mult.
+
+
+def test_cap_mult_zero_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=0)
+    assert db.executed == []
+
+
+def test_cap_mult_negative_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=-2)
+    assert db.executed == []
+
+
+def test_cap_mult_below_one_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    """cap_mult=0.5 опустил бы потолок НИЖЕ заданного ttl_days -- та самая инверсия,
+    которую тест test_cap_never_lowers_ttl_below_configured_value гарантирует для пола."""
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=0.5)
+    assert db.executed == []
+
+
+def test_cap_mult_non_numeric_fails_safe_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Опечатка в jsonb default_params (строка вместо числа) не должна молча пройти
+    в SQL -- TypeError из сравнения `cap_mult < 1` ловится тем же except Exception,
+    что и ValueError-гварды, и маршрутизируется через mark_failed. Никакого SQL не
+    исполняется, ни один active-лот не тронут."""
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(TypeError):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult="6")
+    assert db.executed == []
+
+
+# ── cap_mult / ttl_days -- bool (найдено ревью круга 3, 2026-08-15) ─────────────
+# bool -- подкласс int в Python: `True < 1` ложно, `True <= 0` ложно. Числовые
+# guard'ы выше (`cap_mult < 1`, `ttl_days <= 0`) поэтому НЕ ловят jsonb `true` в
+# default_params расписания -- ровно тот класс опечатки, ради которого guard'ы
+# вообще написаны. `cap_mult=True` даёт потолок == ttl_days (ttl_days * True ==
+# ttl_days) -- пол молча отключается без единого ValueError. `ttl_days=True` даёт
+# ttl_days == 1 -- TTL молча меняется на 1 сутки. Явная type-проверка ловит оба
+# ДО числового сравнения и ДО любого SQL.
+
+
+def test_cap_mult_true_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    """cap_mult=True: `True < 1` ложно -- без явной type-проверки потолок = ttl_days
+    (пол молча отключается) вместо ValueError. Воспроизведено на HEAD ветки."""
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=True)
+    assert db.executed == []
+
+
+def test_cap_mult_false_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    """cap_mult=False уже ловится `cap_mult < 1` (False == 0), но type-guard идёт
+    первым -- проверяем, что путь всё равно ValueError, а не иной exception."""
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99, cap_mult=False)
+    assert db.executed == []
+
+
+def test_ttl_days_true_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    """ttl_days=True: `True <= 0` ложно -- без явной type-проверки TTL молча
+    становится 1 сутки (True ведёт себя как int 1) вместо ValueError."""
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=True, revisit_floor_quantile=0.99)
+    assert db.executed == []
+
+
+def test_ttl_days_false_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    """ttl_days=False уже ловится `ttl_days <= 0` (False == 0), но type-guard идёт
+    первым -- проверяем, что путь всё равно ValueError."""
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        _run(db, monkeypatch, ttl_days=False, revisit_floor_quantile=0.99)
+    assert db.executed == []
+
+
+def test_cap_mult_zero_fails_the_run_via_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Тот же контракт, что и ttl_days<=0: run помечается failed, а не остаётся 'running',
+    и НИ ОДНА строка не деактивируется (в отличие от воспроизведённого на HEAD дефекта, где
+    cap_mult=0 давало effective_ttl_days=0 и снимало весь активный пул источника)."""
+    marked_failed: list[Any] = []
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_failed",
+        lambda db, run_id, err, counters: marked_failed.append((run_id, err, counters)),
+    )
+    db = _FakeDB(floor_days=52.0)
+    with pytest.raises(ValueError):
+        task_mod.deactivate_stale_listings(
+            db,  # type: ignore[arg-type]
+            9,
+            listing_source="avito",
+            ttl_days=10,
+            revisit_floor_quantile=0.99,
+            cap_mult=0,
+        )
+    assert len(marked_failed) == 1
+    assert marked_failed[0][0] == 9
+    assert db.executed == []
+
+
+# ── проводка cap_mult в product_handlers ─────────────────────────────────────────
+
+
+def test_handler_wires_cap_mult_from_schedule_params() -> None:
+    """Тот же приём, что test_handler_wires_revisit_floor_from_schedule_params:
+    читаем исходник файлом (product_handlers тянет scraper_kit, которого в юнит-
+    окружении может не быть) и проверяем именно проводку default_params -> вызов."""
+    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("cap_mult", CAP_MULT)' in flat
+    assert "cap_mult=cap_mult" in job