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

@@ -183,6 +183,16 @@ async def _job_rosreestr_quarter_poll(
     await run_rosreestr_quarter_poll(db, run_id=run_id, params=params)
 
 
+# ── deals_freshness_monitor — sync DB-only freshness check в executor ─────────
+async def _job_deals_freshness_monitor(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.deals_freshness_monitor import check_deals_freshness
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, check_deals_freshness, db, run_id, params)
+
+
 # ── newbuilding_enrich — async, owns lifecycle ───────────────────────────────
 async def _job_newbuilding_enrich(
     db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@@ -348,6 +358,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
         "yandex_address_backfill": Handler(_job_yandex_address_backfill, "yandex_address_backfill"),
         "sber_index_pull": Handler(_job_sber_index_pull, "sber_index_pull"),
         "rosreestr_quarter_poll": Handler(_job_rosreestr_quarter_poll, "rosreestr_quarter_poll"),
+        "deals_freshness_monitor": Handler(_job_deals_freshness_monitor, "deals_freshness_monitor"),
         "newbuilding_enrich": Handler(_job_newbuilding_enrich, "newbuilding_enrich"),
         "yandex_newbuilding_sweep": Handler(
             _job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep"

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

@@ -1165,6 +1165,44 @@ async def trigger_rosreestr_quarter_poll_run(
     return run_id
 
 
+async def trigger_deals_freshness_monitor_run(
+    db: Session, schedule_row: dict[str, Any]
+) -> int | None:
+    """Создать scrape_runs + launch check_deals_freshness в executor (sync DB-only task).
+
+    Мониторинг свежести ДАННЫХ deals (#2212): смотрит max(deal_date) и поднимает
+    ERROR-алерт (→ GlitchTip), когда следующий квартал просрочен публикацией. Джоба-
+    статус rosreestr_* остаётся зелёным при staleness данных — этот монитор ловит
+    именно устаревание данных, а не сбой прогона.
+
+    Задача синхронная (один SELECT max(deal_date), без внешних HTTP) — гоняем в
+    run_in_executor по образцу trigger_listing_source_snapshot_run.
+
+    Returns run_id (или None если skip — есть running run).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.tasks.deals_freshness_monitor import check_deals_freshness
+
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, check_deals_freshness, run_db, run_id, params)
+        except Exception:
+            logger.exception("scheduler: check_deals_freshness crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    _spawn_tracked(_run())
+    logger.info("scheduler: triggered deals_freshness_monitor run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_newbuilding_enrich_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch run_newbuilding_enrich в asyncio.create_task (#973).
 
@@ -1903,6 +1941,8 @@ async def scheduler_loop() -> None:
                         await trigger_sber_index_pull_run(db, sch)
                     elif source == "rosreestr_quarter_poll":
                         await trigger_rosreestr_quarter_poll_run(db, sch)
+                    elif source == "deals_freshness_monitor":
+                        await trigger_deals_freshness_monitor_run(db, sch)
                     elif source == "newbuilding_enrich":
                         await trigger_newbuilding_enrich_run(db, sch)
                     elif source == "yandex_newbuilding_sweep":

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

@@ -0,0 +1,196 @@
+"""Мониторинг свежести ДАННЫХ deals (не статуса джобы) — #2212.
+
+Проблема аудита: rosreestr_quarter_poll / rosreestr_dkp_import честно рапортуют
+`done` (rows_fetched>0, inserted=0 == «всё уже импортировано»), но это МАСКИРУЕТ
+staleness самих данных. `deals.deal_date` — лейбл начала квартала (2026-01-01 = 2026Q1);
+Росреестр публикует квартальными дампами, скачивание дампа — РУЧНОЙ операторский шаг
+(poll лишь алертит о доступности). Если оператор пропустил импорт нового квартала,
+джоба-статус остаётся зелёным, а данные тихо устаревают.
+
+Этот монитор смотрит на `max(deal_date)` и поднимает ERROR-алерт (→ GlitchTip через
+sentry logging integration), когда следующий квартал ПРОСРОЧЕН публикацией.
+
+Арифметика просрочки (проверена на проде 2026-07-02, max=2026Q1, q_end=2026-03-31):
+  latest_q_start    = max(deal_date)                         (2026-01-01)
+  latest_q_end      = latest_q_start + 3 months              (2026-04-01, старт Q2)
+  overdue_threshold = latest_q_end + 3 months + LAG_DAYS     (2026-07-01 + 45д = 2026-08-15)
+  overdue           = now() > overdue_threshold
+Смысл: следующий квартал (Q2) заканчивается через 3 мес после конца имеющегося; даём
+LAG_ALLOWANCE_DAYS (по умолчанию 45) на публикацию дампа Росреестром. Сейчас (2026-07-02)
+до порога 2026-08-15 ещё далеко → алерта НЕТ. Если к 2026-08-16 Q2 не импортирован → алерт.
+
+Задача синхронная (DB-only, один SELECT max(deal_date)) — запускается in-app scheduler'ом
+через trigger_deals_freshness_monitor_run() в run_in_executor, по образцу
+listing_source_snapshot / rosreestr_dkp_import. Вердикт вычисляет ЧИСТАЯ функция
+evaluate_deals_freshness() (frozen-now тестируется без БД).
+
+Прогон НЕ помечается failed при алерте (это МОНИТОР, а не сбой джобы) — ERROR-лог
+достаточен. mark_failed только если deals недоступна/пуста (нечего оценивать).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import UTC, date, datetime, timedelta
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "DEFAULT_LAG_ALLOWANCE_DAYS",
+    "FreshnessVerdict",
+    "check_deals_freshness",
+    "evaluate_deals_freshness",
+]
+
+# Лаг публикации дампа Росреестром: сколько дней после конца следующего квартала
+# считаем допустимой задержкой до алерта. Читается из default_params.lag_allowance_days.
+DEFAULT_LAG_ALLOWANCE_DAYS = 45
+
+_LATEST_DEAL_DATE_SQL = text("SELECT max(deal_date) AS latest FROM deals")
+
+
+@dataclass(frozen=True)
+class FreshnessVerdict:
+    """Вердикт свежести deals по max(deal_date)."""
+
+    latest_q_start: date
+    latest_q_end: date
+    overdue_threshold: date
+    overdue: bool
+    days_overdue: int
+
+
+def _add_months(d: date, months: int) -> date:
+    """Прибавить `months` к дате. deal_date всегда старт квартала (день=1), но
+    хелпер устойчив к произвольному дню — при переполнении зажимает на конец месяца.
+    """
+    total = (d.month - 1) + months
+    year = d.year + total // 12
+    month = total % 12 + 1
+    # Зажим дня на последний день целевого месяца (для дней 29-31; для day=1 no-op).
+    if month == 12:
+        next_month_first = date(year + 1, 1, 1)
+    else:
+        next_month_first = date(year, month + 1, 1)
+    last_day = (next_month_first - timedelta(days=1)).day
+    return date(year, month, min(d.day, last_day))
+
+
+def evaluate_deals_freshness(
+    latest_q_start: date,
+    now: datetime,
+    lag_allowance_days: int = DEFAULT_LAG_ALLOWANCE_DAYS,
+) -> FreshnessVerdict:
+    """Чистая логика: просрочен ли следующий квартал относительно max(deal_date).
+
+    overdue = now() > (latest_q_start + 6 месяцев + lag_allowance_days).
+    days_overdue = max(0, дни сверх порога) — 0 когда не просрочено.
+
+    Тестируется с frozen `now` без БД.
+    """
+    latest_q_end = _add_months(latest_q_start, 3)
+    overdue_threshold = _add_months(latest_q_start, 6) + timedelta(days=lag_allowance_days)
+    now_date = now.date()
+    overdue = now_date > overdue_threshold
+    days_overdue = max(0, (now_date - overdue_threshold).days)
+    return FreshnessVerdict(
+        latest_q_start=latest_q_start,
+        latest_q_end=latest_q_end,
+        overdue_threshold=overdue_threshold,
+        overdue=overdue,
+        days_overdue=days_overdue,
+    )
+
+
+def check_deals_freshness(
+    db: Session,
+    run_id: int,
+    params: dict | None = None,  # type: ignore[type-arg]
+    now: datetime | None = None,
+) -> dict[str, int]:
+    """Проверить свежесть deals по max(deal_date) и поднять алерт при просрочке.
+
+    Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
+    Читает один SELECT max(deal_date), считает вердикт чистой функцией, логирует
+    ERROR при overdue (→ GlitchTip) и финализирует run.
+
+    Params (default_params jsonb):
+        lag_allowance_days: int — допустимый лаг публикации дампа (default 45).
+    `now` инъектируется в тестах (frozen); в проде — None → datetime.now(UTC).
+
+    Returns counters {latest_year, latest_quarter, days_overdue, alert}.
+    mark_failed только если deals пуста/недоступна (нечего оценивать); при алерте
+    прогон помечается done (это монитор, не сбой джобы).
+    """
+    params = params or {}
+    now = now or datetime.now(UTC)
+    counters: dict[str, int] = {
+        "latest_year": 0,
+        "latest_quarter": 0,
+        "days_overdue": 0,
+        "alert": 0,
+    }
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        row = db.execute(_LATEST_DEAL_DATE_SQL).first()
+        latest: date | None = row.latest if row is not None else None
+        if latest is None:
+            logger.warning(
+                "deals freshness: таблица deals пуста/недоступна — оценить свежесть нельзя"
+            )
+            runs_mod.mark_failed(db, run_id, "deals empty or unavailable", counters)
+            return counters
+
+        lag_days = int(params.get("lag_allowance_days", DEFAULT_LAG_ALLOWANCE_DAYS))
+        verdict = evaluate_deals_freshness(latest, now, lag_days)
+
+        quarter = (latest.month - 1) // 3 + 1
+        counters = {
+            "latest_year": latest.year,
+            "latest_quarter": quarter,
+            "days_overdue": verdict.days_overdue,
+            "alert": int(verdict.overdue),
+        }
+
+        if verdict.overdue:
+            logger.error(
+                "deals freshness: max(deal_date)=%s — новый квартал просрочен на %d дней; "
+                "проверь rosreestr_quarter_poll + операторский импорт "
+                "(data/sql/02_load_all_quarters.sh)",
+                latest,
+                verdict.days_overdue,
+            )
+        else:
+            logger.info(
+                "deals freshness: max(deal_date)=%s (Q%d %d) свежий — порог просрочки %s "
+                "ещё не наступил, алерта нет",
+                latest,
+                quarter,
+                latest.year,
+                verdict.overdue_threshold,
+            )
+
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "check_deals_freshness run_id=%d done: latest=%s alert=%d days_overdue=%d",
+            run_id,
+            latest,
+            counters["alert"],
+            counters["days_overdue"],
+        )
+        return counters
+    except Exception as exc:
+        logger.exception("check_deals_freshness run_id=%d failed", run_id)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/162_seed_deals_freshness_monitor.sql

@@ -0,0 +1,75 @@
+-- 162_seed_deals_freshness_monitor.sql
+-- Issue #2212 — seed scrape_schedules row for the daily deals-data freshness monitor.
+--
+-- Context (verified on prod 2026-07-02):
+--   deals.deal_date is the LABEL of a quarter's start (2026-01-01 == 2026Q1); Rosreestr
+--   publishes quarterly dumps. Current max(deal_date) = 2026Q1 — the LATEST published
+--   quarter, so the import is NOT broken (rows_fetched 49791 / inserted 0 == already
+--   fully imported). Downloading a new dump is a MANUAL operator step; rosreestr_quarter_poll
+--   only detects availability and alerts — it does not ingest.
+--
+--   Audit gap: the `done` status of rosreestr_quarter_poll / rosreestr_dkp_import MASKS
+--   staleness of the DATA. A green job-status can sit on top of quarter-stale deals if
+--   the operator missed the manual import. This monitor watches max(deal_date) directly
+--   and raises an ERROR alert (→ GlitchTip via sentry logging integration) when the next
+--   quarter is overdue for publication.
+--
+--   Overdue arithmetic (app/tasks/deals_freshness_monitor.py::evaluate_deals_freshness):
+--     latest_q_start    = max(deal_date)                      (2026-01-01)
+--     latest_q_end      = latest_q_start + 3 months           (2026-04-01, start of Q2)
+--     overdue_threshold = latest_q_end + 3 months + lag_days  (2026-07-01 + 45d = 2026-08-15)
+--     overdue           = now() > overdue_threshold
+--   At 2026-07-02 the threshold (2026-08-15) is not reached -> NO alert. If Q2 is still
+--   not imported by 2026-08-16, the monitor starts alerting daily.
+--
+--   Components deployed together:
+--     1. app/tasks/deals_freshness_monitor.py — check_deals_freshness() + evaluate_deals_freshness()
+--     2. app/services/scheduler.py — trigger_deals_freshness_monitor_run() + dispatch branch
+--        `elif source == "deals_freshness_monitor"`
+--     3. app/services/product_handlers.py — kit Handler 'deals_freshness_monitor'
+--     4. This migration — seeds the scrape_schedules row (enabled=true, daily window)
+--
+-- Schedule window 08:00-09:00 UTC:
+--   Daily cadence, placed AFTER rosreestr_dkp_import (04:16 UTC window) so the monitor
+--   observes the freshest possible deals state each morning. Cheap: one SELECT max(deal_date),
+--   no external HTTP — safe to run daily.
+--
+-- default_params.lag_allowance_days = 45:
+--   Allowance for Rosreestr to publish the next quarter's dump after that quarter ends.
+--
+-- next_run_at bootstrapped to tomorrow 08:00 UTC — scheduler will not fire immediately
+-- on deploy (same pattern as 093/096/160).
+--
+-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
+--
+-- Dependencies:
+--   052_scrape_schedules.sql (table + UNIQUE(source))
+--   002_core_tables.sql (deals table with deal_date)
+--   app/services/scheduler.py + product_handlers.py + tasks/deals_freshness_monitor.py deployed
+--
+-- Deploy order:
+--   Apply after deploying the scheduler.py + product_handlers.py + task changes so the
+--   dispatch can resolve 'deals_freshness_monitor' correctly on first fire.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'deals_freshness_monitor',
+    true,                                        -- SAFE: single SELECT max(deal_date), no ext calls
+    8,                                           -- window 08:00-09:00 UTC (after rosreestr_dkp_import)
+    9,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 8)) AT TIME ZONE 'UTC',
+    '{"lag_allowance_days": 45}'::jsonb          -- publication-lag allowance before alerting
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMIT;

