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

"""In-app scheduler — periodically triggers configured scrapes based on scrape_schedules table.

Запускается в FastAPI lifespan (app.main lifespan context manager) при startup.
Логика:
1. Каждые 60 sec query scrape_schedules WHERE enabled=true AND next_run_at <= NOW().
2. Для каждого: проверить нет ли уже running run этого source — если нет, trigger.
3. После trigger — recompute next_run_at через interval_days суток (из default_params,
   default 1 = daily; например avito_full_load_exhaustive = 7 → weekly).
4. На old runs (orphaned status='running' > 6h без heartbeat) — пометить как 'zombie'.

Sources:
  - avito_city_sweep        → run_avito_city_sweep (scrape_pipeline.py)
  - avito_full_load         → run_avito_full_load (региональный сбор Avito ЕКБ
                              ВТОРИЧКИ без anchor'ов — room×price адаптивное партиционирование
                              через pmin/pmax, secondary_only, incremental on_bucket save;
                              обходит SERP-cap ~5000/запрос. Окно 13:00-15:00 UTC чтобы не
                              конфликтовать на shared apw-прокси с avito_city_sweep (6-7),
                              avito_newbuilding (2-5), avito_detail_backfill (9-12).
                              ПО УМОЛЧАНИЮ инкрементальный: incremental_days из default_params
                              → shallow обход с date early-stop, избегает page-36 429-банов)
  - avito_full_load_exhaustive → run_avito_full_load с incremental_days=None (полный
                              exhaustive обход; weekly cadence через interval_days=7 в
                              default_params — освежает last_seen 10-дневный delisting TTL
                              + ловит тихие правки цены)
  - avito_newbuilding_sweep → run_avito_newbuilding_sweep (scrape_pipeline.py; dedicated
                              novostroyka-filtered citywide SERP → save, 100% new-build cards)
  - yandex_city_sweep       → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT)
  - cian_history_backfill   → run_cian_backfill (this module, #560)
  - rosreestr_dkp_import    → import_rosreestr_dkp (this module, #563)
  - listing_source_snapshot → snapshot_listing_sources (tasks/listing_source_snapshot.py, #570;
                              pure internal DB snapshot — no external calls, enabled by default)
  - asking_to_sold_ratio_refresh → recompute_asking_to_sold_ratios
                              (tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB
                              re-derivation of the asking→sold ratios — no external calls,
                              enabled by default; window after rosreestr_dkp_import)
  - yandex_address_backfill  → run_yandex_address_backfill
                              (tasks/yandex_address_backfill.py, #855; EKB-pilot: fetches
                              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_*      → 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
                              anti-bot/proxy; window 05:00-06:00 UTC, ~28-day cadence)
  - rosreestr_quarter_poll  → run_rosreestr_quarter_poll
                              (tasks/rosreestr_quarter_poll.py, #888; monthly HEAD-check of
                              Rosreestr open-data portal for new quarter availability — pure
                              HTTP detect+alert, NO download/shell-out; window 07:00-08:00 UTC,
                              ~28-day cadence)
  - newbuilding_enrich      → run_newbuilding_enrich
                              (tasks/newbuilding_enrich_backfill.py, #973; nightly slice of the
                              cian_newbuilding houses enrichment — resolves cian_zhk_url, fetches
                              the ЖК page, persists houses_price_dynamics / house_reliability_checks
                              / house_reviews. Cian HTTP (curl_cffi chrome120) + anti-bot delay.
                              Idempotent: skips houses already enriched, per-house SAVEPOINT, so
                              each fire drains the next `limit` pending houses; window 00:00-01:00
                              UTC = 03:00-04:00 МСК, before the 01:00+ UTC sweep block)
  - yandex_newbuilding_sweep → enrich_yandex_newbuilding_sweep
                              (tasks/yandex_newbuilding_sweep.py, #974; enrichment ЖК в
                              market.yandex_jk_enrichment через BrowserFetcher; shipped DORMANT —
                              enable manually after deploy verified; window 02:00-05:00 UTC)
  - cian_city_sweep         → run_cian_city_sweep (newbuilding Phase 4, #973; shipped DORMANT —
                              proxy dead, enable manually after proxy restored. NB-only с #973+:
                              SERP-фаза сохраняет только novostroyki-лоты — вторичкой владеет
                              cian_full_load. DETAIL/HOUSES-фазы не затронуты)
  - cian_full_load          → run_cian_full_load (exhaustive региональный сбор Cian ЕКБ ВТОРИЧКИ
                              без anchor'ов — room×price адаптивное партиционирование,
                              secondary_only, incremental on_bucket save; авторитетный источник
                              вторички. Окно 20:00-22:00 UTC чтобы не конфликтовать на shared
                              kf-прокси с cian_city_sweep (2-5) и cian_history_backfill (2-5))
  - domclick_city_sweep     → run_domclick_city_sweep (scrape_pipeline.py; citywide SERP via
                              camoufox BrowserFetcher; shipped DORMANT enabled=false, enable
                              manually after prod-smoke)
  - geoportal_coords_backfill → run_geoportal_coords_backfill
                              (tasks/backfill_listings_coords_geoportal.py, #1967; nightly
                              exact street+house match против ekb_geoportal_buildings (местный
                              реестр зданий ЕКБ) для listings.lat IS NULL. Pure internal DB op —
                              никакого внешнего HTTP/rate-limit (в отличие от Nominatim ниже).
                              Ранее был manual-only script (#1841); recurring нужен потому что
                              иначе новые NULL-coord листинги ждут geocode_missing_listings и
                              получают более грубый (coarse) city-centroid geocode вместо
                              точного house-level матча. Window 05:00-06:00 UTC — ДО
                              geocode_missing_listings (06:00-09:00 UTC), чтобы точный матч
                              имел приоритет над Nominatim.)
  - geocode_missing_listings → run_geocode_missing_listings
                              (tasks/geocode_missing.py; nightly backfill listings.lat/lon для
                              всех источников с NULL coords; address-dedup batch loop с
                              wall-clock budget; window 06:00-09:00 UTC — после city sweeps
                              ~03-04 UTC, deals OS-cron 05:30/05:45 UTC и geoportal_coords_backfill
                              05:00-06:00 UTC)
  - yandex_detail_backfill  → run_yandex_detail_backfill
                              (tasks/yandex_detail_backfill.py, #1553; nightly backfill of
                              yandex detail_enriched_at for ~3952 listings never enriched by
                              city_sweep; fetch via curl_cffi chrome120 + scraper_proxy_url,
                              parse via YandexDetailScraper.parse;
                              window 12:00-15:00 UTC — после avito_detail_backfill 09-12 UTC)
  - cadastral_geo_match     → run_cadastral_geo_match
                              (tasks/cadastral_geo_match.py; combined nightly job — (1) REFRESH
                              cad_buildings_local from gendesign_cad_buildings FDW (one bulk
                              scan, ~36.7k EKB rows, Point geom + GIST), (2) MATCH set-based
                              LATERAL KNN UPDATE filling listings.building_cadastral_number from
                              the nearest cadastral building within threshold_m (default 50m).
                              GEO-NEAREST approximation (nearest building, not exact cadastral).
                              Pure internal DB op — one FDW read + local UPDATE, no HTTP/anti-bot;
                              window 09:00-10:00 UTC — после geocode_missing_listings 06-09 UTC
                              чтобы listings имели свежие lat/lon/geom)
  - house_imv_backfill      → backfill_house_imv
                              (services/house_imv_backfill.py, #854; nightly bulk Avito IMV
                              на уровне домов — итерирует houses imv_status='pending' с lat/lon
                              + linked listings, считает медианные lot-params, зовёт
                              evaluate_via_imv, наполняет house_imv_evaluations /
                              house_placement_history / house_suggestions, переводит
                              houses.imv_status pending→ok/not_found/error. Resumable +
                              idempotent (UPSERT по house_id, фильтр imv_status). Avito IMV
                              использует собственную curl_cffi-сессию; window 15:00-17:00 UTC —
                              после avito_full_load 13-15 UTC и утреннего Avito-блока, дома
                              уже geocoded/enriched дневными sweep'ами)
  - house_dedup_merge       → run_house_dedup_merge
                              (services/house_dedup_merge.py, #1772; recurring, idempotent,
                              collision-safe house-deduplication merge. Clusters houses by
                              EXACT address (cadastral_number is 100% NULL), picks one canonical
                              keeper per cluster (geom NOT NULL → most listings → most-populated
                              → min id), re-points all FK children with UNIQUE-collision handling
                              — reuses migration 108's proven pattern — deletes loser rows,
                              backfills house_sources/house_address_aliases onto the keeper so the
                              matching pipeline finds the keeper next scrape. All in ONE
                              transaction; dry_run param computes counts then ROLLBACKs.
                              DESTRUCTIVE (DELETEs duplicate house rows) but idempotent: clean
                              table → empty mapping → 0-row no-op. Seed 135 ships the schedule
                              DISABLED (enabled=false) — deploy is neutral; enable deliberately
                              after a validated dry-run + manual run; weekly window 04:00-05:00
                              UTC — quiet hour, no scraper/proxy contention)
  - osm_poi_ekb_refresh     → run_osm_poi_ekb_refresh
                              (tasks/osm_poi_ekb_refresh.py, #2045 BE-3; single bulk FDW scan —
                              TRUNCATE osm_poi_ekb_local; INSERT ... FROM gendesign_osm_poi_ekb
                              (Site Finder OSM POI, EKB), building Point(4326) geom from lat/lon.
                              Feeds app/services/location_coef.py (GET /trade-in/location-coef,
                              LocationDrawer). OSM POI changes rarely → daily cadence is ample.
                              Pure internal DB op — one FDW read + local TRUNCATE+INSERT, no
                              HTTP/anti-bot; window 10:00-11:00 UTC — после cadastral_geo_match
                              09:00-10:00 UTC)
"""

