feat(scraper-kit): copy scheduler with registry-dispatch (25+ trigger_* → dict), strangler (#2136)

tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py

@@ -0,0 +1,471 @@
+"""Golden-parity: `scraper_kit.orchestration.scheduler` registry-dispatch ≡ старый if/elif.
+
+Strangler-инвариант (#2136): registry-реестр (`build_registry` + `resolve_handler` +
+`_dispatch`) должен маршрутизировать те же source'ы к тем же обработчикам, что и боевой
+27-веточный `if/elif` в `app.services.scheduler.scheduler_loop`, и сохранить критичную
+concurrency-логику (`_claim_run` advisory-lock + double-check, `reap_zombies` порог,
+proxy_healthcheck sub-hourly post-claim).
+
+Метод:
+  1. ROUTING: для каждого source прогоняем ОДИН тик боевого `scheduler_loop` (все
+     `trigger_*` замоканы) → фиксируем какой trigger сработал; параллельно `resolve_handler`
+     kit-реестра обязан вернуть handler для того же source. Множества покрытых source'ов
+     обязаны совпасть → таблица маршрутизации эквивалентна.
+  2. CLAIM: `_claim_run` — happy-path, already-running skip, lock-busy skip, appeared-under-
+     lock rollback. Мокаем БД + ctx.runs.
+  3. ZOMBIE: `reap_zombies` возвращает число + commit.
+  4. SUB-HOURLY: `reschedule_after_minutes` post_claim UPDATE + commit (proxy_healthcheck).
+  5. DISPATCH e2e: `_dispatch` для kit-native sweep (мок pipeline) и продуктового handler —
+     claim → свежая сессия → job вызван → close.
+
+Без сети, без БД.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from contextlib import ExitStack
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.orchestration import scheduler as kit_sched
+from scraper_kit.orchestration.scheduler import (
+    Handler,
+    SchedulerContext,
+    _claim_run,
+    _dispatch,
+    build_registry,
+    reap_zombies,
+    reschedule_after_minutes,
+    resolve_handler,
+)
+
+import app.services.scheduler as old_sched
+
+# ── таблица маршрутизации: source → имя боевого trigger_* ─────────────────────
+# (то, что `if/elif` в старом scheduler_loop выбирает для каждого source).
+SOURCE_TO_OLD_TRIGGER: dict[str, str] = {
+    "avito_city_sweep": "trigger_avito_city_sweep_run",
+    "avito_full_load": "trigger_avito_full_load_run",
+    "avito_full_load_exhaustive": "trigger_avito_full_load_exhaustive_run",
+    "avito_newbuilding_sweep": "trigger_avito_newbuilding_sweep_run",
+    "yandex_city_sweep": "trigger_yandex_city_sweep_run",
+    "cian_history_backfill": "trigger_cian_backfill_run",
+    "cian_city_sweep": "trigger_cian_city_sweep_run",
+    "cian_full_load": "trigger_cian_full_load_run",
+    "domclick_city_sweep": "trigger_domclick_city_sweep_run",
+    "rosreestr_dkp_import": "trigger_rosreestr_dkp_run",
+    "listing_source_snapshot": "trigger_listing_source_snapshot_run",
+    "asking_to_sold_ratio_refresh": "trigger_asking_to_sold_ratio_run",
+    "refresh_search_matview": "trigger_refresh_search_matview_run",
+    "yandex_address_backfill": "trigger_yandex_address_backfill_run",
+    "deactivate_stale_avito": "trigger_deactivate_stale_run",
+    "deactivate_stale_yandex": "trigger_deactivate_stale_run",
+    "deactivate_stale_cian": "trigger_deactivate_stale_run",
+    "sber_index_pull": "trigger_sber_index_pull_run",
+    "rosreestr_quarter_poll": "trigger_rosreestr_quarter_poll_run",
+    "newbuilding_enrich": "trigger_newbuilding_enrich_run",
+    "yandex_newbuilding_sweep": "trigger_yandex_newbuilding_sweep_run",
+    "geocode_missing_listings": "trigger_geocode_missing_listings_run",
+    "avito_detail_backfill": "trigger_avito_detail_backfill_run",
+    "yandex_detail_backfill": "trigger_yandex_detail_backfill_run",
+    "cadastral_geo_match": "trigger_cadastral_geo_match_run",
+    "house_imv_backfill": "trigger_house_imv_backfill_run",
+    "house_dedup_merge": "trigger_house_dedup_merge_run",
+    "proxy_healthcheck": "trigger_proxy_healthcheck_run",
+}
+
+# все distinct trigger-имена, которые надо замокать в боевом scheduler_loop
+_ALL_TRIGGER_NAMES = sorted(set(SOURCE_TO_OLD_TRIGGER.values()))
+
+# kit-native (тело в scraper_kit.orchestration.pipeline) — регистрируются встроенно
+_KIT_NATIVE_SOURCES = {
+    "avito_city_sweep",
+    "avito_full_load",
+    "avito_full_load_exhaustive",
+    "avito_newbuilding_sweep",
+    "yandex_city_sweep",
+    "cian_city_sweep",
+    "cian_full_load",
+    "domclick_city_sweep",
+}
+
+
+def _make_sched(source: str) -> dict[str, Any]:
+    return {
+        "id": 1,
+        "source": source,
+        "enabled": True,
+        "window_start_hour": 2,
+        "window_end_hour": 5,
+        "default_params": {},
+        "last_run_id": None,
+        "last_run_at": None,
+        "next_run_at": None,
+    }
+
+
+# ── kit-реестр с продуктовыми handler'ами (recording stubs) ──────────────────
+
+
+def _build_recording_registry() -> tuple[dict[str, Handler], dict[str, MagicMock]]:
+    """Kit-native sweeps + продуктовые recording-handler'ы. Возвращает (registry, fired-map)."""
+    fired: dict[str, MagicMock] = {}
+
+    def _make_job(name: str) -> Any:
+        rec = MagicMock(name=name)
+        fired[name] = rec
+
+        async def _job(db: Any, run_id: int, params: dict[str, Any], ctx: Any) -> None:
+            rec(db, run_id, params)
+
+        return _job
+
+    product: dict[str, Handler] = {}
+    # продуктовые exact-match источники
+    for src in SOURCE_TO_OLD_TRIGGER:
+        if src in _KIT_NATIVE_SOURCES or src.startswith("deactivate_stale_"):
+            continue
+        product[src] = Handler(_make_job(src), src)
+    # deactivate_stale_* — одна wildcard-запись (как боевой startswith)
+    product["deactivate_stale_*"] = Handler(_make_job("deactivate_stale"), "deactivate_stale")
+
+    return build_registry(product), fired
+
+
+# ── 1. ROUTING parity ────────────────────────────────────────────────────────
+
+
+async def _drive_old_one_tick(source: str) -> list[str]:
+    """Прогнать ОДИН тик боевого scheduler_loop для одного source, вернуть сработавшие trigger'ы."""
+    sch = _make_sched(source)
+    triggers = {name: AsyncMock(return_value=1) for name in _ALL_TRIGGER_NAMES}
+
+    with ExitStack() as es:
+        es.enter_context(patch.object(old_sched, "reap_zombies", MagicMock()))
+        es.enter_context(
+            patch.object(old_sched, "get_due_schedules", MagicMock(return_value=[sch]))
+        )
+        es.enter_context(patch.object(old_sched, "SessionLocal", MagicMock()))
+        es.enter_context(patch.object(old_sched.asyncio, "sleep", AsyncMock()))
+        # shutdown: False на входе while, затем True после первого dispatch → выход
+        es.enter_context(
+            patch.object(
+                old_sched, "shutdown_requested", MagicMock(side_effect=[False] + [True] * 10)
+            )
+        )
+        for name, mock in triggers.items():
+            es.enter_context(patch.object(old_sched, name, mock))
+        await old_sched.scheduler_loop()
+
+    return [name for name, m in triggers.items() if m.await_count > 0]
+
+
+@pytest.mark.parametrize("source", sorted(SOURCE_TO_OLD_TRIGGER))
+async def test_routing_parity_per_source(source: str) -> None:
+    """Для каждого source: старый if/elif зовёт ожидаемый trigger, kit-реестр резолвит handler."""
+    expected_trigger = SOURCE_TO_OLD_TRIGGER[source]
+
+    # старый роутинг
+    fired = await _drive_old_one_tick(source)
+    assert fired == [expected_trigger], (
+        f"source={source}: old scheduler fired {fired}, expected [{expected_trigger}]"
+    )
+
+    # kit роутинг — тот же source обязан резолвиться в handler
+    registry, _ = _build_recording_registry()
+    handler = resolve_handler(source, registry)
+    assert handler is not None, f"kit registry has no handler for source={source}"
+
+
+def test_routing_coverage_sets_match() -> None:
+    """Множество source'ов kit-реестра ⊇ множество, покрытое боевым if/elif."""
+    registry, _ = _build_recording_registry()
+    for source in SOURCE_TO_OLD_TRIGGER:
+        assert resolve_handler(source, registry) is not None, f"kit misses source={source}"
+
+
+def test_kit_native_handler_set() -> None:
+    """Ровно 8 kit-native sweep-обработчиков зарегистрированы встроенно."""
+    registry = build_registry()
+    assert set(registry) == _KIT_NATIVE_SOURCES
+    for src in _KIT_NATIVE_SOURCES:
+        assert resolve_handler(src, registry).log_name == src
+
+
+def test_unknown_source_resolves_none() -> None:
+    registry, _ = _build_recording_registry()
+    assert resolve_handler("totally_unknown_source", registry) is None
+
+
+def test_kit_scheduler_has_no_app_imports() -> None:
+    """kit scheduler развязан от app.* — ни одного `import app` / `from app` (AST, не substring)."""
+    import ast
+    import inspect
+
+    source = inspect.getsource(kit_sched)
+    tree = ast.parse(source)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                assert not alias.name.startswith("app"), f"illegal import app: {alias.name}"
+        elif isinstance(node, ast.ImportFrom):
+            assert not (node.module or "").startswith("app"), f"illegal from app: {node.module}"
+
+
+def test_deactivate_stale_wildcard_prefix() -> None:
+    """deactivate_stale_{avito,yandex,cian} → одна wildcard-запись (как боевой startswith)."""
+    registry, _ = _build_recording_registry()
+    h_avito = resolve_handler("deactivate_stale_avito", registry)
+    h_yandex = resolve_handler("deactivate_stale_yandex", registry)
+    h_cian = resolve_handler("deactivate_stale_cian", registry)
+    assert h_avito is not None
+    assert h_avito is h_yandex is h_cian  # один и тот же handler на всё семейство
+
+
+# ── 2. _claim_run advisory-lock parity ───────────────────────────────────────
+
+
+class _FakeResult:
+    def __init__(self, *, scalar: Any = None, fetchone: Any = None, fetchall: Any = None) -> None:
+        self._scalar = scalar
+        self._fetchone = fetchone
+        self._fetchall = fetchall or []
+
+    def scalar(self) -> Any:
+        return self._scalar
+
+    def fetchone(self) -> Any:
+        return self._fetchone
+
+    def fetchall(self) -> Any:
+        return self._fetchall
+
+
+class _FakeClaimDB:
+    """Мок Session для _claim_run: сценарий has_running_run + advisory-lock."""
+
+    def __init__(self, *, running_states: list[bool], lock: bool = True) -> None:
+        # running_states: очередь ответов has_running_run (по вызовам SELECT 1 FROM scrape_runs)
+        self._running = list(running_states)
+        self._lock = lock
+        self.committed = False
+        self.rolled_back = False
+        self.update_calls = 0
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        sql = str(stmt)
+        if "SELECT 1 FROM scrape_runs" in sql:
+            is_running = self._running.pop(0)
+            return _FakeResult(fetchone=(1,) if is_running else None)
+        if "pg_try_advisory_xact_lock" in sql:
+            return _FakeResult(scalar=self._lock)
+        if "UPDATE scrape_schedules" in sql:
+            self.update_calls += 1
+            return _FakeResult()
+        return _FakeResult()
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+
+def _ctx_with_runs(create_run_ret: int = 42) -> SchedulerContext:
+    runs = MagicMock()
+    runs.create_run = MagicMock(return_value=create_run_ret)
+    return SchedulerContext(
+        config=MagicMock(),
+        matcher=MagicMock(),
+        enrichment=MagicMock(),
+        session_factory=MagicMock(),
+        runs=runs,
+    )
+
+
+def test_claim_run_happy_path() -> None:
+    db = _FakeClaimDB(running_states=[False, False], lock=True)
+    ctx = _ctx_with_runs(create_run_ret=42)
+    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
+    assert run_id == 42
+    ctx.runs.create_run.assert_called_once()
+    assert db.update_calls == 1
+    assert db.committed is True
+
+
+def test_claim_run_skip_already_running() -> None:
+    db = _FakeClaimDB(running_states=[True], lock=True)
+    ctx = _ctx_with_runs()
+    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
+    assert run_id is None
+    ctx.runs.create_run.assert_not_called()
+
+
+def test_claim_run_skip_lock_busy() -> None:
+    # pre-check clean, но advisory-lock занят конкурентным тиком → skip
+    db = _FakeClaimDB(running_states=[False], lock=False)
+    ctx = _ctx_with_runs()
+    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
+    assert run_id is None
+    ctx.runs.create_run.assert_not_called()
+
+
+def test_claim_run_running_appeared_under_lock() -> None:
+    # pre-check clean, lock взят, но под локом появился running → rollback + skip (double-check)
+    db = _FakeClaimDB(running_states=[False, True], lock=True)
+    ctx = _ctx_with_runs()
+    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
+    assert run_id is None
+    assert db.rolled_back is True
+    ctx.runs.create_run.assert_not_called()
+
+
+# ── 3. reap_zombies ──────────────────────────────────────────────────────────
+
+
+def test_reap_zombies_counts_and_commits() -> None:
+    db = MagicMock()
+    db.execute.return_value = _FakeResult(fetchall=[MagicMock(id=1), MagicMock(id=2)])
+    n = reap_zombies(db)
+    assert n == 2
+    db.commit.assert_called_once()
+    # порог 6 часов передан в SQL-параметр
+    _stmt, params = db.execute.call_args[0]
+    assert params["interval"] == "6 hours"
+
+
+def test_reap_zombies_threshold_constant() -> None:
+    assert kit_sched.ZOMBIE_THRESHOLD_HOURS == old_sched.ZOMBIE_THRESHOLD_HOURS == 6
+
+
+# ── 4. proxy_healthcheck sub-hourly post_claim ───────────────────────────────
+
+
+def test_reschedule_after_minutes_updates_next_run() -> None:
+    db = MagicMock()
+    hook = reschedule_after_minutes(param="interval_minutes", default=30)
+    ctx = _ctx_with_runs()
+    hook(db, run_id=7, params={"interval_minutes": 15}, ctx=ctx)
+    db.commit.assert_called_once()
+    _stmt, params = db.execute.call_args[0]
+    assert params["mins"] == 15
+    assert params["rid"] == 7
+
+
+def test_reschedule_after_minutes_default() -> None:
+    db = MagicMock()
+    hook = reschedule_after_minutes()
+    hook(db, run_id=9, params={}, ctx=_ctx_with_runs())
+    _stmt, params = db.execute.call_args[0]
+    assert params["mins"] == 30
+
+
+# ── 5. _dispatch end-to-end (claim → fresh session → job → close) ────────────
+
+
+async def test_dispatch_kit_native_sweep_invokes_pipeline() -> None:
+    """kit-native handler: _dispatch клеймит, открывает сессию и зовёт pipeline.run_*."""
+    run_db = MagicMock()
+    ctx = SchedulerContext(
+        config=MagicMock(),
+        matcher=MagicMock(),
+        enrichment=MagicMock(),
+        session_factory=MagicMock(return_value=run_db),
+        runs=MagicMock(),
+    )
+    ctx.runs.create_run = MagicMock(return_value=100)
+
+    db = _FakeClaimDB(running_states=[False, False], lock=True)
+    registry = build_registry()
+    handler = registry["avito_city_sweep"]
+
+    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
+        run_id = await _dispatch(handler, db, _make_sched("avito_city_sweep"), ctx)
+        assert run_id == 100
+        # дождаться detached run-задачи
+        await asyncio.gather(*list(ctx._inflight_tasks))
+
+    mock_run.assert_awaited_once()
+    _args, kwargs = mock_run.call_args
+    assert kwargs["run_id"] == 100
+    assert kwargs["config"] is ctx.config
+    assert kwargs["matcher"] is ctx.matcher
+    run_db.close.assert_called_once()
+
+
+async def test_dispatch_product_handler_invokes_job() -> None:
+    """Продуктовый handler: _dispatch клеймит и зовёт инжектированный job на свежей сессии."""
+    run_db = MagicMock()
+    ctx = SchedulerContext(
+        config=MagicMock(),
+        matcher=MagicMock(),
+        enrichment=MagicMock(),
+        session_factory=MagicMock(return_value=run_db),
+        runs=MagicMock(),
+    )
+    ctx.runs.create_run = MagicMock(return_value=200)
+
+    db = _FakeClaimDB(running_states=[False, False], lock=True)
+    registry, fired = _build_recording_registry()
+    handler = resolve_handler("sber_index_pull", registry)
+
+    run_id = await _dispatch(handler, db, _make_sched("sber_index_pull"), ctx)
+    assert run_id == 200
+    await asyncio.gather(*list(ctx._inflight_tasks))
+
+    fired["sber_index_pull"].assert_called_once()
+    called_args = fired["sber_index_pull"].call_args[0]
+    assert called_args[0] is run_db  # свежая сессия
+    assert called_args[1] == 200  # run_id
+    run_db.close.assert_called_once()
+
+
+async def test_dispatch_pre_claim_gate_skips() -> None:
+    """pre_claim → False: _dispatch пропускает claim (cian cookie-gate семантика)."""
+    ctx = _ctx_with_runs()
+
+    async def _deny(db: Any, sch: dict[str, Any], c: Any) -> bool:
+        return False
+
+    job = AsyncMock()
+
+    async def _job(db: Any, run_id: int, params: dict[str, Any], c: Any) -> None:
+        job()
+
+    handler = Handler(_job, "cian_history_backfill", pre_claim=_deny)
+    db = _FakeClaimDB(running_states=[], lock=True)
+    run_id = await _dispatch(handler, db, _make_sched("cian_history_backfill"), ctx)
+    assert run_id is None
+    ctx.runs.create_run.assert_not_called()
+    job.assert_not_called()
+
+
+async def test_dispatch_post_claim_hook_runs() -> None:
+    """post_claim вызывается сразу после claim (proxy_healthcheck sub-hourly reschedule)."""
+    run_db = MagicMock()
+    ctx = SchedulerContext(
+        config=MagicMock(),
+        matcher=MagicMock(),
+        enrichment=MagicMock(),
+        session_factory=MagicMock(return_value=run_db),
+        runs=MagicMock(),
+    )
+    ctx.runs.create_run = MagicMock(return_value=300)
+    post = MagicMock()
+
+    async def _job(db: Any, run_id: int, params: dict[str, Any], c: Any) -> None:
+        pass
+
+    handler = Handler(_job, "proxy_healthcheck", post_claim=post)
+    db = _FakeClaimDB(running_states=[False, False], lock=True)
+    run_id = await _dispatch(handler, db, _make_sched("proxy_healthcheck"), ctx)
+    assert run_id == 300
+    await asyncio.gather(*list(ctx._inflight_tasks))
+    post.assert_called_once()
+    assert post.call_args[0][1] == 300  # run_id

tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py

@@ -0,0 +1,694 @@
+"""In-app scheduler — strangler-копия `app.services.scheduler` в scraper_kit (#2136).
+
+Разница с боевым `app.services.scheduler`:
+
+  1. Полная развязка от `app.*` — все продуктовые зависимости инжектируются через
+     `SchedulerContext` (config/matcher/enrichment/session_factory/runs/shutdown),
+     описанные Protocol'ами в `scraper_kit.contracts`. `grep "from app"` — пусто.
+
+  2. **Registry-dispatch вместо copy-paste.** Боевой модуль держит 25+ почти
+     одинаковых `trigger_*_run` (claim → new-session → create_task → close) плюс
+     27-веточный `if/elif` в `scheduler_loop`. Здесь — единственный параметризованный
+     путь `_dispatch()` + `HANDLERS: dict[str, Handler]`. Каждый source — одна запись
+     реестра; общий claim/session/drain-boilerplate вынесен в `_dispatch`.
+
+Kit-native (sweep-оркестраторы, уже перенесённые в `scraper_kit.orchestration.pipeline`)
+регистрируются встроенно (`_default_kit_handlers`). Продуктовые джобы, тело которых
+осталось в `app` (rosreestr_dkp / sber_index / deactivate_stale / *_backfill / …),
+инжектируются извне как `Handler` через `build_registry(product_handlers=...)`.
+
+Боевой рантайм (scraper-контейнер) по-прежнему крутит старый `app.services.scheduler` —
+это COPY, не MOVE. Переключение — отдельный поздний strangler-шаг.
+
+Критичная concurrency-логика (`_claim_run` advisory-lock + double-check, `reap_zombies`
+порог, heartbeat, SIGTERM-drain) перенесена ДОСЛОВНО — тот же SQL, то же ветвление.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import random
+from collections.abc import Callable, Coroutine, Mapping
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, time, timedelta
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import text
+
+from scraper_kit.orchestration import runs as _kit_runs
+from scraper_kit.orchestration.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,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+    from scraper_kit.contracts import (
+        EnrichmentJobs,
+        HouseMatcher,
+        ScraperConfig,
+        SessionFactory,
+    )
+
+logger = logging.getLogger(__name__)
+
+# Loop interval — check каждую минуту
+SCHEDULER_TICK_SEC = 60
+ZOMBIE_THRESHOLD_HOURS = 6
+
+# Бюджет дренажа детей < scheduler_main._DRAIN_TIMEOUT_S (100s) < docker grace (120s)
+# (см. #1182 P2 — идентично боевому scheduler'у).
+_CHILD_DRAIN_TIMEOUT_S = 80.0
+
+# ── типы job/handler ─────────────────────────────────────────────────────────
+# Job получает свежую сессию (открыта `_dispatch`), run_id, params и весь контекст
+# (config/matcher/enrichment/runs) — чтобы иметь доступ к инжектированным зависимостям.
+Job = Callable[["Session", int, dict[str, Any], "SchedulerContext"], Coroutine[Any, Any, None]]
+# Pre-claim gate: вернуть False → пропустить claim (напр. cian: нет/протухли cookies).
+PreClaim = Callable[["Session", dict[str, Any], "SchedulerContext"], Coroutine[Any, Any, bool]]
+# Post-claim hook: сразу после успешного claim (напр. proxy_healthcheck sub-hourly reschedule).
+PostClaim = Callable[["Session", int, dict[str, Any], "SchedulerContext"], None]
+
+
+@dataclass(frozen=True)
+class Handler:
+    """Одна запись реестра source→обработчик.
+
+    Инкапсулирует всё, что раньше размазывалось по индивидуальному `trigger_*_run`:
+      - `job`        — асинхронное тело (SERP/backfill/enrichment). Общий boilerplate
+                       (claim, свежая сессия, create_task, close, log) делает `_dispatch`.
+      - `pre_claim`  — опциональный гейт ДО claim (cookie-проверка cian → defer).
+      - `post_claim` — опциональный хук сразу ПОСЛЕ claim (sub-hourly reschedule).
+      - `log_name`   — имя источника в логах ("scheduler: triggered <log_name> run_id=…").
+    """
+
+    job: Job
+    log_name: str
+    pre_claim: PreClaim | None = None
+    post_claim: PostClaim | None = None
+
+
+@dataclass
+class SchedulerContext:
+    """Инжектируемые зависимости + рантайм-состояние планировщика.
+
+    Заменяет прямые импорты `app.core.config.settings` / `app.services.matching` /
+    `app.services.house_imv_backfill` / `app.core.db.SessionLocal` /
+    `app.services.scrape_runs` / `app.core.shutdown.shutdown_requested`.
+    """
+
+    config: ScraperConfig
+    matcher: HouseMatcher
+    enrichment: EnrichmentJobs
+    session_factory: SessionFactory
+    shutdown_requested: Callable[[], bool] = lambda: False
+    # runs-модуль (create_run/update_heartbeat/mark_done/… ). По умолчанию — kit-копия
+    # scraper_kit.orchestration.runs; в тестах подменяется recorder'ом.
+    runs: Any = _kit_runs
+    # #1182 P2: strong-ref'ы detached run-задач для graceful SIGTERM-drain'а.
+    _inflight_tasks: set[asyncio.Task[None]] = field(default_factory=set)
+
+    def spawn_tracked(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
+        """create_task + регистрация в _inflight_tasks для graceful-drain'а (#1182 P2).
+
+        strong-ref в set'е держит задачу до завершения (RUF006) и даёт scheduler_loop'у
+        дождаться её на SIGTERM-drain'е; done-callback ретривит exception и убирает задачу.
+        """
+        task = asyncio.create_task(coro)
+        self._inflight_tasks.add(task)
+
+        def _on_done(t: asyncio.Task[None]) -> None:
+            self._inflight_tasks.discard(t)
+            if not t.cancelled():
+                t.exception()
+
+        task.add_done_callback(_on_done)
+        return task
+
+    async def drain_inflight(self) -> None:
+        """Дождаться завершения detached run-задач перед teardown'ом (#1182 P2).
+
+        Кооперативные дети дочекивают текущий unit + mark_done и резолвятся;
+        некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S и остаются на внешний
+        hard-cancel fallback. Не busy-spin — один asyncio.wait.
+        """
+        pending = [t for t in self._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], ctx: SchedulerContext) -> int | None:
+    """Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at.
+
+    Returns run_id или None если уже есть running run для этого source.
+    Общий helper для всех source-handler'ов. Advisory-lock concurrency-логика перенесена
+    ДОСЛОВНО из боевого scheduler'а (создание run — через инжектированный ctx.runs).
+    """
+    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 = ctx.runs.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).
+
+    Используется когда pre_claim делает early-return (например cian: отсутствуют/протухли
+    cookies) ДО _claim_run. Без этого get_due_schedules переотбирает schedule на каждом
+    тике (SCHEDULER_TICK_SEC), и pre-check гоняется раз в минуту круглосуточно.
+    """
+    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(),
+    )
+
+
+def reschedule_after_minutes(*, param: str = "interval_minutes", default: int = 30) -> PostClaim:
+    """Фабрика post_claim-хука: sub-hourly re-schedule сразу после claim (#2162).
+
+    Боевой scheduler держит суточную гранулярность в compute_next_run_at, а
+    proxy_healthcheck хочет каденс раз в N минут. Хук СРАЗУ после claim переопределяет
+    next_run_at на now() + interval_minutes (source-специфично, не трогая shared
+    compute_next_run_at). Если run ещё идёт на следующем тике — has_running_run в
+    _claim_run вернёт None (skip) и next_run_at не сбросится → авто-throttle по факту.
+    """
+
+    def _hook(db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext) -> None:
+        # source читаем из ЗАКЛЕЙМЛЕННОГО schedule — берём из params-owner через closure нельзя,
+        # поэтому апдейтим по run_id → source (schedule owner). Проще: UPDATE по source из
+        # scrape_runs текущего run_id, что эквивалентно WHERE source = <this source>.
+        minutes = int(params.get(param, default))
+        db.execute(
+            text(
+                """
+                UPDATE scrape_schedules
+                SET next_run_at = now() + make_interval(mins => CAST(:mins AS integer)),
+                    updated_at = now()
+                WHERE source = (SELECT source FROM scrape_runs WHERE id = CAST(:rid AS bigint))
+                """
+            ),
+            {"mins": minutes, "rid": run_id},
+        )
+        db.commit()
+
+    return _hook
+
+
+# ── единственный параметризованный dispatch (заменяет 25 trigger_* + 27 if/elif) ──
+
+
+async def _dispatch(
+    handler: Handler,
+    db: Session,
+    schedule_row: dict[str, Any],
+    ctx: SchedulerContext,
+) -> int | None:
+    """Общий путь claim→spawn для ЛЮБОГО source (весь boilerplate прежних trigger_*_run).
+
+    1. Опциональный pre_claim-гейт (False → skip без claim).
+    2. _claim_run (advisory-lock; None → уже running / concurrent claim).
+    3. Опциональный post_claim-хук (sub-hourly reschedule).
+    4. spawn detached run-задачи: свежая сессия → handler.job → close (+ exception-log).
+    """
+    if handler.pre_claim is not None:
+        proceed = await handler.pre_claim(db, schedule_row, ctx)
+        if not proceed:
+            return None
+
+    run_id = _claim_run(db, schedule_row, ctx)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    if handler.post_claim is not None:
+        handler.post_claim(db, run_id, params, ctx)
+
+    async def _run() -> None:
+        run_db = ctx.session_factory()
+        try:
+            await handler.job(run_db, run_id, params, ctx)
+        except Exception:
+            logger.exception("scheduler: %s crashed run_id=%d", handler.log_name, run_id)
+        finally:
+            run_db.close()
+
+    ctx.spawn_tracked(_run())
+    logger.info("scheduler: triggered %s run_id=%d", handler.log_name, run_id)
+    return run_id
+
+
+def resolve_handler(source: str, registry: Mapping[str, Handler]) -> Handler | None:
+    """Найти handler по source: exact-match, затем wildcard-префикс ("deactivate_stale_*").
+
+    Единственная точка ветвления вместо 27 if/elif. Wildcard-ключ (кончается на "*")
+    матчит по префиксу — покрывает семейство deactivate_stale_avito/yandex/cian одной
+    записью реестра (как боевой `source.startswith("deactivate_stale_")`).
+    """
+    handler = registry.get(source)
+    if handler is not None:
+        return handler
+    for key, h in registry.items():
+        if key.endswith("*") and source.startswith(key[:-1]):
+            return h
+    return None
+
+
+# ── kit-native sweep-обработчики (тело в scraper_kit.orchestration.pipeline) ──────
+# Param-чтение идентично боевым trigger_*_run — та же семантика окон/дефолтов.
+
+
+async def _job_avito_city_sweep(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    await run_avito_city_sweep(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        enrichment=ctx.enrichment,
+        shutdown_requested=ctx.shutdown_requested,
+        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)),
+    )
+
+
+async def _job_avito_newbuilding_sweep(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    await run_avito_newbuilding_sweep(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        shutdown_requested=ctx.shutdown_requested,
+        pages=int(params.get("pages", 20)),
+        request_delay_sec=float(params.get("request_delay_sec", 7.0)),
+    )
+
+
+async def _job_avito_full_load(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    _incremental_days = params.get("incremental_days")
+    incremental_days = int(_incremental_days) if _incremental_days is not None else None
+    await run_avito_full_load(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        shutdown_requested=ctx.shutdown_requested,
+        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,
+    )
+
+
+async def _job_avito_full_load_exhaustive(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    # incremental_days форсится в None → полный exhaustive обход (weekly cadence).
+    await run_avito_full_load(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        shutdown_requested=ctx.shutdown_requested,
+        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,
+    )
+
+
+async def _job_yandex_city_sweep(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    await run_yandex_city_sweep(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        enrichment=ctx.enrichment,
+        shutdown_requested=ctx.shutdown_requested,
+        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"),
+    )
+
+
+async def _job_cian_city_sweep(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    await run_cian_city_sweep(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        shutdown_requested=ctx.shutdown_requested,
+        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)),
+    )
+
+
+async def _job_cian_full_load(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    await run_cian_full_load(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        shutdown_requested=ctx.shutdown_requested,
+        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,
+    )
+
+
+async def _job_domclick_city_sweep(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    await run_domclick_city_sweep(
+        db,
+        run_id=run_id,
+        config=ctx.config,
+        matcher=ctx.matcher,
+        shutdown_requested=ctx.shutdown_requested,
+        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)),
+    )
+
+
+def _default_kit_handlers() -> dict[str, Handler]:
+    """Встроенные kit-native sweep-обработчики (тело в scraper_kit.orchestration.pipeline)."""
+    return {
+        "avito_city_sweep": Handler(_job_avito_city_sweep, "avito_city_sweep"),
+        "avito_newbuilding_sweep": Handler(_job_avito_newbuilding_sweep, "avito_newbuilding_sweep"),
+        "avito_full_load": Handler(_job_avito_full_load, "avito_full_load"),
+        "avito_full_load_exhaustive": Handler(
+            _job_avito_full_load_exhaustive, "avito_full_load_exhaustive"
+        ),
+        "yandex_city_sweep": Handler(_job_yandex_city_sweep, "yandex_city_sweep"),
+        "cian_city_sweep": Handler(_job_cian_city_sweep, "cian_city_sweep"),
+        "cian_full_load": Handler(_job_cian_full_load, "cian_full_load"),
+        "domclick_city_sweep": Handler(_job_domclick_city_sweep, "domclick_city_sweep"),
+    }
+
+
+def build_registry(product_handlers: Mapping[str, Handler] | None = None) -> dict[str, Handler]:
+    """Собрать реестр source→Handler: kit-native sweeps + инжектированные продуктовые джобы.
+
+    `product_handlers` — джобы, тело которых осталось в `app` (rosreestr_dkp / sber_index /
+    deactivate_stale_* / *_backfill / cadastral_geo_match / house_* / proxy_healthcheck / …).
+    Их строит app-side wiring и передаёт сюда. Продуктовые ключи МОГУТ переопределять
+    kit-native (последнее слово за продуктом).
+
+    Wildcard-ключи ("deactivate_stale_*") резолвятся `resolve_handler` по префиксу.
+    """
+    registry = dict(_default_kit_handlers())
+    if product_handlers:
+        registry.update(product_handlers)
+    return registry
+
+
+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(ctx: SchedulerContext, registry: Mapping[str, Handler]) -> None:
+    """Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд.
+
+    Структура (reap → get_due → dispatch → SIGTERM-drain) идентична боевому
+    scheduler'у; 27-веточный if/elif заменён единственным `resolve_handler` + `_dispatch`.
+    """
+    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 ctx.shutdown_requested():
+            logger.info("scheduler: SIGTERM-drain — stop claiming new runs, exiting tick loop")
+            break
+        try:
+            db = ctx.session_factory()
+            try:
+                # Reap zombies first
+                reap_zombies(db)
+                # Process due schedules
+                due = get_due_schedules(db)
+                for sch in due:
+                    source = sch["source"]
+                    handler = resolve_handler(source, registry)
+                    if handler is None:
+                        logger.warning("scheduler: unknown source=%s, skip", source)
+                    else:
+                        await _dispatch(handler, db, sch, ctx)
+
+                    # #1182 Phase 2: после каждого dispatch'а проверяем drain — текущий
+                    # run уже отпущен в свою asyncio-задачу (сам докоммитит/выйдет по
+                    # своему checkpoint'у), а новые в этом тике не запускаем.
+                    if ctx.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 ctx.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 ctx.drain_inflight()
+    logger.info("scheduler: tick loop exited (in-flight drain complete)")
+
+
+__all__ = [
+    "SCHEDULER_TICK_SEC",
+    "ZOMBIE_THRESHOLD_HOURS",
+    "Handler",
+    "SchedulerContext",
+    "build_registry",
+    "compute_next_run_at",
+    "get_due_schedules",
+    "has_running_run",
+    "reap_zombies",
+    "reschedule_after_minutes",
+    "resolve_handler",
+    "scheduler_loop",
+]