tradein-mvp/backend/tests/test_deals_freshness_monitor.py

@@ -0,0 +1,251 @@
+"""Freshness-монитор данных deals по max(deal_date) — #2212.
+
+Покрывает:
+  1. Чистую логику evaluate_deals_freshness (frozen now, без БД):
+     - нет алерта: 2026Q1 @ 2026-07-02 (порог 2026-08-15 не наступил);
+     - алерт:     2026Q1 @ 2026-08-20 (days_overdue=5);
+     - граница порога (== threshold → нет алерта; +1 день → алерт);
+     - кастомный lag_allowance_days.
+  2. check_deals_freshness с FakeDB (no-alert / alert / empty→mark_failed).
+  3. Свойства миграции 162 (по образцу test_deactivate_stale_listings).
+  4. Регистрацию в scheduler-dispatch + kit product_handlers (registry остаётся зелёным).
+"""
+
+from __future__ import annotations
+
+import inspect
+import os
+import re
+from datetime import UTC, date, datetime
+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.services.product_handlers import build_product_handlers
+from app.tasks import deals_freshness_monitor as mon
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_162 = _SQL_DIR / "162_seed_deals_freshness_monitor.sql"
+
+# max(deal_date) = 2026Q1 label (проверено на проде 2026-07-02).
+_Q1_2026 = date(2026, 1, 1)
+
+
+# ── evaluate_deals_freshness (чистая логика, frozen now) ──────────────────────
+
+
+def _now(y: int, m: int, d: int) -> datetime:
+    return datetime(y, m, d, tzinfo=UTC)
+
+
+def test_no_alert_now_2026_07_02() -> None:
+    """Сейчас: 2026Q1 @ 2026-07-02 — порог просрочки 2026-08-15 не наступил."""
+    v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 7, 2))
+    assert v.overdue is False
+    assert v.days_overdue == 0
+    assert v.latest_q_end == date(2026, 4, 1)
+    assert v.overdue_threshold == date(2026, 8, 15)
+
+
+def test_alert_2026_08_20() -> None:
+    """К 2026-08-20 Q2 не импортирован — алерт, просрочка 5 дней (порог 2026-08-15)."""
+    v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 8, 20))
+    assert v.overdue is True
+    assert v.days_overdue == 5
+
+
+def test_threshold_boundary_exact_no_alert() -> None:
+    """Ровно в день порога (2026-08-15) алерта ещё нет (строгое >)."""
+    v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 8, 15))
+    assert v.overdue is False
+    assert v.days_overdue == 0
+
+
+def test_threshold_boundary_next_day_alert() -> None:
+    """Порог + 1 день (2026-08-16) — первый алерт, просрочка 1 день."""
+    v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 8, 16))
+    assert v.overdue is True
+    assert v.days_overdue == 1
+
+
+def test_custom_lag_allowance_shifts_threshold() -> None:
+    """lag_allowance_days=0 сдвигает порог на конец следующего квартала (2026-07-01)."""
+    v = mon.evaluate_deals_freshness(_Q1_2026, _now(2026, 7, 2), lag_allowance_days=0)
+    assert v.overdue_threshold == date(2026, 7, 1)
+    assert v.overdue is True
+    assert v.days_overdue == 1
+
+
+def test_add_months_quarter_starts() -> None:
+    assert mon._add_months(date(2026, 1, 1), 3) == date(2026, 4, 1)
+    assert mon._add_months(date(2026, 1, 1), 6) == date(2026, 7, 1)
+    assert mon._add_months(date(2026, 10, 1), 3) == date(2027, 1, 1)
+
+
+def test_add_months_end_of_month_clamp() -> None:
+    # 31 янв + 1 мес → 28 фев (зажим на конец месяца).
+    assert mon._add_months(date(2026, 1, 31), 1) == date(2026, 2, 28)
+
+
+# ── check_deals_freshness (FakeDB) ────────────────────────────────────────────
+
+
+class _Row:
+    def __init__(self, latest: date | None) -> None:
+        self.latest = latest
+
+
+class _FakeResult:
+    def __init__(self, latest: date | None) -> None:
+        self._latest = latest
+
+    def first(self) -> _Row:
+        return _Row(self._latest)
+
+
+class _FakeDB:
+    def __init__(self, latest: date | None) -> None:
+        self._latest = latest
+        self.rolled_back = False
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        return _FakeResult(self._latest)
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+
+def _patch_runs(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
+    calls: dict[str, Any] = {"done": None, "failed": None, "heartbeat": 0}
+    monkeypatch.setattr(
+        mon.runs_mod,
+        "update_heartbeat",
+        lambda *a, **k: calls.__setitem__("heartbeat", calls["heartbeat"] + 1),
+    )
+    monkeypatch.setattr(
+        mon.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: calls.__setitem__("done", dict(counters)),
+    )
+    monkeypatch.setattr(
+        mon.runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: calls.__setitem__("failed", err),
+    )
+    return calls
+
+
+def test_check_no_alert_marks_done(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch_runs(monkeypatch)
+    db = _FakeDB(_Q1_2026)
+    out = mon.check_deals_freshness(db, run_id=1, params={}, now=_now(2026, 7, 2))  # type: ignore[arg-type]
+    assert out == {"latest_year": 2026, "latest_quarter": 1, "days_overdue": 0, "alert": 0}
+    assert calls["done"] == out
+    assert calls["failed"] is None
+
+
+def test_check_overdue_marks_done_with_alert(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch_runs(monkeypatch)
+    db = _FakeDB(_Q1_2026)
+    out = mon.check_deals_freshness(db, run_id=2, params={}, now=_now(2026, 8, 20))  # type: ignore[arg-type]
+    assert out["alert"] == 1
+    assert out["days_overdue"] == 5
+    # Монитор НЕ падает при алерте — прогон done, а не failed.
+    assert calls["done"] == out
+    assert calls["failed"] is None
+
+
+def test_check_empty_deals_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch_runs(monkeypatch)
+    db = _FakeDB(None)
+    out = mon.check_deals_freshness(db, run_id=3, params={}, now=_now(2026, 7, 2))  # type: ignore[arg-type]
+    assert out["alert"] == 0
+    assert calls["done"] is None
+    assert calls["failed"] is not None
+
+
+def test_check_reads_lag_from_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch_runs(monkeypatch)
+    db = _FakeDB(_Q1_2026)
+    # lag=0 → порог 2026-07-01 → @2026-07-02 уже просрочено.
+    out = mon.check_deals_freshness(
+        db, run_id=4, params={"lag_allowance_days": 0}, now=_now(2026, 7, 2)
+    )  # type: ignore[arg-type]
+    assert out["alert"] == 1
+
+
+# ── Миграция 162 ──────────────────────────────────────────────────────────────
+
+
+def test_migration_162_exists() -> None:
+    assert _MIGRATION_162.is_file(), f"missing migration: {_MIGRATION_162}"
+
+
+def test_migration_162_seeds_source() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert "'deals_freshness_monitor'" in sql
+
+
+def test_migration_162_enabled_true() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert "true" in sql
+
+
+def test_migration_162_is_idempotent() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+
+
+def test_migration_162_is_transactional() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_162_window_8_to_9_utc() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert re.search(r"\b8\b", sql), "window_start_hour 8 missing"
+    assert re.search(r"\b9\b", sql), "window_end_hour 9 missing"
+
+
+def test_migration_162_lag_allowance_45() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert "lag_allowance_days" in sql
+    assert "45" in sql
+
+
+def test_migration_162_no_psycopg_trap() -> None:
+    sql = _MIGRATION_162.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+# ── Регистрация в scheduler + kit registry ────────────────────────────────────
+
+
+def test_trigger_exists_and_is_async() -> None:
+    fn = getattr(scheduler, "trigger_deals_freshness_monitor_run", None)
+    assert fn is not None, "trigger_deals_freshness_monitor_run missing from scheduler"
+    assert inspect.iscoroutinefunction(fn)
+
+
+def test_dispatch_branch_present() -> None:
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source == "deals_freshness_monitor"' in loop_src
+    assert "trigger_deals_freshness_monitor_run(db, sch)" in loop_src
+
+
+def test_trigger_runs_sync_task_in_executor() -> None:
+    trig_src = inspect.getsource(scheduler.trigger_deals_freshness_monitor_run)
+    assert "run_in_executor(" in trig_src
+    assert "check_deals_freshness" in trig_src
+    assert "_claim_run(db, schedule_row)" in trig_src
+
+
+def test_kit_product_handler_registered() -> None:
+    # ctx не нужен для сборки dict ключей — build_product_handlers его не замыкает.
+    handlers = build_product_handlers(ctx=None)  # type: ignore[arg-type]
+    assert "deals_freshness_monitor" in handlers