feat(scrapers): segment-aware generic deactivate_stale for yandex/cian

Generalize the avito-only stale-listing deactivation into a per-source,
segment-aware task and enable it for yandex/cian vtorichka.

Why segment-aware (not blanket TTL like avito): yandex/cian city sweeps do
not maintain full inventory coverage (cian_city_sweep disabled, yandex narrow
5-anchor nightly sweep refreshes ~48/day of 3969 active). A blanket TTL would
deactivate live inventory — including 9659 active first-party novostroyki —
because 'not seen in N days' means 'not re-covered', not 'sold'. So yandex/cian
deactivate ONLY listing_segment='vtorichka' at TTL=30; novostroyki and
NULL-segment are protected. avito unchanged (all segments, TTL=10).

- deactivate_stale_listings(db, run_id, *, listing_source, ttl_days, segments)
  generic fn; deactivate_stale_avito_listings kept as back-compat wrapper.
- scheduler dispatch: source.startswith('deactivate_stale_') -> generic trigger
  reading listing_source/ttl_days/segments from default_params jsonb.
- 114 seed: deactivate_stale_yandex + deactivate_stale_cian (vtorichka, TTL=30).
- segments is None => all segments; [] => none (ANY(ARRAY[]) matches nothing).

Refs #759

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

@@ -23,10 +23,13 @@ Sources:
                               yandex detail pages, extracts full address from <title>,
                               updates listings.address for street-only addresses;
                               nightly 02:00-03:00 UTC, before refresh_search_matview)