from __future__ import annotations

import asyncio
import logging
import random
from collections.abc import Coroutine
from datetime import UTC, datetime, time, timedelta
from typing import Any

import sentry_sdk
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.core.shutdown import shutdown_requested
from app.services import scrape_runs as runs_mod
from app.services.scrape_pipeline import (
    run_avito_city_sweep,
    run_avito_full_load,
    run_avito_newbuilding_sweep,
    run_cian_city_sweep,
    run_cian_full_load,
    run_domclick_city_sweep,
    run_yandex_city_sweep,
)

logger = logging.getLogger(__name__)

# Loop interval — check каждую минуту
SCHEDULER_TICK_SEC = 60
ZOMBIE_THRESHOLD_HOURS = 6

# #1182 Phase 2: graceful SIGTERM-drain. Scrape-работа крутится в detached
# asyncio.create_task (см. _spawn_tracked) — coordinator (scheduler_loop) их
# НЕ join'ит. Без явного дренажа asyncio.run() teardown хард-кансельнул бы их
# mid-await (тот самый #1182 failure mode). Регистр strong-ref'ов даёт coordinator'у
# дождаться детей перед выходом из tick-loop'а.
_inflight_tasks: set[asyncio.Task[None]] = set()

# Бюджет дренажа детей < scheduler_main._DRAIN_TIMEOUT_S (100s) < docker grace (120s):
# coordinator успевает дождаться кооперативных детей (avito_detail_backfill,
# rosreestr-executor дочекивают карточку/батч + mark_done и резолвятся) ВНУТРИ
# внешнего wait_for, а некооперативные упираются в этот timeout и падают на внешний
# hard-cancel fallback.
_CHILD_DRAIN_TIMEOUT_S = 80.0


def _spawn_tracked(coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
    """create_task + регистрация задачи в _inflight_tasks для graceful-drain'а (#1182 P2).

    Заменяет прежний `task = create_task(_run()); task.add_done_callback(...)`:
    strong-ref в set'е держит задачу до завершения (RUF006) и даёт scheduler_loop'у
    дождаться её на SIGTERM-drain'е; done-callback ретривит exception (чтобы не было
    'Task exception was never retrieved') и убирает задачу из set'а.
    """
    task = asyncio.create_task(coro)
    _inflight_tasks.add(task)

    def _on_done(t: asyncio.Task[None]) -> None:
        _inflight_tasks.discard(t)
        if not t.cancelled():
            t.exception()

    task.add_done_callback(_on_done)
    return task


async def _drain_inflight() -> None:
    """Дождаться завершения detached run-задач перед teardown'ом процесса (#1182 P2).

    Зовётся из scheduler_loop при выходе из tick-loop'а по SIGTERM-drain'у. Кооперативные
    дети дочекивают текущий unit (карточку/батч), делают mark_done и резолвятся;
    некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S и остаются на внешний hard-cancel
    fallback (scheduler_main 100s + docker 120s). Не busy-spin — один asyncio.wait.
    """
    pending = [t for t in _inflight_tasks if not t.done()]
    if not pending:
        return
    logger.info("scheduler: draining %d in-flight run task(s) on shutdown", len(pending))
    _done, still = await asyncio.wait(pending, timeout=_CHILD_DRAIN_TIMEOUT_S)
    if still:
        logger.warning(
            "scheduler: %d task(s) did not drain in %.0fs — leaving for hard-cancel",
            len(still),
            _CHILD_DRAIN_TIMEOUT_S,
        )
    else:
        logger.info("scheduler: all in-flight run task(s) drained cleanly")


def compute_next_run_at(
    window_start_hour: int,
    window_end_hour: int,
    *,
    now: datetime | None = None,
    interval_days: int = 1,
) -> datetime:
    """Pick random datetime в window [start, end) UTC, через interval_days суток после now.

    interval_days задаёт каденс источника: 1 (default) = daily (back-compat), 7 = weekly.
    Берётся из schedule.default_params["interval_days"] вызывающим кодом (_claim_run /
    _defer_next_run_at); отсутствие ключа → 1 → прежнее ежедневное поведение.

    Если window_end_hour <= window_start_hour → cross-midnight window
    (например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59).
    """
    now = now or datetime.now(tz=UTC)
    interval_days = max(1, int(interval_days))
    # Целевая дата = now + interval_days суток (interval_days=1 → завтра, как раньше).
    target = (now + timedelta(days=interval_days)).date()

    if window_end_hour > window_start_hour:
        # Обычное окно (например 2..5 → 02:00-04:59)
        start_seconds = window_start_hour * 3600
        end_seconds = window_end_hour * 3600
        rand_seconds = random.randint(start_seconds, end_seconds - 1)
        return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds)
    else:
        # Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59)
        # Длина окна = (24-start) + end часов
        total_seconds = ((24 - window_start_hour) + window_end_hour) * 3600
        rand_seconds = random.randint(0, total_seconds - 1)
        # Если rand попадает в первую часть (start..24)
        first_half = (24 - window_start_hour) * 3600
        if rand_seconds < first_half:
            # interval_days=1: текущая дата (если окно ещё не наступило сегодня) или next day.
            # interval_days>1: всегда целевая дата (стаггер на N суток вперёд).
            today_ok = interval_days == 1 and now.hour < window_start_hour
            base_date = now.date() if today_ok else target
            return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta(
                seconds=window_start_hour * 3600 + rand_seconds
            )
        else:
            # Во второй части (0..end), целевого дня
            offset = rand_seconds - first_half
            return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset)


def has_running_run(db: Session, source: str) -> bool:
    """Есть ли активный run для source (status='running')."""
    row = db.execute(
        text(
            """
            SELECT 1 FROM scrape_runs
            WHERE source = :source AND status = 'running'
            LIMIT 1
            """
        ),
        {"source": source},
    ).fetchone()
    return row is not None


def reap_zombies(db: Session) -> int:
    """Mark scrape_runs as 'zombie' если heartbeat не обновлялся > ZOMBIE_THRESHOLD_HOURS hours."""
    zombie_interval = f"{ZOMBIE_THRESHOLD_HOURS} hours"
    result = db.execute(
        text(
            """
            UPDATE scrape_runs
            SET status = 'zombie', finished_at = NOW()
            WHERE status = 'running'
              AND (heartbeat_at IS NULL
                   OR heartbeat_at < NOW() - CAST(:interval AS interval))
            RETURNING id
            """
        ),
        {"interval": zombie_interval},
    )
    rows = result.fetchall()
    db.commit()
    if rows:
        logger.warning("scheduler: reaped %d zombie runs: %s", len(rows), [r.id for r in rows])
    return len(rows)