-  - deactivate_stale_avito  → deactivate_stale_avito_listings
-                              (tasks/deactivate_stale_avito.py, #759; marks avito listings
-                              with last_seen_at > TTL days as is_active=false — no DELETE,
-                              no external calls, enabled by default; window after avito sweep)
+  - deactivate_stale_*      → trigger_deactivate_stale_run → deactivate_stale_listings
+                              (tasks/deactivate_stale_avito.py, #759 + segment-aware
+                              generalisation; marks listings with last_seen_at > TTL days as
+                              is_active=false — no DELETE, no external calls. Source prefix
+                              dispatch covers deactivate_stale_avito (all segments, TTL=10) +
+                              deactivate_stale_yandex/cian (vtorichka only, TTL=30, segments
+                              from default_params). Window after the source sweep.)
   - sber_index_pull         → run_sber_index_pull
                               (tasks/sber_index_pull.py, #887; monthly pull of СберИндекс
                               city-level secondary-housing price index — pure HTTP+DB, no
@@ -708,6 +711,59 @@ async def trigger_deactivate_stale_avito_run(
     return run_id
 
 
+async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Универсальный триггер для всех deactivate_stale_* источников.
+
+    Читает listing_source, ttl_days, segments из schedule_row["default_params"] (jsonb).
+    Для источника 'deactivate_stale_avito' с пустым default_params использует defaults:
+      listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None.
+    Для yandex/cian -- default_params обязан содержать listing_source + ttl_days + segments.
+
+    Запускает deactivate_stale_listings в executor (sync DB-only task).
+    Возвращает run_id или None (skip -- already running).
+    """
+    from app.core.config import settings as _settings
+
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+    listing_source: str = params.get("listing_source", "avito")
+    ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days)
+    segments: list[str] | None = params.get("segments")  # list or None
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.tasks.deactivate_stale_avito import deactivate_stale_listings
+
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(
+                None,
+                lambda: deactivate_stale_listings(
+                    run_db,
+                    run_id,
+                    listing_source=listing_source,
+                    ttl_days=ttl_days,
+                    segments=segments,
+                ),
+            )
+        except Exception:
+            logger.exception(
+                "scheduler: deactivate_stale source=%s crashed run_id=%d",
+                listing_source,
+                run_id,
+            )
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered deactivate_stale source=%s run_id=%d", listing_source, run_id)
+    return run_id
+
+
 async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch run_sber_index_pull в asyncio.create_task.
 
@@ -1268,8 +1324,8 @@ async def scheduler_loop() -> None:
                         await trigger_refresh_search_matview_run(db, sch)
                     elif source == "yandex_address_backfill":
                         await trigger_yandex_address_backfill_run(db, sch)
-                    elif source == "deactivate_stale_avito":
-                        await trigger_deactivate_stale_avito_run(db, sch)
+                    elif source.startswith("deactivate_stale_"):
+                        await trigger_deactivate_stale_run(db, sch)
                     elif source == "sber_index_pull":
                         await trigger_sber_index_pull_run(db, sch)
                     elif source == "rosreestr_quarter_poll":

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

@@ -1,22 +1,27 @@
-"""Deactivate stale avito listings that have not been seen in >TTL days (#759).
+"""Deactivate stale listings that have not been seen in >TTL days.
 
-Нечасто обновляемые объявления на Avito (last_seen_at < NOW() - TTL) помечаются
-is_active=false, чтобы coverage/estimator denominators не раздувались мёртвыми
-объявлениями. Строки НЕ удаляются — история нужна для бэктеста (#667).
+Первоначально создано для avito (#759). Теперь поддерживает любой источник
+(listing_source) с опциональной фильтрацией по listing_segment.
 
-Только source='avito'. Cian/Yandex не затрагиваются.
+Ключевые решения:
+- Cian/Yandex не поддерживают full-coverage sweep -> паушальный TTL сломает живой
+  инвентарь. DECISION: для yandex/cian деактивировать ТОЛЬКО listing_segment='vtorichka',
+  TTL=30. novostroyki (9659 активных первичных строк) и NULL-сегмент не трогаем.
+- avito: все сегменты (segments=None), TTL=10 дней -- поведение без изменений.
+- Строки НЕ удаляются -- история нужна для бэктеста (#667).
 
-Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app
-scheduler'ом через trigger_deactivate_stale_avito_run() (scheduler.py),
+Задача синхронная (DB-only, никаких внешних HTTP-вызовов) -- запускается in-app
+scheduler'ом через trigger_deactivate_stale_run() (scheduler.py),
 по образцу snapshot_listing_sources / recompute_asking_to_sold_ratios
 (sync task в run_in_executor).
 
-TTL берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10).
+TTL для avito берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10).
 """
 
 from __future__ import annotations
 
 import logging
+from typing import Any
 
 from sqlalchemy import text
 from sqlalchemy.orm import Session
@@ -26,45 +31,107 @@ from app.services import scrape_runs as runs_mod
 
 logger = logging.getLogger(__name__)
 
-# UPDATE: помечаем is_active=false для avito-объявлений, не виденных >TTL дней.
-# CAST(:ttl_days || ' days' AS interval) — psycopg v3 safe (никаких :param::type).
-# Только source='avito' и is_active=true — чтобы не трогать уже неактивные строки
-# и не затрагивать cian/yandex.
-_DEACTIVATE_SQL = text(
+# Базовый UPDATE без фильтра по сегменту: все сегменты для данного source.
+# CAST(:ttl_days || ' days' AS interval) -- psycopg v3 safe (никаких :param::type).
+_DEACTIVATE_SQL_ALL_SEGMENTS = text(
     """
     UPDATE listings
     SET is_active = false
-    WHERE source = 'avito'
+    WHERE source = :listing_source
       AND is_active = true
       AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
     """
 )
 
+# UPDATE с фильтром по сегменту (segments задан): затрагивает только указанные сегменты.
+# = ANY(CAST(:segments AS text[])) -- psycopg v3 адаптирует Python list -> text[].
+_DEACTIVATE_SQL_SEGMENTS = text(
+    """
+    UPDATE listings
+    SET is_active = false
+    WHERE source = :listing_source
+      AND is_active = true
+      AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval)
+      AND listing_segment = ANY(CAST(:segments AS text[]))
+    """
+)
 
-def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]:
-    """Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад.
+# Алиас для обратной совместимости -- использовался в тестах через task_mod._DEACTIVATE_SQL
+_DEACTIVATE_SQL = _DEACTIVATE_SQL_ALL_SEGMENTS
+
+
+def deactivate_stale_listings(
+    db: Session,
+    run_id: int,
+    *,
+    listing_source: str,
+    ttl_days: int,
+    segments: list[str] | None = None,
+) -> dict[str, int]:
+    """Пометить is_active=false объявления с last_seen_at > ttl_days дней назад.
+
+    Параметры:
+        listing_source: значение поля source в таблице listings ('avito', 'yandex', 'cian').
+        ttl_days: количество дней TTL; объявления старше этого порога деактивируются.
+        segments: если задан -- деактивировать только объявления с указанными
+            listing_segment значениями. None -> все сегменты (поведение avito по умолчанию).
 
     Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
     Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed).
 
-    Returns {"deactivated": N} — количество обновлённых строк.
+    Returns {"deactivated": N} -- количество обновлённых строк.
     """
     counters: dict[str, int] = {"deactivated": 0}
     try:
-        result = db.execute(_DEACTIVATE_SQL, {"ttl_days": settings.avito_stale_ttl_days})
+        # segments is None -> все сегменты (поведение avito). segments=[...] -> только
+        # перечисленные сегменты. Используем `is not None` (НЕ truthy): пустой список []
+        # означает "ни один сегмент" (= ANY(ARRAY[]) ничего не матчит, деактивирует 0),
+        # а НЕ "все сегменты" — иначе случайный [] стёр бы весь источник.
+        if segments is not None:
+            params: dict[str, Any] = {
+                "listing_source": listing_source,
+                "ttl_days": ttl_days,
+                "segments": segments,
+            }
+            result = db.execute(_DEACTIVATE_SQL_SEGMENTS, params)
+        else:
+            params = {"listing_source": listing_source, "ttl_days": ttl_days}
+            result = db.execute(_DEACTIVATE_SQL_ALL_SEGMENTS, params)
+
         counters["deactivated"] = result.rowcount or 0
 
         db.commit()
         runs_mod.mark_done(db, run_id, counters)
         logger.info(
-            "deactivate_stale_avito run_id=%d done: deactivated=%d (ttl_days=%d)",
+            "deactivate_stale source=%s run_id=%d done: deactivated=%d "
+            "(ttl_days=%d, segments=%r)",
+            listing_source,
             run_id,
             counters["deactivated"],
-            settings.avito_stale_ttl_days,
+            ttl_days,
+            segments,
         )
         return counters
     except Exception as exc:
-        logger.exception("deactivate_stale_avito run_id=%d failed", run_id)
+        logger.exception("deactivate_stale source=%s run_id=%d failed", listing_source, run_id)
         db.rollback()
         runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
         raise
+
+
+def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]:
+    """Обёртка для обратной совместимости -- deactivate_stale_avito (все сегменты, TTL из settings).
+
+    Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад.
+    Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
+    Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed).
+
+    Returns {"deactivated": N} -- количество обновлённых строк.
+    """
+    return deactivate_stale_listings(
+        db,
+        run_id,
+        listing_source="avito",
+        ttl_days=settings.avito_stale_ttl_days,
+        segments=None,
+    )

tradein-mvp/backend/data/sql/114_scrape_schedules_seed_deactivate_stale_yandex_cian.sql

@@ -0,0 +1,68 @@
+-- 114_scrape_schedules_seed_deactivate_stale_yandex_cian.sql
+-- Seed scrape_schedules rows for nightly segment-aware stale-listing deactivation
+-- of yandex and cian vtorichka (#759 generalisation).
+--
+-- Rationale (segment-aware vs blanket TTL):
+--   Yandex and Cian city sweeps do NOT maintain full inventory coverage -- they
+--   sample a rolling window rather than crawling every live listing on every run.
+--   A blanket TTL (as used for Avito) would therefore wrongly deactivate live
+--   novostroyki and unknown-segment (NULL) rows that simply were not re-visited
+--   in the sample window.
+--
+--   DECISION (user-approved): deactivate ONLY listing_segment='vtorichka' rows,
+--   with TTL=30 days.  This is conservative enough that at first deploy it will
+--   touch 0 rows (no vtorichka listing has been absent for 30 days yet), but will
+--   catch genuinely-stale secondhand inventory going forward.
+--
+--   novostroyki rows (9659 active first-party rows as of 2026-06-16) are protected:
+--   they are managed by dedicated newbuilding pipelines and must not be deactivated
+--   by this task.  NULL-segment rows are also excluded as their segment is unknown.
+--
+-- Schedule window 07:00-08:00 UTC:
+--   After avito deactivate_stale (06:00-07:00 UTC) and after the main city sweeps
+--   (02:00-05:00 UTC) and Rosreestr OS-cron (05:30/05:45 UTC).
+--   Pattern: next_run_at bootstrapped to tomorrow 07:00 UTC so the scheduler does
+--   not fire immediately on deploy.
+--
+-- Idempotent: ON CONFLICT (source) DO NOTHING -- safe to re-apply.
+--
+-- Dependencies:
+--   052_scrape_schedules.sql (table + UNIQUE(source)).
+--   listings table with listing_segment column.
+--   app/tasks/deactivate_stale_avito.py -- deactivate_stale_listings() (generic fn).
+--   app/services/scheduler.py -- trigger_deactivate_stale_run() + dispatch branch
+--     source.startswith("deactivate_stale_").
+--
+-- Deploy order:
+--   Apply after deploying the scheduler.py + task changes.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'deactivate_stale_yandex',
+    true,
+    7,
+    8,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
+    '{"listing_source":"yandex","ttl_days":30,"segments":["vtorichka"]}'::jsonb
+),
+(
+    'deactivate_stale_cian',
+    true,
+    7,
+    8,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 7)) AT TIME ZONE 'UTC',
+    '{"listing_source":"cian","ttl_days":30,"segments":["vtorichka"]}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMIT;

tradein-mvp/backend/tests/test_deactivate_stale_avito.py

@@ -25,14 +25,18 @@ _MIGRATION_090 = _SQL_DIR / "090_scrape_schedules_seed_deactivate_stale_avito.sq
 
 _DEACTIVATE_SQL = str(task_mod._DEACTIVATE_SQL.text)
 _TASK_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings)
+_GENERIC_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_listings)
 
 
 # ── SQL shape ─────────────────────────────────────────────────────────────────
 
 
 def test_sql_targets_only_avito_source() -> None:
-    """UPDATE must filter source='avito' — cian/yandex rows must not be touched."""
-    assert "source = 'avito'" in _DEACTIVATE_SQL
+    """Compat wrapper enforces source='avito' via listing_source param — cian/yandex not touched."""
+    # _DEACTIVATE_SQL is now the generic all-segments SQL using :listing_source bind param.
+    assert ":listing_source" in _DEACTIVATE_SQL
+    # The avito constraint is enforced by the compat wrapper passing listing_source="avito".
+    assert 'listing_source="avito"' in _TASK_SRC
 
 
 def test_sql_only_deactivates_not_deletes() -> None:
@@ -83,17 +87,20 @@ def test_task_reads_ttl_from_settings() -> None:
 
 
 def test_task_returns_deactivated_counter() -> None:
-    assert '"deactivated"' in _TASK_SRC or "'deactivated'" in _TASK_SRC
+    # counter logic lives in the generic deactivate_stale_listings function
+    assert '"deactivated"' in _GENERIC_TASK_SRC or "'deactivated'" in _GENERIC_TASK_SRC
 
 
 def test_task_finalises_run() -> None:
-    assert "mark_done" in _TASK_SRC
-    assert "mark_failed" in _TASK_SRC
+    # mark_done/mark_failed live in the generic deactivate_stale_listings function
+    assert "mark_done" in _GENERIC_TASK_SRC
+    assert "mark_failed" in _GENERIC_TASK_SRC
 
 
 def test_task_commits_and_rollbacks_on_error() -> None:
-    assert "db.commit()" in _TASK_SRC
-    assert "db.rollback()" in _TASK_SRC
+    # commit/rollback live in the generic deactivate_stale_listings function
+    assert "db.commit()" in _GENERIC_TASK_SRC
+    assert "db.rollback()" in _GENERIC_TASK_SRC
 
 
 # ── Scheduler wiring ──────────────────────────────────────────────────────────
@@ -106,10 +113,10 @@ def test_trigger_fn_exists_and_is_async() -> None:
 
 
 def test_dispatch_branch_wired() -> None:
-    """scheduler_loop dispatches source='deactivate_stale_avito' to the trigger."""
+    """scheduler_loop dispatches deactivate_stale_* sources to the generic trigger."""
     loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "deactivate_stale_avito"' in loop_src
-    assert "trigger_deactivate_stale_avito_run(db, sch)" in loop_src
+    assert 'source.startswith("deactivate_stale_")' in loop_src
+    assert "trigger_deactivate_stale_run(db, sch)" in loop_src
 
 
 def test_trigger_runs_sync_task_in_executor() -> None:

tradein-mvp/backend/tests/test_deactivate_stale_listings.py

@@ -0,0 +1,305 @@
+from __future__ import annotations
+
+import inspect
+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.services import scheduler
+from app.tasks import deactivate_stale_avito as task_mod
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_114 = _SQL_DIR / "114_scrape_schedules_seed_deactivate_stale_yandex_cian.sql"
+
+_GENERIC_FN_SRC = inspect.getsource(task_mod.deactivate_stale_listings)
+_COMPAT_FN_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings)
+
+
+def test_segment_sql_uses_cast_interval_not_double_colon() -> None:
+    sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
+    assert "CAST(:ttl_days || ' days' AS interval)" in sql
+    assert not re.search(r"\w+::", sql)
+
+
+def test_segment_sql_uses_any_cast_text_array() -> None:
+    sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
+    assert "ANY(CAST(:segments AS text[]))" in sql
+
+
+def test_segment_sql_filters_is_active_true() -> None:
+    sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
+    assert "is_active = true" in sql
+
+
+def test_segment_sql_only_deactivates_not_deletes() -> None:
+    sql = task_mod._DEACTIVATE_SQL_SEGMENTS.text
+    assert "SET is_active = false" in sql
+    assert "DELETE" not in sql.upper()
+
+
+def test_all_segments_sql_uses_listing_source_param() -> None:
+    sql = task_mod._DEACTIVATE_SQL_ALL_SEGMENTS.text
+    assert ":listing_source" in sql
+    assert "source = 'avito'" not in sql
+
+
+def test_deactivate_sql_alias_is_all_segments() -> None:
+    assert task_mod._DEACTIVATE_SQL is task_mod._DEACTIVATE_SQL_ALL_SEGMENTS
+
+
+def test_generic_fn_exists() -> None:
+    assert callable(task_mod.deactivate_stale_listings)
+
+
+def test_generic_fn_returns_deactivated_counter() -> None:
+    assert "deactivated" in _GENERIC_FN_SRC
+
+
+def test_generic_fn_finalises_run() -> None:
+    assert "mark_done" in _GENERIC_FN_SRC
+    assert "mark_failed" in _GENERIC_FN_SRC
+
+
+def test_generic_fn_commits_and_rollbacks() -> None:
+    assert "db.commit()" in _GENERIC_FN_SRC
+    assert "db.rollback()" in _GENERIC_FN_SRC
+
+
+def test_compat_wrapper_calls_generic_with_avito_defaults() -> None:
+    assert "deactivate_stale_listings(" in _COMPAT_FN_SRC
+    assert 'listing_source="avito"' in _COMPAT_FN_SRC
+    assert "settings.avito_stale_ttl_days" in _COMPAT_FN_SRC
+    assert "segments=None" in _COMPAT_FN_SRC
+
+
+def test_trigger_deactivate_stale_run_exists_and_is_async() -> None:
+    fn = getattr(scheduler, "trigger_deactivate_stale_run", None)
+    assert fn is not None, "trigger_deactivate_stale_run missing from scheduler"
+    assert inspect.iscoroutinefunction(fn)
+
+
+def test_dispatch_branch_uses_startswith() -> None:
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source.startswith("deactivate_stale_")' in loop_src
+    assert "trigger_deactivate_stale_run(db, sch)" in loop_src
+
+
+def test_trigger_runs_sync_task_in_executor() -> None:
+    trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run)
+    assert "run_in_executor(" in trig_src
+    assert "deactivate_stale_listings" in trig_src
+    assert "_claim_run(db, schedule_row)" in trig_src
+
+
+def test_trigger_reads_params_from_default_params() -> None:
+    trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run)
+    assert "listing_source" in trig_src
+    assert "ttl_days" in trig_src
+    assert "segments" in trig_src
+
+
+class _FakeResult:
+    def __init__(self, rowcount: int) -> None:
+        self.rowcount = rowcount
+
+
+class _FakeDB:
+    def __init__(self, rowcount: int = 0) -> None:
+        self._rowcount = rowcount
+        self.executed: list[tuple[Any, Any]] = []
+        self.committed = False
+        self.rolled_back = False
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        self.executed.append((stmt, params))
+        return _FakeResult(self._rowcount)
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+
+def test_vtorichka_stale_beyond_ttl_deactivated(monkeypatch: pytest.MonkeyPatch) -> None:
+    marked: dict[str, Any] = {}
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=5)
+    out = task_mod.deactivate_stale_listings(
+        db, run_id=1, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
+    )  # type: ignore[arg-type]
+    assert out == {"deactivated": 5}
+    assert db.committed is True
+    _stmt, params = db.executed[0]
+    assert params is not None
+    assert params["segments"] == ["vtorichka"]
+    assert params["listing_source"] == "yandex"
+    assert params["ttl_days"] == 30
+    assert marked["counters"] == {"deactivated": 5}
+
+
+def test_novostroyki_not_in_segment_filter(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=0)
+    task_mod.deactivate_stale_listings(
+        db, run_id=2, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
+    )  # type: ignore[arg-type]
+    _stmt, params = db.executed[0]
+    assert params is not None
+    assert "segments" in params
+    assert "novostroyki" not in params["segments"]
+
+
+def test_null_segment_not_in_filter(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=0)
+    task_mod.deactivate_stale_listings(
+        db, run_id=3, listing_source="cian", ttl_days=30, segments=["vtorichka"]
+    )  # type: ignore[arg-type]
+    _stmt, params = db.executed[0]
+    assert params is not None
+    assert None not in params["segments"]
+
+
+def test_other_source_listing_source_param_matches(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=0)
+    task_mod.deactivate_stale_listings(
+        db, run_id=4, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
+    )  # type: ignore[arg-type]
+    _stmt, params = db.executed[0]
+    assert params is not None
+    assert params["listing_source"] == "yandex"
+
+
+def test_fresh_listing_zero_rowcount(monkeypatch: pytest.MonkeyPatch) -> None:
+    marked: dict[str, Any] = {}
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=0)
+    out = task_mod.deactivate_stale_listings(
+        db, run_id=5, listing_source="cian", ttl_days=30, segments=["vtorichka"]
+    )  # type: ignore[arg-type]
+    assert out == {"deactivated": 0}
+    assert db.committed is True
+    assert marked["counters"] == {"deactivated": 0}
+
+
+def test_segments_none_uses_all_segments_sql(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=3)
+    out = task_mod.deactivate_stale_listings(
+        db, run_id=6, listing_source="avito", ttl_days=10, segments=None
+    )  # type: ignore[arg-type]
+    assert out == {"deactivated": 3}
+    _stmt, params = db.executed[0]
+    assert params is not None
+    assert "segments" not in params
+    assert params["listing_source"] == "avito"
+    assert params["ttl_days"] == 10
+
+
+def test_avito_compat_wrapper_counter(monkeypatch: pytest.MonkeyPatch) -> None:
+    marked: dict[str, Any] = {}
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    db = _FakeDB(rowcount=42)
+    out = task_mod.deactivate_stale_avito_listings(db, run_id=7)  # type: ignore[arg-type]
+    assert out == {"deactivated": 42}
+    assert db.committed is True
+    _stmt, params = db.executed[0]
+    assert params is not None
+    assert params["listing_source"] == "avito"
+    assert "segments" not in params
+    assert marked["run_id"] == 7
+
+
+def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    failed: dict[str, 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: failed.update(run_id=run_id, err=err),
+    )
+
+    class _BoomDB(_FakeDB):
+        def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+            raise RuntimeError("db exploded")
+
+    db = _BoomDB()
+    with pytest.raises(RuntimeError, match="db exploded"):
+        task_mod.deactivate_stale_listings(
+            db, run_id=8, listing_source="yandex", ttl_days=30, segments=["vtorichka"]
+        )  # type: ignore[arg-type]
+    assert db.rolled_back is True
+    assert failed["run_id"] == 8
+
+
+def test_migration_114_exists() -> None:
+    assert _MIGRATION_114.is_file(), f"missing migration: {_MIGRATION_114}"
+
+
+def test_migration_114_seeds_yandex_and_cian() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert "'deactivate_stale_yandex'" in sql
+    assert "'deactivate_stale_cian'" in sql
+
+
+def test_migration_114_is_idempotent() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+
+
+def test_migration_114_enabled_true() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert "true" in sql
+
+
+def test_migration_114_is_transactional() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_114_window_7_to_8_utc() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert re.search(r"\b7\b", sql), "window_start_hour 7 missing"
+    assert re.search(r"\b8\b", sql), "window_end_hour 8 missing"
+
+
+def test_migration_114_segments_vtorichka_in_default_params() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert "vtorichka" in sql
+
+
+def test_migration_114_ttl_30_days() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert "30" in sql
+
+
+def test_migration_114_no_psycopg_trap() -> None:
+    sql = _MIGRATION_114.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)