def _claim_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at.

    Returns run_id или None если уже есть running run для этого source.
    Общий helper для всех source-handler'ов.
    """
    source = schedule_row["source"]
    if has_running_run(db, source):
        logger.info("scheduler: skip — already running for source=%s", source)
        return None

    # #750: атомарный claim через transaction-scoped advisory lock. Сериализует
    # конкурентные _claim_run одного source (redeploy-overlap старого+нового контейнера
    # / 2 реплики), иначе оба пройдут has_running_run-pre-check и запустят ДВОЙНОЙ sweep
    # → 2× rate к Avito/Cian → бан. pg_try_advisory_xact_lock НЕ блокирует: lock занят
    # другим тиком → false → return None. Лок авто-снимается на commit/rollback этой
    # транзакции (has_running_run выше — быстрый pre-check, обычно без взятия лока).
    got_lock = db.execute(
        text("SELECT pg_try_advisory_xact_lock(hashtext(:source))"),
        {"source": source},
    ).scalar()
    if not got_lock:
        logger.info("scheduler: skip — concurrent claim in progress for source=%s", source)
        return None

    # Double-checked под локом: конкурентный тик мог закоммитить running-run МЕЖДУ
    # pre-check и взятием лока (затем освободить lock на commit). Перепроверяем под
    # локом — иначе словим двойной run на узком окне.
    if has_running_run(db, source):
        logger.info("scheduler: skip — running appeared under lock for source=%s", source)
        db.rollback()  # освобождаем advisory lock (claim не состоялся)
        return None

    params = schedule_row.get("default_params") or {}
    run_id = runs_mod.create_run(db, source=source, params=params)

    next_at = compute_next_run_at(
        schedule_row["window_start_hour"],
        schedule_row["window_end_hour"],
        interval_days=int(params.get("interval_days", 1)),
    )
    db.execute(
        text(
            """
            UPDATE scrape_schedules
            SET last_run_id = :run_id, last_run_at = NOW(),
                next_run_at = :next_at, updated_at = NOW()
            WHERE source = :source
            """
        ),
        {"run_id": run_id, "next_at": next_at, "source": source},
    )
    db.commit()
    logger.info(
        "scheduler: claimed run_id=%d source=%s next_run_at=%s",
        run_id,
        source,
        next_at.isoformat(),
    )
    return run_id


def _defer_next_run_at(db: Session, schedule_row: dict[str, Any]) -> None:
    """Сдвинуть next_run_at на следующее окно БЕЗ создания run (#1522).

    Используется когда trigger делает early-return (например cian: отсутствуют/протухли
    cookies) ДО _claim_run. Без этого get_due_schedules переотбирает schedule на каждом
    тике (SCHEDULER_TICK_SEC), и pre-check (verify_session → реальный HTTP к Cian) гоняется
    раз в минуту круглосуточно. Сдвиг откладывает следующую попытку до завтрашнего окна.
    """
    source = schedule_row["source"]
    params = schedule_row.get("default_params") or {}
    next_at = compute_next_run_at(
        schedule_row["window_start_hour"],
        schedule_row["window_end_hour"],
        interval_days=int(params.get("interval_days", 1)),
    )
    db.execute(
        text(
            """
            UPDATE scrape_schedules
            SET next_run_at = :next_at, updated_at = NOW()
            WHERE source = :source
            """
        ),
        {"next_at": next_at, "source": source},
    )
    db.commit()
    logger.info(
        "scheduler: deferred next_run_at source=%s next_run_at=%s (no run claimed)",
        source,
        next_at.isoformat(),
    )


async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task.

    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 {}

    # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
    async def _run() -> None:
        run_db = SessionLocal()
        try:
            await run_avito_city_sweep(
                run_db,
                run_id=run_id,
                pages_per_anchor=int(params.get("pages_per_anchor", 3)),
                detail_top_n=int(params.get("detail_top_n", 20)),
                request_delay_sec=float(params.get("request_delay_sec", 7.0)),
                enrich_houses=bool(params.get("enrich_houses", True)),
                radius_m=int(params.get("radius_m", 1500)),
            )
        except Exception:
            logger.exception("scheduler: run_avito_city_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered city_sweep run_id=%d", run_id)
    return run_id


async def trigger_avito_newbuilding_sweep_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать новый scrape_runs + launch run_avito_newbuilding_sweep в asyncio.create_task.

    Зеркало trigger_avito_city_sweep_run, но citywide novostroyka-обход (без anchor/
    houses/detail-параметров): dedicated novostroyka-SERP → save.

    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 {}

    # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
    async def _run() -> None:
        run_db = SessionLocal()
        try:
            await run_avito_newbuilding_sweep(
                run_db,
                run_id=run_id,
                pages=int(params.get("pages", 20)),
                request_delay_sec=float(params.get("request_delay_sec", 7.0)),
            )
        except Exception:
            logger.exception("scheduler: run_avito_newbuilding_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered newbuilding_sweep run_id=%d", run_id)
    return run_id


async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать новый scrape_runs + launch run_yandex_city_sweep в asyncio.create_task.

    Зеркало trigger_avito_city_sweep_run, но проще (#561): у Yandex sweep нет
    houses/detail-параметров. Shipped DORMANT — schedule seed enabled=false (078),
    включается оператором вручную после #623 (egress-IP rotation).

    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 {}

    # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
    async def _run() -> None:
        run_db = SessionLocal()
        try:
            await run_yandex_city_sweep(
                run_db,
                run_id=run_id,
                pages_per_anchor=int(params.get("pages_per_anchor", 2)),
                request_delay_sec=float(params.get("request_delay_sec", 9.0)),
                radius_m=int(params.get("radius_m", 1500)),
                enrich_address=bool(params.get("enrich_address", True)),
                segments=params.get("segments"),
            )
        except Exception:
            logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered yandex_city_sweep run_id=%d", run_id)
    return run_id


async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_cian_backfill в asyncio.create_task.

    Pre-checks Cian session validity before claiming the run.
    Marks nothing as failed if cookies are absent/expired — just logs + GlitchTip warning.
    Returns run_id или None (skip / expired cookies).
    """
    from app.services.cian_session import load_session, verify_session

    # Pre-check: load and verify cookies before claiming the run slot
    cookies = load_session(db)
    if cookies is None:
        logger.warning("scheduler: cian_history_backfill skipped — no valid session cookies in DB")
        # #1522: всё равно сдвигаем next_run_at на следующее окно, иначе get_due_schedules
        # переотбирает этот schedule на КАЖДОМ тике (60с) и verify_session долбит Cian
        # круглосуточно, пока оператор вручную не зальёт свежие cookies — burn proxy-квоты
        # + риск TLS/bot-бана.
        _defer_next_run_at(db, schedule_row)
        return None

    state = await verify_session(cookies)
    if state is None:
        logger.warning(
            "scheduler: cian_history_backfill — cookies expired or invalid, skipping run"
        )
        try:
            sentry_sdk.capture_message(
                "cian_history_backfill skipped: Cian session cookies expired — "
                "please re-upload via admin UI",
                level="warning",
            )
        except Exception:
            pass  # sentry_sdk not initialised in dev
        # #1522: cookies протухли — сдвигаем next_run_at (см. комментарий выше), чтобы
        # verify_session не бил по Cian каждые 60с до ручной перезаливки cookies.
        _defer_next_run_at(db, schedule_row)
        return None

    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:
            await _execute_cian_backfill(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: _execute_cian_backfill crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered cian_history_backfill run_id=%d", run_id)
    return run_id


async def trigger_cian_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_cian_city_sweep в asyncio.create_task (#973).

    Несёт Phase 4 newbuilding-enrichment (enrich_houses). Shipped DORMANT — seed
    enabled=false (107), включается оператором вручную после восстановления прокси.
    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:
            await run_cian_city_sweep(
                run_db,
                run_id=run_id,
                pages_per_anchor=int(params.get("pages_per_anchor", 3)),
                request_delay_sec=float(params.get("request_delay_sec", 5.0)),
                radius_m=int(params.get("radius_m", 1500)),
                detail_top_n=int(params.get("detail_top_n", 10)),
                enrich_houses=bool(params.get("enrich_houses", True)),
                newbuilding_only=bool(params.get("newbuilding_only", True)),
            )
        except Exception:
            logger.exception("scheduler: run_cian_city_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered cian_city_sweep run_id=%d", run_id)
    return run_id


async def trigger_cian_full_load_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_cian_full_load в asyncio.create_task.

    Exhaustive региональный сбор Cian ЕКБ ВТОРИЧКИ (room×price партиционирование,
    secondary_only, incremental on_bucket save) — авторитетный источник вторички.
    Образец trigger_cian_city_sweep_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:
            await run_cian_full_load(
                run_db,
                run_id=run_id,
                price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
                concurrency=int(params.get("concurrency", 5)),
                request_delay_sec=float(params.get("request_delay_sec", 4.0)),
                enrich_detail=bool(params.get("enrich_detail", False)),
                detail_top_n=int(params.get("detail_top_n", 0)),
                resume_run_id=None,
            )
        except Exception:
            logger.exception("scheduler: run_cian_full_load crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered cian_full_load run_id=%d", run_id)
    return run_id


async def trigger_avito_full_load_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_avito_full_load в asyncio.create_task.

    Региональный сбор Avito ЕКБ ВТОРИЧКИ (room×price партиционирование через
    pmin/pmax, secondary_only, incremental on_bucket save) — обходит SERP-cap
    ~5000 на запрос. Зеркало trigger_cian_full_load_run. apw-прокси выбирает сам
    scraper (AvitoScraper).

    Режим по умолчанию — инкрементальный: incremental_days читается из
    default_params (shallow обход с date early-stop → избегает page-36 429-банов).
    Для полного exhaustive обхода используется отдельный source
    avito_full_load_exhaustive (incremental_days форсится в None).

    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 {}
    _incremental_days = params.get("incremental_days")
    incremental_days = int(_incremental_days) if _incremental_days is not None else None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            await run_avito_full_load(
                run_db,
                run_id=run_id,
                price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
                concurrency=int(params.get("concurrency", 5)),
                request_delay_sec=float(params.get("request_delay_sec", 7.0)),
                secondary_only=bool(params.get("secondary_only", True)),
                resume_run_id=None,
                incremental_days=incremental_days,
            )
        except Exception:
            logger.exception("scheduler: run_avito_full_load crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered avito_full_load run_id=%d", run_id)
    return run_id


async def trigger_avito_full_load_exhaustive_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch run_avito_full_load (EXHAUSTIVE) в asyncio.create_task.

    Зеркало trigger_avito_full_load_run, но incremental_days форсится в None →
    полный exhaustive обход (root-бисекция всего ценового диапазона на комнатность).
    Запускается реже (weekly через interval_days=7 в default_params → compute_next_run_at
    откладывает next_run_at на 7 суток) чтобы освежить last_seen (10-дневный delisting TTL)
    и поймать тихие правки цены, которые shallow-инкремент пропускает.
    Все прочие params (concurrency, price_cap_per_bucket, ...) читаются из
    default_params как у avito_full_load. Returns run_id (или None если skip).
    """
    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:
            await run_avito_full_load(
                run_db,
                run_id=run_id,
                price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
                concurrency=int(params.get("concurrency", 5)),
                request_delay_sec=float(params.get("request_delay_sec", 7.0)),
                secondary_only=bool(params.get("secondary_only", True)),
                resume_run_id=None,
                incremental_days=None,  # force exhaustive full walk
            )
        except Exception:
            logger.exception(
                "scheduler: run_avito_full_load (exhaustive) crashed run_id=%d", run_id
            )
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered avito_full_load_exhaustive run_id=%d", run_id)
    return run_id


async def trigger_domclick_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_domclick_city_sweep в asyncio.create_task.

    Зеркало trigger_yandex_city_sweep_run, но citywide (city_id, без anchor/geo):
    DomClick SERP не поддерживает geo-radius. Shipped DORMANT — seed enabled=false
    (122), включается оператором вручную после prod-smoke.

    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:
            await run_domclick_city_sweep(
                run_db,
                run_id=run_id,
                city_id=int(params.get("city_id", 4)),
                rooms=params.get("rooms"),
                pages=int(params.get("pages_per_anchor", 5)),
                request_delay_sec=float(params.get("request_delay_sec", 6.0)),
            )
        except Exception:
            logger.exception("scheduler: run_domclick_city_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered domclick_city_sweep run_id=%d", run_id)
    return run_id


async def _execute_cian_backfill(
    db: Session,
    *,
    run_id: int,
    params: dict[str, Any],
) -> None:
    """Orchestrate Cian history backfill with heartbeat + checkpoint.

    Wraps backfill_cian_history(), updating scrape_runs counters (via update_heartbeat)
    before and after the batch call for zombie-detection visibility.

    Checkpoint/resume semantics: backfill_cian_history() queries rows WHERE history IS
    NULL via LEFT JOIN — so re-running after a partial completion naturally skips
    already-processed rows (idempotent by design).

    Params (from default_params jsonb):
        batch_size: int — rows per run (listings + houses counted separately).
    """
    from app.tasks.cian_history_backfill import backfill_cian_history

    batch_size = int(params.get("batch_size", 100))
    counters: dict[str, int] = {
        "listings_processed": 0,
        "listings_succeeded": 0,
        "listings_failed": 0,
        "houses_processed": 0,
        "houses_succeeded": 0,
        "houses_failed": 0,
    }

    try:
        runs_mod.update_heartbeat(db, run_id, counters)

        result = await backfill_cian_history(
            db,
            batch_size=batch_size,
            do_listings=True,
            do_houses=True,
            do_valuations=False,
        )

        counters = {
            "listings_processed": result.listings_processed,
            "listings_succeeded": result.listings_succeeded,
            "listings_failed": result.listings_failed_fetch + result.listings_failed_save,
            "houses_processed": result.houses_processed,
            "houses_succeeded": result.houses_succeeded,
            "houses_failed": result.houses_failed_fetch + result.houses_failed_save,
            "duration_sec": int(result.duration_sec),
        }
        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "scheduler: cian_history_backfill run_id=%d done — listings=%d/%d houses=%d/%d %.1fs",
            run_id,
            result.listings_succeeded,
            result.listings_total,
            result.houses_succeeded,
            result.houses_total,
            result.duration_sec,
        )
    except Exception as exc:
        logger.exception("scheduler: cian_history_backfill run_id=%d failed", run_id)
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise


async def trigger_yandex_address_backfill_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch run_yandex_address_backfill в asyncio.create_task.

    EKB-only pilot (#855): backfills listings.address for yandex listings that have
    street-only addresses (no house number). Fetches detail pages via curl_cffi chrome120,
    extracts full address from <title>, updates listings WHERE address differs.

    Mirrors trigger_cian_backfill_run: claim run → create_task → mark_done/failed.
    executor logic lives in tasks/yandex_address_backfill.py which delegates to
    services/yandex_address_backfill.py (backfill_yandex_addresses).

    Returns run_id или None (skip — already running).
    """
    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.yandex_address_backfill import run_yandex_address_backfill

            await run_yandex_address_backfill(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_yandex_address_backfill crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered yandex_address_backfill run_id=%d", run_id)
    return run_id


async def trigger_geoportal_coords_backfill_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch run_geoportal_coords_backfill в executor (sync DB-only task).

    Exact street+house match против ekb_geoportal_buildings (#1967 — was manual-only #1841).
    Pure local DB op — никакого внешнего HTTP/rate-limit, поэтому запускается ПЕРЕД
    geocode_missing_listings (Nominatim): точный house-level матч должен получить шанс
    первым, иначе более грубый (coarse) geocode Nominatim'а забирает адрес раньше и
    listing выпадает из WHERE lat IS NULL до следующего цикла.

    Sync task (SELECT + per-row local lookup + UPDATE, no async HTTP) — run in run_in_executor
    by the same pattern as trigger_cadastral_geo_match_run. run_geoportal_coords_backfill owns
    the scrape_runs lifecycle (mark_done/mark_failed).

    Returns run_id или None (skip — already running).
    """
    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.backfill_listings_coords_geoportal import (
                run_geoportal_coords_backfill,
            )

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda: run_geoportal_coords_backfill(run_db, run_id=run_id, params=params),
            )
        except Exception:
            logger.exception("scheduler: run_geoportal_coords_backfill crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered geoportal_coords_backfill run_id=%d", run_id)
    return run_id


async def trigger_geocode_missing_listings_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch run_geocode_missing_listings в asyncio.create_task.

    Nightly backfill listings.lat/lon для всех источников с NULL coords (#1: geom-coverage gap).
    Адрес-dedup batch loop с wall-clock budget (default 30 мин).

    Root cause: scrapers сохраняют listings с lat=lon=NULL — «геокод подтянет cron»,
    но cron геокодирует только deals, не listings. Эта функция замыкает петлю: нигде
    снаружи нет OS cron для listings — только этот планировщик.

    Mirrors trigger_yandex_address_backfill_run: claim run → create_task → mark_done/failed.

    Returns run_id или None (skip — already running).
    """
    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.geocode_missing import run_geocode_missing_listings

            await run_geocode_missing_listings(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_geocode_missing_listings crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered geocode_missing_listings run_id=%d", run_id)
    return run_id


async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task)."""
    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:
            # import_rosreestr_dkp is synchronous (DB-only, no async HTTP)
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, import_rosreestr_dkp, run_db, run_id, params)
        except Exception:
            logger.exception("scheduler: import_rosreestr_dkp crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered rosreestr_dkp_import run_id=%d", run_id)
    return run_id


async def trigger_listing_source_snapshot_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch snapshot_listing_sources в executor (sync DB-only task).

    Per-source price-history daily snapshot (#570). Mirror trigger_rosreestr_dkp_run:
    задача синхронная (чистый internal DB-snapshot, БЕЗ внешних HTTP-вызовов / анти-бота),
    поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (079),
    в отличие от avito/yandex sweep (dormant).

    Returns run_id (или None если skip — есть running run).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            # snapshot_listing_sources is synchronous (DB-only, no async HTTP)
            from app.tasks.listing_source_snapshot import snapshot_listing_sources

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, snapshot_listing_sources, run_db, run_id)
        except Exception:
            logger.exception("scheduler: snapshot_listing_sources crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered listing_source_snapshot run_id=%d", run_id)
    return run_id


async def trigger_refresh_search_matview_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + запустить REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv.

    listings_search_mv активно читается search_query.py (SELECT FROM listings_search_mv).
    Без периодического refresh матвью устаревает по мере появления новых listings.
    Задача синхронная (DB-only, БЕЗ внешних HTTP-вызовов) — гоняем в run_in_executor
    по образцу trigger_asking_to_sold_ratio_run (#648).

    refresh_search_matview() использует собственное psycopg-соединение с autocommit=True
    (REFRESH CONCURRENTLY не может работать внутри транзакции). scrape_runs lifecycle
    (mark_done/mark_failed) управляется через отдельную SessionLocal.

    SAFE to enable — schedule seed enabled=true (088), pure internal DB op.
    Окно 03:00-04:00 UTC — ночное, до rosreestr_dkp_import (04:00-06:00 UTC).

    Returns run_id (или None если skip — есть running run).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            from app.tasks.refresh_search_matview import refresh_search_matview

            loop = asyncio.get_event_loop()
            # refresh_search_matview() manages its own connection with autocommit=True
            # (required for REFRESH MATERIALIZED VIEW CONCURRENTLY outside a transaction).
            await loop.run_in_executor(None, refresh_search_matview)
            runs_mod.mark_done(run_db, run_id, {})
        except Exception:
            logger.exception("scheduler: refresh_search_matview crashed run_id=%d", run_id)
            runs_mod.mark_failed(run_db, run_id, "refresh_search_matview failed", {})
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered refresh_search_matview run_id=%d", run_id)
    return run_id


async def trigger_deactivate_stale_avito_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch deactivate_stale_avito_listings в executor (sync DB-only task).

    Помечает avito-объявления, не виденные >TTL дней, как is_active=false (#759).
    Задача синхронная (чистый internal UPDATE, БЕЗ внешних HTTP-вызовов) — гоняем в
    run_in_executor по образцу trigger_listing_source_snapshot_run. SAFE to enable —
    schedule seed enabled=true (090), pure internal DB op.
    Окно 06:00-07:00 UTC — после avito_city_sweep (02:00-05:00 UTC).

    Returns run_id (или None если skip — есть running run).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            from app.tasks.deactivate_stale_avito import deactivate_stale_avito_listings

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, deactivate_stale_avito_listings, run_db, run_id)
        except Exception:
            logger.exception("scheduler: deactivate_stale_avito crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered deactivate_stale_avito run_id=%d", run_id)
    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, staleness_column из
    schedule_row["default_params"] (jsonb).
    Для источника 'deactivate_stale_avito' с пустым default_params использует defaults:
      listing_source='avito', ttl_days=settings.avito_stale_ttl_days, segments=None,
      staleness_column='last_seen_at'.
    Для yandex/cian -- default_params обязан содержать listing_source + ttl_days + segments.
    Для domklik (#2204) -- staleness_column='scraped_at' (нетрекаемый bulk-touch
    двигает last_seen_at, честная свежесть = scraped_at).

    Запускает 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
    staleness_column: str = params.get("staleness_column", "last_seen_at")

    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,
                    staleness_column=staleness_column,
                ),
            )
        except Exception:
            logger.exception(
                "scheduler: deactivate_stale source=%s crashed run_id=%d",
                listing_source,
                run_id,
            )
        finally:
            run_db.close()

    _spawn_tracked(_run())
    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.

    Monthly pull (#887): fetches city-level secondary-housing price index from
    sberindex.ru/api/sowa across three dashboards (real_estate_deals,
    residential_real_estate_prices, dinamika-tsen-obyavlenii) and upserts into
    sber_price_index. Pure HTTP+DB, no scraper proxy, no anti-bot.

    Mirrors trigger_yandex_address_backfill_run: claim run → create_task → mark_done/failed
    delegated to tasks/sber_index_pull.py.

    Returns run_id или None (skip — already running).
    """
    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.sber_index_pull import run_sber_index_pull

            await run_sber_index_pull(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_sber_index_pull crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered sber_index_pull run_id=%d", run_id)
    return run_id


async def trigger_rosreestr_quarter_poll_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch run_rosreestr_quarter_poll в asyncio.create_task.

    Monthly HEAD-check (#888): polls rosreestr.gov.ru for the next quarter's open-data
    ZIP to detect when a new quarter is published. Pure HTTP detect+alert — does NOT
    download the multi-GB dataset or shell out to the loaders. When available=True, a
    clear actionable INFO log is emitted for the operator.

    Mirrors trigger_sber_index_pull_run: claim run → create_task → mark_done/failed
    delegated to tasks/rosreestr_quarter_poll.py.

    Returns run_id или None (skip — already running).
    """
    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.rosreestr_quarter_poll import run_rosreestr_quarter_poll

            await run_rosreestr_quarter_poll(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_rosreestr_quarter_poll crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered rosreestr_quarter_poll run_id=%d", run_id)
    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).

    Nightly slice of the cian_newbuilding enrichment backfill: resolves cian_zhk_url from
    house_sources.ext_id when missing, fetches the ЖК page (curl_cffi chrome120) and persists
    houses_price_dynamics / house_reliability_checks / house_reviews. This is the standalone
    schedule the inline cian full-sweep path never had — 306 fetchable cian houses on prod,
    only ~3 enriched so far, so the backlog drains over successive nights.

    Mirrors trigger_sber_index_pull_run: claim run → create_task → mark_done/failed delegated
    to tasks/newbuilding_enrich_backfill.run_newbuilding_enrich (which owns the run lifecycle).

    Idempotency: run_newbuilding_enrich → backfill_newbuilding_enrichment skips houses that
    already have BOTH price_dynamics AND reliability rows (force=False) and uses a per-house
    SAVEPOINT, so a re-fire never re-enriches done houses or double-appends on a partial run.
    `limit` (default_params) bounds how many houses one fire fetches — anti-bot politeness;
    the nightly cadence drains the rest. zombie-reap is inherited from the shared loop.

    Returns run_id или None (skip — already running).
    """
    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.newbuilding_enrich_backfill import run_newbuilding_enrich

            await run_newbuilding_enrich(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_newbuilding_enrich crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered newbuilding_enrich run_id=%d", run_id)
    return run_id


async def trigger_yandex_newbuilding_sweep_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch enrich_yandex_newbuilding_sweep в asyncio.create_task (#974).

    Nightly enrichment sweep: SELECT pending yandex_realty_nb houses → resolve slug
    (BrowserFetcher SERP) → fetch_jk (BrowserFetcher) → UPSERT market.yandex_jk_enrichment.

    Shipped DORMANT (seed 106, enabled=false). Включается оператором вручную:
        UPDATE scrape_schedules SET enabled = true WHERE source = 'yandex_newbuilding_sweep';

    Mirrors trigger_newbuilding_enrich_run: claim run → create_task → mark_done/failed
    делегируется tasks/yandex_newbuilding_sweep.enrich_yandex_newbuilding_sweep.

    Returns run_id или None (skip — already running).
    """
    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.yandex_newbuilding_sweep import enrich_yandex_newbuilding_sweep

            result = await enrich_yandex_newbuilding_sweep(
                run_db,
                limit=int(params.get("limit", 5)),
                request_delay_sec=float(params.get("request_delay_sec", 8.0)),
                city=str(params.get("city", "ekaterinburg")),
            )
            runs_mod.mark_done(run_db, run_id, result.to_dict())
        except Exception:
            logger.exception("scheduler: enrich_yandex_newbuilding_sweep crashed run_id=%d", run_id)
            try:
                runs_mod.mark_failed(run_db, run_id, "crashed", {})
            except Exception:
                pass
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered yandex_newbuilding_sweep run_id=%d", run_id)
    return run_id


async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).

    Daily TRUE-MIRROR refresh асking→sold коэффициентов (#648 Stage 4): DELETE WHERE
    district='' + re-derive INSERT (та же 080-derivation) в одной транзакции, чтобы ratio
    не устаревал по мере ночного импорта ДКП. Mirror trigger_listing_source_snapshot_run:
    задача синхронная (чистый internal DB re-derive, БЕЗ внешних HTTP-вызовов / анти-бота),
    поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (082),
    в отличие от avito/yandex sweep (dormant).

    Returns run_id (или None если skip — есть running run).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            # recompute_asking_to_sold_ratios is synchronous (DB-only, no async HTTP)
            from app.tasks.asking_to_sold_ratio import recompute_asking_to_sold_ratios

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, recompute_asking_to_sold_ratios, run_db, run_id)
        except Exception:
            logger.exception("scheduler: recompute_asking_to_sold_ratios crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered asking_to_sold_ratio_refresh run_id=%d", run_id)
    return run_id


def import_rosreestr_dkp(
    db: Session,
    run_id: int,
    params: dict[str, Any],
) -> None:
    """Import ДКП-сделок из gendesign rosreestr_deals через postgres_fdw.

    Python-порт import-rosreestr.sh (Variant C из #563).

    Источник: foreign table gendesign_rosreestr_deals (создана в migration 072).
    SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql.
    USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader).

    Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
      - region_code = 66 (Свердловская область)
      - city ILIKE '%катеринбург%'
      - realestate_type_code = '002001003000' (квартира)
      - area BETWEEN 18 AND 200
      - deal_price BETWEEN 1000000 AND 100000000
      - street IS NOT NULL AND trim(street) != ''
      - doc_type = 'ДКП'  (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24)
      - period_start_date >= since  (default '2024-01-01')

    dedup_hash: 'ros:dkp:' || id — плоский натуральный ключ (инъективный, без коллизий,
      human-readable). До #576 здесь был md5('ros:dkp:' || id); миграция 077 конвертировала
      существующие строки. source_id хранит исходный rosreestr id (дедуп переустанавливаем).
    Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат).

    Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию
    (WHERE id > last_id ORDER BY id). Heartbeat обновляется каждый батч (= checkpoint).
    SAVEPOINT per row — один сбойный row не откатывает батч.

    Координаты: NULL после импорта — геокодинг остаётся follow-up (geocode-deals).
    TODO (follow-up): запустить geocode backfill после import.

    Cleanup: удаляет legacy строки address='Екатеринбург, реальная сделка' (pre-#549).
    """
    since: str = str(params.get("since", "2024-01-01"))
    batch_size: int = int(params.get("batch_size", 2000))

    counters: dict[str, int] = {
        "rows_fetched": 0,
        "rows_inserted": 0,
        "rows_skipped": 0,
        "batches_done": 0,
    }

    # Cleanup legacy synthetic rows (pre-#549, idempotent)
    try:
        deleted = db.execute(
            text("DELETE FROM deals WHERE address = CAST(:addr AS text) RETURNING id"),
            {"addr": "Екатеринбург, реальная сделка"},
        ).fetchall()
        db.commit()
        if deleted:
            logger.info(
                "rosreestr_dkp_import run_id=%d: removed %d legacy synthetic rows",
                run_id,
                len(deleted),
            )
    except Exception as exc:
        logger.warning(
            "rosreestr_dkp_import run_id=%d: cleanup failed (non-fatal): %s", run_id, exc
        )
        db.rollback()

    last_id: int = 0
    total_batches = 0

    try:
        while True:
            cancelled = runs_mod.is_cancelled(db, run_id)
            if cancelled or shutdown_requested():
                if cancelled:
                    # User-cancel: семантика без изменений — mark_cancelled.
                    logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id)
                    runs_mod.mark_cancelled(db, run_id)
                    return
                # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
                # Это НЕ user-cancel → mark_done (partial), не mark_cancelled. Курсор
                # уже зафиксирован update_heartbeat'ом каждый батч; следующий run
                # пере-сканирует с id=0 (INSERT ... ON CONFLICT DO NOTHING идемпотентен)
                # → уже вставленные строки пропускаются, остаток до-импортируется.
                # mark_done выводит run из 'running' → reap_zombies его не тронет.
                logger.info(
                    "rosreestr_dkp_import run_id=%d: SIGTERM-drain — committing partial "
                    "(last_id=%d, batches=%d) and exiting",
                    run_id,
                    last_id,
                    total_batches,
                )
                runs_mod.update_heartbeat(db, run_id, counters)
                runs_mod.mark_done(db, run_id, counters)
                return

            # Cursor-based pagination via foreign table gendesign_rosreestr_deals.
            # FDW pushes WHERE + ORDER BY + LIMIT to gendesign-postgres automatically
            # (postgres_fdw 'use_remote_estimate' is off by default but clause push-down
            # still happens for simple predicates on the remote side).
            batch_rows = (
                db.execute(
                    text("""
                    SELECT
                        id,
                        id                                       AS source_id_src,
                        'ros:dkp:' || CAST(id AS text)           AS dedup_hash,
                        'Екатеринбург, ' || trim(street)         AS address,
                        CASE
                            WHEN area < 30 THEN 0
                            WHEN area < 44 THEN 1
                            WHEN area < 62 THEN 2
                            WHEN area < 85 THEN 3
                            ELSE 4
                        END                                      AS rooms,
                        round(area, 2)                           AS area_m2,
                        LEAST(
                            NULLIF(
                                -- #1525: '-?[0-9]+' сохраняет знак минус, иначе цоколь/подвал
                                -- '-1' импортируется как 1. Для '5/9' по-прежнему берётся 5.
                                substring(floor FROM '-?[0-9]+'), ''
                            )::int,
                            100
                        )                                        AS floor_num,
                        year_build                               AS year_built,
                        round(deal_price)::bigint                AS price_rub,
                        round(price_per_sqm)::int                AS price_per_m2,
                        period_start_date                        AS deal_date
                    FROM gendesign_rosreestr_deals
                    WHERE region_code = 66
                      AND city ILIKE '%катеринбург%'
                      AND realestate_type_code = '002001003000'
                      AND area BETWEEN 18 AND 200
                      AND deal_price BETWEEN 1000000 AND 100000000
                      AND street IS NOT NULL AND trim(street) <> ''
                      AND doc_type = 'ДКП'
                      AND period_start_date >= CAST(:since AS date)
                      AND id > CAST(:last_id AS bigint)
                    ORDER BY id
                    LIMIT CAST(:batch_size AS int)
                """),
                    {"since": since, "last_id": last_id, "batch_size": batch_size},
                )
                .mappings()
                .all()
            )

            if not batch_rows:
                break

            total_batches += 1
            batch_inserted = 0
            batch_skipped = 0
            batch_max_id = last_id

            for row in batch_rows:
                row_id: int = int(row["id"])
                if row_id > batch_max_id:
                    batch_max_id = row_id

                try:
                    with db.begin_nested():  # SAVEPOINT per row
                        inserted = db.execute(
                            text("""
                                INSERT INTO deals (
                                    source, dedup_hash, source_id, address, rooms, area_m2,
                                    floor, year_built, price_rub, price_per_m2, deal_date
                                )
                                VALUES (
                                    'rosreestr',
                                    CAST(:dedup_hash AS text),
                                    CAST(:source_id AS text),
                                    CAST(:address AS text),
                                    CAST(:rooms AS int),
                                    CAST(:area_m2 AS numeric),
                                    CAST(:floor_num AS int),
                                    CAST(:year_built AS int),
                                    CAST(:price_rub AS bigint),
                                    CAST(:price_per_m2 AS int),
                                    CAST(:deal_date AS date)
                                )
                                ON CONFLICT (dedup_hash) DO NOTHING
                                RETURNING id
                            """),
                            {
                                "dedup_hash": row["dedup_hash"],
                                "source_id": str(row["source_id_src"]),
                                "address": row["address"],
                                "rooms": row["rooms"],
                                "area_m2": row["area_m2"],
                                "floor_num": row["floor_num"],
                                "year_built": row["year_built"],
                                "price_rub": row["price_rub"],
                                "price_per_m2": row["price_per_m2"],
                                "deal_date": row["deal_date"],
                            },
                        ).fetchone()
                        if inserted is not None:
                            batch_inserted += 1
                        else:
                            batch_skipped += 1
                except Exception as exc:
                    logger.warning(
                        "rosreestr_dkp_import run_id=%d: row id=%d INSERT failed: %s",
                        run_id,
                        row_id,
                        exc,
                    )
                    batch_skipped += 1

            last_id = batch_max_id
            db.commit()

            counters["rows_fetched"] += len(batch_rows)
            counters["rows_inserted"] += batch_inserted
            counters["rows_skipped"] += batch_skipped
            counters["batches_done"] = total_batches
            counters["last_id"] = last_id  # type: ignore[assignment]

            # Heartbeat = checkpoint: allows zombie detection + resume visibility
            runs_mod.update_heartbeat(db, run_id, counters)
            logger.info(
                "rosreestr_dkp_import run_id=%d: batch=%d fetched=%d "
                "inserted=%d skipped=%d last_id=%d",
                run_id,
                total_batches,
                len(batch_rows),
                batch_inserted,
                batch_skipped,
                last_id,
            )

            if len(batch_rows) < batch_size:
                break  # Last partial batch — no more rows

        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "rosreestr_dkp_import run_id=%d done: "
            "total_fetched=%d inserted=%d skipped=%d batches=%d",
            run_id,
            counters["rows_fetched"],
            counters["rows_inserted"],
            counters["rows_skipped"],
            total_batches,
        )

    except Exception as exc:
        logger.exception("rosreestr_dkp_import run_id=%d failed at last_id=%d", run_id, last_id)
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise


async def trigger_avito_detail_backfill_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Create scrape_run + launch run_avito_detail_backfill in asyncio.create_task.

    Nightly backfill detail_enriched_at for legacy avito listings (#1551).
    ~15243/15917 avito listings have detail_enriched_at IS NULL -- never enriched
    by city_sweep ENRICH_DETAIL (only top-N proximate per anchor are touched).

    Mirrors trigger_geocode_missing_listings_run: claim run -> create_task ->
    mark_done/failed delegated to tasks/avito_detail_backfill.run_avito_detail_backfill.

    Returns run_id or None (skip -- already running).
    """
    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.avito_detail_backfill import run_avito_detail_backfill

            await run_avito_detail_backfill(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_avito_detail_backfill crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered avito_detail_backfill run_id=%d", run_id)
    return run_id


async def trigger_yandex_detail_backfill_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Create scrape_run + launch run_yandex_detail_backfill in asyncio.create_task.

    Nightly backfill detail_enriched_at for legacy yandex listings (#1553).
    ~3952 yandex listings have detail_enriched_at IS NULL -- SERP-only data,
    never enriched by YandexDetailScraper.

    Mirrors trigger_avito_detail_backfill_run: claim run -> create_task ->
    mark_done/failed delegated to tasks/yandex_detail_backfill.run_yandex_detail_backfill.

    Returns run_id or None (skip -- already running).
    """
    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.yandex_detail_backfill import run_yandex_detail_backfill

            await run_yandex_detail_backfill(run_db, run_id=run_id, params=params)
        except Exception:
            logger.exception("scheduler: run_yandex_detail_backfill crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered yandex_detail_backfill run_id=%d", run_id)
    return run_id


async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_cadastral_geo_match в executor (sync DB-only task).

    Combined refresh+match (#cadastral-geo-match): (1) REFRESH cad_buildings_local from the
    gendesign_cad_buildings FDW (one bulk scan, builds Point geom), (2) MATCH set-based
    LATERAL KNN UPDATE filling listings.building_cadastral_number from the nearest cadastral
    building within threshold_m (default 50m). GEO-NEAREST approximation.

    Sync task (one FDW read + local UPDATE, no async HTTP) — run in run_in_executor by the
    same pattern as trigger_listing_source_snapshot_run / trigger_asking_to_sold_ratio_run.
    run_cadastral_geo_match owns the scrape_runs lifecycle (mark_done/mark_failed). SAFE to
    enable — seed 125 enabled=true, pure internal DB op.

    Returns run_id или None (skip — already running).
    """
    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.cadastral_geo_match import run_cadastral_geo_match

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda: run_cadastral_geo_match(run_db, run_id=run_id, params=params),
            )
        except Exception:
            logger.exception("scheduler: run_cadastral_geo_match crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered cadastral_geo_match run_id=%d", run_id)
    return run_id


async def trigger_osm_poi_ekb_refresh_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_osm_poi_ekb_refresh в executor (#2045 BE-3).

    Single bulk FDW scan (TRUNCATE osm_poi_ekb_local; INSERT ... FROM gendesign_osm_poi_ekb) —
    feeds app/services/location_coef.py. No params consumed.

    Sync task (one FDW read + local TRUNCATE+INSERT, no async HTTP) — run in run_in_executor
    by the same pattern as trigger_cadastral_geo_match_run. run_osm_poi_ekb_refresh owns the
    scrape_runs lifecycle (mark_done/mark_failed). SAFE to enable — seed 170 enabled=true,
    pure internal DB op.

    Returns run_id или None (skip — already running).
    """
    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.osm_poi_ekb_refresh import run_osm_poi_ekb_refresh

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda: run_osm_poi_ekb_refresh(run_db, run_id=run_id, params=params),
            )
        except Exception:
            logger.exception("scheduler: run_osm_poi_ekb_refresh crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered osm_poi_ekb_refresh run_id=%d", run_id)
    return run_id


async def trigger_house_imv_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch backfill_house_imv в asyncio.create_task (#854).

    Bulk Avito IMV-оценка на уровне домов: итерирует houses с imv_status=:only_status
    (default 'pending'), у которых есть lat/lon + хотя бы один listing с rooms+area,
    считает медианные lot-params, зовёт evaluate_via_imv и наполняет
    house_imv_evaluations + house_placement_history + house_suggestions (3×UPSERT),
    переводя houses.imv_status pending→ok/not_found/error/transient_error.

    Зеркало trigger_avito_detail_backfill_run / trigger_yandex_address_backfill_run:
    _claim_run → asyncio.create_task → fresh SessionLocal → add_done_callback (RUF006).
    Отличие: у house_imv_backfill нет tasks/*-обёртки, владеющей lifecycle, поэтому
    финализацию scrape_runs (mark_done/mark_failed) делаем здесь, как в
    trigger_refresh_search_matview_run. heartbeat-колбэк дёргает update_heartbeat,
    чтобы reap_zombies не пометил живой долгий IMV-run 'zombie' (#1363).

    Resumable + idempotent: backfill_house_imv фильтрует imv_status и UPSERT'ит по
    house_id, поэтому повторный fire дренит следующий batch без дублей.

    Returns run_id или None (skip — already running).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    params = schedule_row.get("default_params") or {}
    batch_size = int(params.get("batch_size", 50))
    request_delay_sec = float(params.get("request_delay_sec", 5.0))
    only_status = str(params.get("only_status", "pending"))

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            from app.services.house_imv_backfill import backfill_house_imv

            # heartbeat дёргается каждые N домов внутри backfill — обновляем
            # scrape_runs.heartbeat_at, иначе reap_zombies снимет живой run (#1363).
            def _heartbeat() -> None:
                runs_mod.update_heartbeat(run_db, run_id, {})

            result = await backfill_house_imv(
                run_db,
                batch_size=batch_size,
                request_delay_sec=request_delay_sec,
                only_status=only_status,
                heartbeat=_heartbeat,
            )
            runs_mod.mark_done(
                run_db,
                run_id,
                {
                    "checked": result.checked,
                    "saved": result.saved,
                    "skipped": result.skipped,
                    "errors": result.errors,
                    "duration_sec": int(result.duration_sec),
                },
            )
        except Exception as exc:
            logger.exception("scheduler: backfill_house_imv crashed run_id=%d", run_id)
            try:
                runs_mod.mark_failed(run_db, run_id, str(exc)[:1000], {})
            except Exception:
                logger.exception("scheduler: mark_failed crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered house_imv_backfill run_id=%d", run_id)
    return run_id


async def trigger_house_dedup_merge_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_house_dedup_merge в executor (sync DB-only task, #1772).

    Recurring, idempotent, collision-safe house-deduplication merge: clusters `houses` by
    EXACT address, picks one canonical keeper per cluster (geom NOT NULL → most listings →
    most-populated → min id), re-points all FK children with UNIQUE-collision handling
    (reuses migration 108's proven pattern), deletes the loser rows, backfills
    house_sources/house_address_aliases onto the keeper — all in ONE transaction.

    DESTRUCTIVE (DELETEs duplicate house rows) but idempotent: a clean table yields an empty
    mapping → every statement is a 0-row no-op. Seed 135 ships the schedule DISABLED
    (enabled=false) — the deploy is neutral; the orchestrator dry-runs + does one validated
    manual run, then enables deliberately. default_params.dry_run=true previews without writes.

    Sync task (set-based UPDATE/DELETE, no async HTTP) — run in run_in_executor by the same
    pattern as trigger_cadastral_geo_match_run. run_house_dedup_merge owns the scrape_runs
    lifecycle (mark_done/mark_failed).

    Returns run_id или None (skip — already running).
    """
    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.services.house_dedup_merge import run_house_dedup_merge

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda: run_house_dedup_merge(run_db, run_id=run_id, params=params),
            )
        except Exception:
            logger.exception("scheduler: run_house_dedup_merge crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered house_dedup_merge run_id=%d", run_id)
    return run_id


async def trigger_proxy_healthcheck_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать scrape_runs + launch run_proxy_healthcheck в asyncio.create_task (#2162).

    Периодический health-check пула scrape_proxies: reap_stale_leases + ipify-проба через
    каждый enabled-прокси → mark_health (сброс fails/exit_ip/latency при успехе; инкремент
    + авто-disable при DISABLE_THRESHOLD на фейле). АДДИТИВНО — не трогает боевые скрейперы.

    Зеркало trigger_yandex_newbuilding_sweep_run: у run_proxy_healthcheck нет tasks/*-обёртки,
    владеющей lifecycle, поэтому финализацию scrape_runs (mark_done/mark_failed) делаем здесь.

    Каденс: _claim_run ставит next_run_at через compute_next_run_at (суточная гранулярность),
    но health-check хочется чаще. Поэтому СРАЗУ после claim переопределяем next_run_at на
    now() + interval_minutes (default 30) — source-специфичный sub-hourly re-schedule, не
    трогая shared compute_next_run_at. Если run ещё идёт на следующем тике — has_running_run
    в _claim_run вернёт None (skip) и next_run_at не сбросится, т.е. авто-throttle по факту.

    Returns run_id или None (skip — already running).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    params = schedule_row.get("default_params") or {}
    interval_minutes = int(params.get("interval_minutes", 30))
    db.execute(
        text(
            """
            UPDATE scrape_schedules
            SET next_run_at = now() + make_interval(mins => CAST(:mins AS integer)),
                updated_at = now()
            WHERE source = 'proxy_healthcheck'
            """
        ),
        {"mins": interval_minutes},
    )
    db.commit()

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            from app.services.proxy_pool import run_proxy_healthcheck

            counters = await run_proxy_healthcheck(run_db)
            runs_mod.mark_done(run_db, run_id, counters)
        except Exception as exc:
            logger.exception("scheduler: run_proxy_healthcheck crashed run_id=%d", run_id)
            try:
                runs_mod.mark_failed(run_db, run_id, str(exc)[:1000], {})
            except Exception:
                logger.exception("scheduler: mark_failed crashed run_id=%d", run_id)
        finally:
            run_db.close()

    _spawn_tracked(_run())
    logger.info("scheduler: triggered proxy_healthcheck run_id=%d", run_id)
    return run_id


def get_due_schedules(db: Session) -> list[dict[str, Any]]:
    """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
    rows = (
        db.execute(
            text(
                """
            SELECT id, source, enabled, window_start_hour, window_end_hour,
                   default_params, last_run_id, last_run_at, next_run_at
            FROM scrape_schedules
            WHERE enabled = true
              AND (next_run_at IS NULL OR next_run_at <= NOW())
            """
            ),
        )
        .mappings()
        .all()
    )
    return [dict(r) for r in rows]


async def scheduler_loop() -> None:
    """Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд."""
    logger.info("scheduler: started (tick=%ds)", SCHEDULER_TICK_SEC)
    # Initial sleep 30s чтобы дать FastAPI startup завершиться
    await asyncio.sleep(30)
    while True:
        # #1182 Phase 2: кооперативный SIGTERM-drain. Не reap'аем и не claim'аем
        # новые run'ы во время shutdown — даём текущему dispatch'у докатиться и выходим.
        if shutdown_requested():
            logger.info("scheduler: SIGTERM-drain — stop claiming new runs, exiting tick loop")
            break
        try:
            db = SessionLocal()
            try:
                # Reap zombies first
                reap_zombies(db)
                # Process due schedules
                due = get_due_schedules(db)
                for sch in due:
                    source = sch["source"]
                    if source == "avito_city_sweep":
                        await trigger_avito_city_sweep_run(db, sch)
                    elif source == "avito_full_load":
                        await trigger_avito_full_load_run(db, sch)
                    elif source == "avito_full_load_exhaustive":
                        await trigger_avito_full_load_exhaustive_run(db, sch)
                    elif source == "avito_newbuilding_sweep":
                        await trigger_avito_newbuilding_sweep_run(db, sch)
                    elif source == "yandex_city_sweep":
                        await trigger_yandex_city_sweep_run(db, sch)
                    elif source == "cian_history_backfill":
                        await trigger_cian_backfill_run(db, sch)
                    elif source == "cian_city_sweep":
                        await trigger_cian_city_sweep_run(db, sch)
                    elif source == "cian_full_load":
                        await trigger_cian_full_load_run(db, sch)
                    elif source == "domclick_city_sweep":
                        await trigger_domclick_city_sweep_run(db, sch)
                    elif source == "rosreestr_dkp_import":
                        await trigger_rosreestr_dkp_run(db, sch)
                    elif source == "listing_source_snapshot":
                        await trigger_listing_source_snapshot_run(db, sch)
                    elif source == "asking_to_sold_ratio_refresh":
                        await trigger_asking_to_sold_ratio_run(db, sch)
                    elif source == "refresh_search_matview":
                        await trigger_refresh_search_matview_run(db, sch)
                    elif source == "yandex_address_backfill":
                        await trigger_yandex_address_backfill_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":
                        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":
                        await trigger_yandex_newbuilding_sweep_run(db, sch)
                    elif source == "geoportal_coords_backfill":
                        await trigger_geoportal_coords_backfill_run(db, sch)
                    elif source == "geocode_missing_listings":
                        await trigger_geocode_missing_listings_run(db, sch)
                    elif source == "avito_detail_backfill":
                        await trigger_avito_detail_backfill_run(db, sch)
                    elif source == "yandex_detail_backfill":
                        await trigger_yandex_detail_backfill_run(db, sch)
                    elif source == "cadastral_geo_match":
                        await trigger_cadastral_geo_match_run(db, sch)
                    elif source == "osm_poi_ekb_refresh":
                        await trigger_osm_poi_ekb_refresh_run(db, sch)
                    elif source == "house_imv_backfill":
                        await trigger_house_imv_backfill_run(db, sch)
                    elif source == "house_dedup_merge":
                        await trigger_house_dedup_merge_run(db, sch)
                    elif source == "proxy_healthcheck":
                        await trigger_proxy_healthcheck_run(db, sch)
                    else:
                        logger.warning("scheduler: unknown source=%s, skip", source)

                    # #1182 Phase 2: после каждого dispatch'а проверяем drain — текущий
                    # run уже отпущен в свою asyncio-задачу (сам докоммитит/выйдет по
                    # своему checkpoint'у), а новые в этом тике не запускаем.
                    if shutdown_requested():
                        logger.info(
                            "scheduler: SIGTERM-drain — stop dispatch after source=%s", source
                        )
                        break
            finally:
                db.close()
        except Exception:
            logger.exception("scheduler: tick failed")
        # Промежуточная проверка перед 60s-сном: выходим сразу, не ждём целый тик.
        if shutdown_requested():
            logger.info("scheduler: SIGTERM-drain — exiting tick loop after dispatch")
            break
        await asyncio.sleep(SCHEDULER_TICK_SEC)

    # Tick-loop вышел только по SIGTERM-drain'у (иначе while True бесконечен): дожидаемся
    # detached run-задач (_spawn_tracked) — пусть докоммитят текущий unit и сделают
    # mark_done, а не дадим asyncio.run() teardown'у хард-кансельнуть их mid-await (#1182).
    await _drain_inflight()
    logger.info("scheduler: tick loop exited (in-flight drain complete)")