Merge pull request 'refactor(tradein/scheduler): удалить legacy scheduler_loop, kit единственный путь (#2397 Part C)' (#2402) from refactor/tradein-remove-legacy-scheduler-loop into main

tradein-mvp/backend/app/core/config.py

@@ -18,11 +18,12 @@ class Settings(BaseSettings):
     # Default true preserves existing behaviour.
     scheduler_enable: bool = Field(default=True, validation_alias="SCHEDULER_ENABLE")
 
-    # ── strangler-миграция scheduler → scraper_kit (#2192, ship-dark) ─────────
+    # ── strangler-миграция scheduler → scraper_kit (#2192, завершена #2397 Part C) ────
-    # Дефолт False → scheduler_main идёт СТАРЫМ путём (app.services.scheduler.scheduler_loop),
+    # scheduler_main.py теперь безусловно идёт kit-путём (scraper_kit.orchestration.scheduler
-    # прод не меняет поведение на деплое. True → kit-путь (scraper_kit.orchestration.scheduler
+    # + product_handlers) — legacy app.services.scheduler.scheduler_loop fallback удалён
-    # + product_handlers). Переключается вручную после prod-smoke; боевой scheduler остаётся
+    # (был мёртвым грузом: прод давно на USE_KIT_SCHEDULER=true). Поле оставлено (extra=
-    # мгновенным fallback.
+    # "ignore" в model_config защищает от startup-краха на leftover env var), но больше
+    # ни на что не влияет.
     use_kit_scheduler: bool = Field(default=False, validation_alias="USE_KIT_SCHEDULER")
     # Proxy-pool флаги — объявлены заранее (потребители P3/P4 отдельно), дефолт False.
     use_proxy_pool_curl: bool = Field(default=False, validation_alias="USE_PROXY_POOL_CURL")

tradein-mvp/backend/app/main.py

@@ -6,7 +6,6 @@ Standalone версия, выделена из основного gendesign repo
 
 from __future__ import annotations
 
-import asyncio
 import logging
 import os
 import re
@@ -31,7 +30,6 @@ from app.core.db import SessionLocal
 from app.core.fdw import ensure_fdw_user_mapping
 from app.core.ratelimit import RateLimitMiddleware
 from app.observability.sentry_scrub import scrub_pii_event
-from app.services.scheduler import scheduler_loop
 
 logger = logging.getLogger(__name__)
 
@@ -85,23 +83,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     except Exception:
         logger.exception("FDW user mapping bootstrap failed — cadastral queries may fail")
 
-    # Startup: launch scheduler background task (disabled when SCHEDULER_ENABLE=false,
+    # #2397 Part C: legacy in-app scheduler launch removed — scheduling lives exclusively
-    # e.g. when using the systemd timer trigger — see tradein-mvp/ops/systemd/ #581).
+    # in the tradein-scraper container (`python -m app.scheduler_main`, kit scheduler).
-    scheduler_task = None
+    # Prod backend has always run with SCHEDULER_ENABLE=false (see docker-compose.prod.yml);
-    if settings.scheduler_enable:
+    # this API process never actually launched scheduler_loop() in production.
-        scheduler_task = asyncio.create_task(scheduler_loop())
-        logger.info("FastAPI lifespan: scheduler task spawned")
-    else:
-        logger.info("FastAPI lifespan: in-app scheduler disabled (SCHEDULER_ENABLE=false)")
     yield
-    # Shutdown: cancel scheduler if it was started
-    if scheduler_task is not None:
-        scheduler_task.cancel()
-        try:
-            await scheduler_task
-        except asyncio.CancelledError:
-            pass
-        logger.info("FastAPI lifespan: scheduler cancelled cleanly")
 
 
 app = FastAPI(

tradein-mvp/backend/app/scheduler_main.py

@@ -6,8 +6,8 @@
 
 Запуск: python -m app.scheduler_main
 
-Zombie-reap: reap_zombies() вызывается в начале каждого тика scheduler_loop()
+Zombie-reap: reap_zombies() вызывается в начале каждого тика kit scheduler_loop
-(см. app/services/scheduler.py), дублировать здесь не нужно.
+(см. scraper_kit/orchestration/scheduler.py), дублировать здесь не нужно.
 """
 
 from __future__ import annotations
@@ -163,23 +163,19 @@ async def _run_kit_scheduler() -> None:
 
 
 async def _run() -> None:
-    """Async entrypoint: scheduler_loop() с кооперативным SIGTERM/SIGINT-drain'ом.
+    """Async entrypoint: kit-scheduler с кооперативным SIGTERM/SIGINT-drain'ом.
 
     SIGTERM/SIGINT → request_shutdown() (НЕ task.cancel()): бегущий scrape-unit
-    докоммитит текущую карточку и выйдет сам на ближайшем between-unit checkpoint'е
+    докоммитит текущую карточку и выйдет сам на ближайшем between-unit checkpoint'е.
-    (scheduler_loop / avito_detail_backfill / import_rosreestr_dkp). Safety-net в
+    Safety-net в _await_scheduler гарантирует выход в пределах docker grace.
-    _await_scheduler гарантирует выход в пределах docker grace.
 
-    Развилка #2192 (ship-dark): USE_KIT_SCHEDULER=false (дефолт) → боевой
+    #2397 Part C: legacy app.services.scheduler.scheduler_loop fallback (USE_KIT_SCHEDULER=
-    app.services.scheduler.scheduler_loop без изменений; =true → kit-путь.
+    false ship-dark путь из #2192) удалён — kit (_run_kit_scheduler) теперь единственный
+    путь. Прод уже давно на kit (USE_KIT_SCHEDULER=true), legacy был мёртвым грузом.
+    settings.use_kit_scheduler остаётся в конфиге (extra="ignore" защищает от startup-краха
+    на leftover env var), но на ветвление больше не влияет.
     """
-    if settings.use_kit_scheduler:
+    task = asyncio.create_task(_run_kit_scheduler())
-        logger.info("scheduler_main: USE_KIT_SCHEDULER=true — running kit-scheduler path")
-        task = asyncio.create_task(_run_kit_scheduler())
-    else:
-        from app.services.scheduler import scheduler_loop
-
-        task = asyncio.create_task(scheduler_loop())
 
     loop = asyncio.get_running_loop()
 

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

@@ -732,9 +732,9 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i
 def run_house_dedup_merge(db: Session, *, run_id: int, params: dict) -> dict[str, int]:
     """Run-lifecycle wrapper for the recurring house-dedup merge (sync, DB-only).
 
-    Launched by the in-app scheduler (source='house_dedup_merge') via
+    Launched by the kit scheduler (source='house_dedup_merge') via
-    trigger_house_dedup_merge_run, or manually. Mirrors run_cadastral_geo_match: pure
+    product_handlers._job_house_dedup_merge, or manually. Mirrors run_cadastral_geo_match:
-    internal DB op, finalises scrape_runs (mark_done / mark_failed) with counters.
+    pure internal DB op, finalises scrape_runs (mark_done / mark_failed) with counters.
 
     Params (default_params jsonb):
         dry_run: compute + return counts then ROLLBACK without writing (default false).

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

@@ -13,9 +13,9 @@ TRUE-MIRROR REFRESH (флаг database-expert): 080-seed использовал
 заново гоняет ту же 080-derivation INSERT...SELECT. DELETE+INSERT в ОДНОЙ транзакции
 (атомарно — таблица никогда не пуста mid-refresh). После commit таблица == свежий re-seed.
 
-Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app
+Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается kit-scheduler'ом
-scheduler'ом через trigger_asking_to_sold_ratio_run() (scheduler.py), по образцу
+через product_handlers._job_asking_to_sold_ratio (run_in_executor), по образцу
-snapshot_listing_sources / import_rosreestr_dkp (sync task в run_in_executor).
+snapshot_listing_sources / import_rosreestr_dkp.
 
 Окно расписания 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC), чтобы
 refresh потреблял свежие ДКП-сделки того же дня.

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

@@ -281,11 +281,11 @@ def run_geoportal_coords_backfill(
 ) -> BackfillCoordsResult:
     """Run-lifecycle wrapper for backfill_coords_from_geoportal (sync, DB-only task).
 
-    Launched by the in-app scheduler (source='geoportal_coords_backfill') via
+    Launched by the kit scheduler (source='geoportal_coords_backfill') via
-    trigger_geoportal_coords_backfill_run, или вручную. Pure local exact street+house match
+    product_handlers._job_geoportal_coords_backfill, или вручную. Pure local exact
-    против ekb_geoportal_buildings — никакого внешнего HTTP, поэтому весь backlog (или
+    street+house match против ekb_geoportal_buildings — никакого внешнего HTTP, поэтому
-    ограниченный `limit`) отрабатывается за один прогон без wall-clock budget (в отличие
+    весь backlog (или ограниченный `limit`) отрабатывается за один прогон без wall-clock
-    от geocode_missing_listings, который rate-limited Nominatim'ом).
+    budget (в отличие от geocode_missing_listings, который rate-limited Nominatim'ом).
 
     Params (default_params jsonb):
         limit:       максимум листингов за прогон (default None = весь backlog).

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

@@ -267,10 +267,10 @@ def refresh_and_match(
 def run_cadastral_geo_match(db: Session, *, run_id: int, params: dict) -> CadMatchResult:
     """Run-lifecycle wrapper for the combined refresh+match job (sync, DB-only).
 
-    Launched by the in-app scheduler (source='cadastral_geo_match') via
+    Launched by the kit scheduler (source='cadastral_geo_match') via
-    trigger_cadastral_geo_match_run, or manually. Refreshes cad_buildings_local from the
+    product_handlers._job_cadastral_geo_match, or manually. Refreshes cad_buildings_local
-    FDW (one bulk scan) then geo-matches listings.building_cadastral_number to the nearest
+    from the FDW (one bulk scan) then geo-matches listings.building_cadastral_number to the
-    cadastral building within threshold_m.
+    nearest cadastral building within threshold_m.
 
     Params (default_params jsonb):
         threshold_m:  max nearest-building distance to accept (metres, default 50).

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

@@ -10,8 +10,8 @@
 - avito: все сегменты (segments=None), TTL=10 дней -- поведение без изменений.
 - Строки НЕ удаляются -- история нужна для бэктеста (#667).
 
-Задача синхронная (DB-only, никаких внешних HTTP-вызовов) -- запускается in-app
+Задача синхронная (DB-only, никаких внешних HTTP-вызовов) -- запускается kit-scheduler'ом
-scheduler'ом через trigger_deactivate_stale_run() (scheduler.py),
+через product_handlers._job_deactivate_stale (wildcard-handler deactivate_stale_*),
 по образцу snapshot_listing_sources / recompute_asking_to_sold_ratios
 (sync task в run_in_executor).
 

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

@@ -19,8 +19,8 @@ sentry logging integration), когда следующий квартал ПРО
 LAG_ALLOWANCE_DAYS (по умолчанию 45) на публикацию дампа Росреестром. Сейчас (2026-07-02)
 до порога 2026-08-15 ещё далеко → алерта НЕТ. Если к 2026-08-16 Q2 не импортирован → алерт.
 
-Задача синхронная (DB-only, один SELECT max(deal_date)) — запускается in-app scheduler'ом
+Задача синхронная (DB-only, один SELECT max(deal_date)) — запускается kit-scheduler'ом
-через trigger_deals_freshness_monitor_run() в run_in_executor, по образцу
+через product_handlers._job_deals_freshness_monitor в run_in_executor, по образцу
 listing_source_snapshot / rosreestr_dkp_import. Вердикт вычисляет ЧИСТАЯ функция
 evaluate_deals_freshness() (frozen-now тестируется без БД).
 

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

@@ -5,8 +5,8 @@
 listing_source_events. Так история per-source цены копится СРАЗУ, независимо от
 (сейчас DORMANT) скраперов — см. шапку data/sql/079_listing_source_history.sql.
 
-Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app
+Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается kit-scheduler'ом
-scheduler'ом через trigger_listing_source_snapshot_run() (scheduler.py),
+через product_handlers._job_listing_source_snapshot,
 по образцу import_rosreestr_dkp (sync task в run_in_executor).
 
 Вся работа — два set-based SQL statement'а (snapshot upsert + event-diff CTE),

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

@@ -75,9 +75,9 @@ class OsmPoiRefreshResult:
 def run_osm_poi_ekb_refresh(db: Session, *, run_id: int, params: dict) -> OsmPoiRefreshResult:
     """Run-lifecycle wrapper for the OSM POI local-mirror refresh (scheduler entrypoint).
 
-    Launched by the in-app scheduler (source='osm_poi_ekb_refresh') via
+    Launched by the kit scheduler (source='osm_poi_ekb_refresh') via
-    trigger_osm_poi_ekb_refresh_run, or manually. No params consumed — single unconditional
+    product_handlers._job_osm_poi_ekb_refresh, or manually. No params consumed — single
-    bulk refresh (`params` accepted for scheduler-signature compatibility / future use).
+    unconditional bulk refresh (`params` accepted for scheduler-signature compatibility).
 
     Finalises scrape_runs (mark_done / mark_failed) with counters.
     """

tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py

@@ -10,14 +10,13 @@
 - Идемпотентность: UPDATE с WHERE lat IS NULL не затирает уже заполненное.
 - SAVEPOINT откат батча при исключении в DB.execute.
 - run_geoportal_coords_backfill (#1967): run-lifecycle wrapper mark_done/mark_failed.
-- Scheduler wiring (#1967): trigger fn + dispatch branch + migration 171 seed content.
+- Migration 171 seed content.
 
 Все сетевые вызовы и DB замоканы.
 """
 
 from __future__ import annotations
 
-import inspect
 import os
 import sys
 from pathlib import Path
@@ -31,7 +30,6 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:
 _wp_mock = MagicMock()
 sys.modules.setdefault("weasyprint", _wp_mock)
 
-from app.services import scheduler  # noqa: E402
 from app.services.geocoder import GeocodeSuggestion  # noqa: E402
 from app.tasks import backfill_listings_coords_geoportal as bcg  # noqa: E402
 from app.tasks.backfill_listings_coords_geoportal import (  # noqa: E402
@@ -437,38 +435,6 @@ def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> N
     assert "db down" in failed["err"]
 
 
-# ── Scheduler wiring (#1967) ───────────────────────────────────────────────────
-
-
-def test_scheduler_has_trigger_and_dispatch() -> None:
-    assert hasattr(scheduler, "trigger_geoportal_coords_backfill_run")
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "geoportal_coords_backfill"' in loop_src
-    assert "trigger_geoportal_coords_backfill_run(db, sch)" in loop_src
-
-
-def test_trigger_claims_run_and_runs_in_executor() -> None:
-    src = inspect.getsource(scheduler.trigger_geoportal_coords_backfill_run)
-    assert "_claim_run(db, schedule_row)" in src
-    # sync DB-only task → run_in_executor (not a bare async call).
-    assert "run_in_executor" in src
-    assert "run_geoportal_coords_backfill" in src
-
-
-def test_dispatch_runs_before_geocode_missing_listings_in_source_docstring() -> None:
-    """Header docs must place geoportal_coords_backfill BEFORE geocode_missing_listings.
-
-    Ordering is the whole point (#1967): exact house-level match should claim NULL-coord
-    listings before the coarser Nominatim/Yandex backfill does.
-    """
-    header = scheduler.__doc__ or ""
-    geoportal_idx = header.find("geoportal_coords_backfill")
-    nominatim_idx = header.find("geocode_missing_listings")
-    assert geoportal_idx != -1
-    assert nominatim_idx != -1
-    assert geoportal_idx < nominatim_idx
-
-
 # ── Migration content (#1967) ──────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py

@@ -6,7 +6,6 @@ emitted SQL via the text() clauses and inspect.getsource and check:
   - the LATERAL KNN shape (ST_DWithin geography gate + `<->` KNN order + LIMIT 1),
   - the threshold/only_missing params are bound (psycopg-v3 CAST discipline, no :p::type),
   - the refresh does a single bulk FDW scan (TRUNCATE + INSERT ... FROM gendesign_cad_buildings),
-  - scheduler wiring (trigger fn + dispatch branch),
   - migration 124 (table + GIST) and 125 (schedule seed) contents.
 
 Plus a fake-db behavioural test driving the chunk-loop + counter logic without Postgres.
@@ -30,7 +29,6 @@ import pytest
 # settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db.
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
-from app.services import scheduler
 from app.tasks import cadastral_geo_match as cgm
 
 _SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql"
@@ -146,24 +144,6 @@ def test_matcher_docstring_marks_approximation_and_threshold() -> None:
     assert "threshold_m=%d" in _MATCH_SRC
 
 
-# ── Scheduler wiring ──────────────────────────────────────────────────────────
-
-
-def test_scheduler_has_trigger_and_dispatch() -> None:
-    assert hasattr(scheduler, "trigger_cadastral_geo_match_run")
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "cadastral_geo_match"' in loop_src
-    assert "trigger_cadastral_geo_match_run(db, sch)" in loop_src
-
-
-def test_trigger_claims_run_and_runs_in_executor() -> None:
-    src = inspect.getsource(scheduler.trigger_cadastral_geo_match_run)
-    assert "_claim_run(db, schedule_row)" in src
-    # sync DB-only task → run_in_executor (not a bare async call).
-    assert "run_in_executor" in src
-    assert "run_cadastral_geo_match" in src
-
-
 # ── Migration content ─────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py

@@ -521,15 +521,12 @@ async def test_backfill_resolve_then_idempotent_rerun_no_reresolve() -> None:
 
 
 # ---------------------------------------------------------------------------
-# #973 (950-E2): nightly schedule — run_newbuilding_enrich wrapper, scheduler
+# #973 (950-E2): nightly schedule — run_newbuilding_enrich wrapper + seed migration 103.
-# trigger/dispatch wiring, and the seed migration 103.
 # ---------------------------------------------------------------------------
 
-import inspect  # noqa: E402
 import re  # noqa: E402
 from pathlib import Path  # noqa: E402
 
-from app.services import scheduler  # noqa: E402
 from app.tasks import newbuilding_enrich_backfill as task_mod  # noqa: E402
 
 _SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql"
@@ -616,71 +613,6 @@ async def test_run_wrapper_marks_failed_and_reraises(monkeypatch: pytest.MonkeyP
     assert "cian exploded" in failed["err"]
 
 
-# ── Scheduler wiring: trigger fn claims + dispatches; skip when already running ──
-
-
-def test_trigger_fn_exists_and_is_async() -> None:
-    fn = getattr(scheduler, "trigger_newbuilding_enrich_run", None)
-    assert fn is not None, "trigger_newbuilding_enrich_run missing from scheduler"
-    assert inspect.iscoroutinefunction(fn)
-
-
-def test_dispatch_branch_wired() -> None:
-    """scheduler_loop dispatches source='newbuilding_enrich' to the trigger."""
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "newbuilding_enrich"' in loop_src
-    assert "trigger_newbuilding_enrich_run(db, sch)" in loop_src
-
-
-def test_trigger_claims_run_and_delegates_to_task() -> None:
-    """Trigger claims via _claim_run and delegates to run_newbuilding_enrich (async task)."""
-    trig_src = inspect.getsource(scheduler.trigger_newbuilding_enrich_run)
-    assert "_claim_run(db, schedule_row)" in trig_src
-    assert "run_newbuilding_enrich" in trig_src
-    # #1182 P2: detached spawn централизован в _spawn_tracked (внутри — asyncio.create_task
-    # + registry strong-ref для graceful SIGTERM-drain).
-    assert "_spawn_tracked(_run())" in trig_src
-
-
-@pytest.mark.asyncio
-async def test_trigger_claims_and_dispatches(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Happy path: _claim_run returns a run_id → an asyncio task is spawned."""
-    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 4242)
-    spawned: list = []
-    real_create_task = scheduler.asyncio.create_task
-
-    def _spy_create_task(coro):
-        spawned.append(coro)
-        # Wrap the real coro so it actually runs (and closes the SessionLocal).
-        return real_create_task(coro)
-
-    monkeypatch.setattr(scheduler.asyncio, "create_task", _spy_create_task)
-    # Stop the spawned _run() from touching a real DB / importing the heavy task.
-    monkeypatch.setattr(scheduler, "SessionLocal", lambda: MagicMock())
-
-    run_id = await scheduler.trigger_newbuilding_enrich_run(
-        MagicMock(), {"source": "newbuilding_enrich", "default_params": {}}
-    )
-
-    assert run_id == 4242
-    assert len(spawned) == 1
-
-
-@pytest.mark.asyncio
-async def test_trigger_skips_when_already_running(monkeypatch: pytest.MonkeyPatch) -> None:
-    """_claim_run None (already running / lock held) → trigger returns None, no task."""
-    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: None)
-    spawned: list = []
-    monkeypatch.setattr(scheduler.asyncio, "create_task", lambda coro: spawned.append(coro))
-
-    run_id = await scheduler.trigger_newbuilding_enrich_run(
-        MagicMock(), {"source": "newbuilding_enrich", "default_params": {}}
-    )
-
-    assert run_id is None
-    assert spawned == []
-
-
 # ── Migration 103 ─────────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_asking_to_sold_ratio.py

@@ -4,8 +4,7 @@ recompute_asking_to_sold_ratios is SQL-heavy, so most assertions are static: we
 emitted SQL via .text and check the TRUE-MIRROR shape (DELETE WHERE district='' BEFORE the
 re-derivation INSERT), that the derivation reuses the exact 080 logic (deals + listings, the
 trailing-12mo window, the 30/30 threshold, the global -1 fallback row), and the psycopg-v3
-cast discipline (no :param::type). We also assert the scheduler wiring (trigger fn + dispatch
+cast discipline (no :param::type). We also assert the migration 082 contents.
-branch) and the migration 082 contents.
 
 Plus one cheap behavioural test: a fake db (monkeypatched .execute) drives the counter logic
 without a real Postgres.
@@ -26,7 +25,6 @@ import pytest
 # static / fake-db; no live database is touched.
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
-from app.services import scheduler
 from app.tasks import asking_to_sold_ratio as ratio_mod
 
 _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
@@ -196,29 +194,6 @@ def test_task_returns_counters_and_finalises_run() -> None:
     assert "raise" in _TASK_SRC  # re-raise on failure — no silent swallow
 
 
-# ── Scheduler wiring ──────────────────────────────────────────────────────────
-
-
-def test_trigger_fn_exists_and_is_async() -> None:
-    fn = getattr(scheduler, "trigger_asking_to_sold_ratio_run", None)
-    assert fn is not None, "trigger_asking_to_sold_ratio_run missing from scheduler"
-    assert inspect.iscoroutinefunction(fn)
-
-
-def test_dispatch_branch_wired() -> None:
-    """scheduler_loop dispatches source='asking_to_sold_ratio_refresh' to the trigger."""
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "asking_to_sold_ratio_refresh"' in loop_src
-    assert "trigger_asking_to_sold_ratio_run(db, sch)" in loop_src
-
-
-def test_trigger_runs_sync_task_in_executor() -> None:
-    """Sync DB-only task → run_in_executor (mirror snapshot), not a bare await."""
-    trig_src = inspect.getsource(scheduler.trigger_asking_to_sold_ratio_run)
-    assert "run_in_executor(None, recompute_asking_to_sold_ratios" in trig_src
-    assert "_claim_run(db, schedule_row)" in trig_src
-
-
 # ── Migration 082 ─────────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_deactivate_stale_avito.py

@@ -1,8 +1,8 @@
 """Tests for the stale avito listing deactivation task (#759).
 
-Static assertions (SQL shape, psycopg-v3 cast discipline, scheduler wiring) mirror
+Static assertions (SQL shape, psycopg-v3 cast discipline) mirror the pattern established
-the pattern established in test_listing_source_snapshot.py. One behavioural test drives
+in test_listing_source_snapshot.py. One behavioural test drives the counter logic via a
-the counter logic via a fake db without a live database.
+fake db without a live database.
 """
 
 import inspect
@@ -13,11 +13,10 @@ from typing import Any
 
 import pytest
 
-# Stub DATABASE_URL before any app import (same guard as test_scheduler.py /
+# Stub DATABASE_URL before any app import (same guard as test_listing_source_snapshot.py —
-# test_listing_source_snapshot.py — these tests are static/fake-db, no live DB).
+# these tests are static/fake-db, no live DB).
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
-from app.services import scheduler
 from app.tasks import deactivate_stale_avito as task_mod
 
 _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
@@ -103,31 +102,6 @@ def test_task_commits_and_rollbacks_on_error() -> None:
     assert "db.rollback()" in _GENERIC_TASK_SRC
 
 
-# ── Scheduler wiring ──────────────────────────────────────────────────────────
-
-
-def test_trigger_fn_exists_and_is_async() -> None:
-    fn = getattr(scheduler, "trigger_deactivate_stale_avito_run", None)
-    assert fn is not None, "trigger_deactivate_stale_avito_run missing from scheduler"
-    assert inspect.iscoroutinefunction(fn)
-
-
-def test_dispatch_branch_wired() -> None:
-    """scheduler_loop dispatches deactivate_stale_* sources to the generic trigger."""
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source.startswith("deactivate_stale_")' in loop_src
-    assert "trigger_deactivate_stale_run(db, sch)" in loop_src
-
-
-def test_trigger_runs_sync_task_in_executor() -> None:
-    """Sync DB-only task → run_in_executor, not a bare await."""
-    trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_avito_run)
-    # run_in_executor call may span two lines; check both parts separately.
-    assert "run_in_executor(" in trig_src
-    assert "deactivate_stale_avito_listings" in trig_src
-    assert "_claim_run(db, schedule_row)" in trig_src
-
-
 # ── Migration 090 ─────────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_deactivate_stale_listings.py

@@ -10,7 +10,6 @@ import pytest
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
-from app.services import scheduler
 from app.tasks import deactivate_stale_avito as task_mod
 
 _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
@@ -77,32 +76,6 @@ def test_compat_wrapper_calls_generic_with_avito_defaults() -> None:
     assert "segments=None" in _COMPAT_FN_SRC
 
 
-def test_trigger_deactivate_stale_run_exists_and_is_async() -> None:
-    fn = getattr(scheduler, "trigger_deactivate_stale_run", None)
-    assert fn is not None, "trigger_deactivate_stale_run missing from scheduler"
-    assert inspect.iscoroutinefunction(fn)
-
-
-def test_dispatch_branch_uses_startswith() -> None:
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source.startswith("deactivate_stale_")' in loop_src
-    assert "trigger_deactivate_stale_run(db, sch)" in loop_src
-
-
-def test_trigger_runs_sync_task_in_executor() -> None:
-    trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run)
-    assert "run_in_executor(" in trig_src
-    assert "deactivate_stale_listings" in trig_src
-    assert "_claim_run(db, schedule_row)" in trig_src
-
-
-def test_trigger_reads_params_from_default_params() -> None:
-    trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_run)
-    assert "listing_source" in trig_src
-    assert "ttl_days" in trig_src
-    assert "segments" in trig_src
-
-
 class _FakeResult:
     def __init__(self, rowcount: int) -> None:
         self.rowcount = rowcount

tradein-mvp/backend/tests/test_deals_freshness_monitor.py

@@ -8,12 +8,11 @@
      - кастомный lag_allowance_days.
   2. check_deals_freshness с FakeDB (no-alert / alert / empty→mark_failed).
   3. Свойства миграции 162 (по образцу test_deactivate_stale_listings).
-  4. Регистрацию в scheduler-dispatch + kit product_handlers (registry остаётся зелёным).
+  4. Регистрацию в kit product_handlers (registry остаётся зелёным).
 """
 
 from __future__ import annotations
 
-import inspect
 import os
 import re
 from datetime import UTC, date, datetime
@@ -24,7 +23,6 @@ import pytest
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
-from app.services import scheduler
 from app.services.product_handlers import build_product_handlers
 from app.tasks import deals_freshness_monitor as mon
 
@@ -223,26 +221,7 @@ def test_migration_162_no_psycopg_trap() -> None:
     assert not re.search(r":\w+::", sql)
 
 
-# ── Регистрация в scheduler + kit registry ────────────────────────────────────
+# ── Регистрация в kit registry ─────────────────────────────────────────────────
-
-
-def test_trigger_exists_and_is_async() -> None:
-    fn = getattr(scheduler, "trigger_deals_freshness_monitor_run", None)
-    assert fn is not None, "trigger_deals_freshness_monitor_run missing from scheduler"
-    assert inspect.iscoroutinefunction(fn)
-
-
-def test_dispatch_branch_present() -> None:
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "deals_freshness_monitor"' in loop_src
-    assert "trigger_deals_freshness_monitor_run(db, sch)" in loop_src
-
-
-def test_trigger_runs_sync_task_in_executor() -> None:
-    trig_src = inspect.getsource(scheduler.trigger_deals_freshness_monitor_run)
-    assert "run_in_executor(" in trig_src
-    assert "check_deals_freshness" in trig_src
-    assert "_claim_run(db, schedule_row)" in trig_src
 
 
 def test_kit_product_handler_registered() -> None:

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -26,7 +26,6 @@ import pytest
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
 from app.services import house_dedup_merge as hdm
-from app.services import scheduler
 
 _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
 _MIGRATION_135 = _SQL_DIR / "135_scrape_schedules_seed_house_dedup_merge.sql"
@@ -560,30 +559,6 @@ def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> N
     assert "merge exploded" in failed["err"]
 
 
-# ── Scheduler wiring ──────────────────────────────────────────────────────────
-
-
-def test_scheduler_has_trigger_and_dispatch() -> None:
-    assert hasattr(scheduler, "trigger_house_dedup_merge_run")
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "house_dedup_merge"' in loop_src
-    assert "trigger_house_dedup_merge_run(db, sch)" in loop_src
-
-
-def test_trigger_claims_run_and_runs_in_executor() -> None:
-    src = inspect.getsource(scheduler.trigger_house_dedup_merge_run)
-    assert "_claim_run(db, schedule_row)" in src
-    assert "if run_id is None:" in src
-    # sync DB-only task → run_in_executor (mirror cadastral_geo_match, not a bare await).
-    assert "run_in_executor" in src
-    assert "run_house_dedup_merge" in src
-    # fresh session inside the spawned task + #1182 P2 detached spawn via _spawn_tracked
-    # (registry strong-ref = RUF006 keep-alive + graceful SIGTERM-drain).
-    assert "run_db = SessionLocal()" in src
-    assert "_spawn_tracked(_run())" in src
-    assert "run_db.close()" in src
-
-
 # ── Migration content ─────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py

@@ -1,237 +0,0 @@
-"""Tests for wiring house_imv_backfill as an in-app scheduler source (#854).
-
-The bulk IMV backfill service (services/house_imv_backfill.backfill_house_imv) is
-already built + covered elsewhere; this gate covers ONLY the scheduler wiring added
-by #854:
-  - trigger_house_imv_backfill_run shape (_claim_run → asyncio.create_task → fresh
-    SessionLocal → mark_done/mark_failed lifecycle → add_done_callback / RUF006),
-  - dispatch branch in scheduler_loop routes source=="house_imv_backfill",
-  - migration 132 seeds an idempotent scrape_schedules row with default_params.
-
-Convention mirrors tests/tasks/test_cadastral_geo_match.py: hermetic static analysis
-(inspect.getsource + migration file read) plus a fake-async behavioural test driving
-the trigger's lifecycle without a live Postgres.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import inspect
-import os
-import re
-from pathlib import Path
-from typing import Any
-
-import pytest
-
-# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db.
-os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
-
-from app.services import scheduler
-
-_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
-_MIGRATION_132 = _SQL_DIR / "132_scrape_schedules_seed_house_imv.sql"
-
-_TRIGGER_SRC = inspect.getsource(scheduler.trigger_house_imv_backfill_run)
-
-
-# ── Scheduler wiring ──────────────────────────────────────────────────────────
-
-
-def test_scheduler_has_trigger_and_dispatch() -> None:
-    assert hasattr(scheduler, "trigger_house_imv_backfill_run")
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "house_imv_backfill"' in loop_src
-    assert "trigger_house_imv_backfill_run(db, sch)" in loop_src
-
-
-def test_trigger_mirrors_backfill_sibling_pattern() -> None:
-    """_claim_run gate → fresh SessionLocal → backfill_house_imv → add_done_callback."""
-    assert "_claim_run(db, schedule_row)" in _TRIGGER_SRC
-    assert "if run_id is None:" in _TRIGGER_SRC
-    # fresh session inside the spawned task (not the scheduler-tick session).
-    assert "run_db = SessionLocal()" in _TRIGGER_SRC
-    assert "backfill_house_imv" in _TRIGGER_SRC
-    # async service → bare await (NOT run_in_executor, unlike the sync DB-only triggers).
-    assert "run_in_executor" not in _TRIGGER_SRC
-    # #1182 P2: detached spawn через _spawn_tracked — strong-ref в registry (RUF006 +
-    # graceful SIGTERM-drain), done-callback ретривит exception. Заменил прежний
-    # `task = asyncio.create_task(_run()); task.add_done_callback(...)`.
-    assert "_spawn_tracked(_run())" in _TRIGGER_SRC
-    assert "run_db.close()" in _TRIGGER_SRC
-
-
-def test_trigger_owns_run_lifecycle() -> None:
-    """No tasks/*-wrapper for this service → the trigger finalises scrape_runs itself."""
-    assert "runs_mod.mark_done(" in _TRIGGER_SRC
-    assert "runs_mod.mark_failed(" in _TRIGGER_SRC
-    assert "logger.exception" in _TRIGGER_SRC
-    # heartbeat callback keeps the long IMV run alive vs reap_zombies (#1363).
-    assert "heartbeat=" in _TRIGGER_SRC
-    assert "update_heartbeat" in _TRIGGER_SRC
-
-
-def test_trigger_reads_default_params() -> None:
-    """default_params drive batch_size / request_delay_sec / only_status with safe defaults."""
-    assert 'params.get("batch_size"' in _TRIGGER_SRC
-    assert 'params.get("request_delay_sec"' in _TRIGGER_SRC
-    assert 'params.get("only_status"' in _TRIGGER_SRC
-
-
-# ── Migration content ─────────────────────────────────────────────────────────
-
-
-def test_migration_132_seeds_schedule_idempotent() -> None:
-    sql = _MIGRATION_132.read_text(encoding="utf-8")
-    assert "INSERT INTO scrape_schedules" in sql
-    assert "'house_imv_backfill'" in sql
-    assert "ON CONFLICT (source) DO NOTHING" in sql
-    assert "BEGIN;" in sql and "COMMIT;" in sql
-
-
-def test_migration_132_next_run_not_null_tomorrow() -> None:
-    """next_run_at NOT NULL = tomorrow (avoid deploy-time fire), per seed convention."""
-    sql = _MIGRATION_132.read_text(encoding="utf-8")
-    assert "CURRENT_DATE + INTERVAL '1 day'" in sql
-    assert "make_interval(hours => 15)" in sql
-
-
-def test_migration_132_default_params() -> None:
-    sql = _MIGRATION_132.read_text(encoding="utf-8")
-    assert '"batch_size": 50' in sql
-    assert '"request_delay_sec": 5' in sql
-    assert '"only_status": "pending"' in sql
-
-
-def test_migration_132_no_psycopg_v3_colon_colon_cast() -> None:
-    """psycopg v3: never :param::type — must use CAST(:param AS type)."""
-    sql = _MIGRATION_132.read_text(encoding="utf-8")
-    assert not re.search(r":\w+::", sql)
-
-
-def test_migration_132_matches_seed_column_set() -> None:
-    """Exact column set of existing scrape_schedules seeds (source..default_params)."""
-    sql = _MIGRATION_132.read_text(encoding="utf-8")
-    for col in (
-        "source",
-        "enabled",
-        "window_start_hour",
-        "window_end_hour",
-        "next_run_at",
-        "default_params",
-    ):
-        assert col in sql
-
-
-# ── Fake-async behavioural: trigger lifecycle without Postgres ────────────────
-
-
-class _FakeDB:
-    """Minimal Session stand-in: only needs to be a distinct object + close()."""
-
-    def __init__(self) -> None:
-        self.closed = 0
-
-    def close(self) -> None:
-        self.closed += 1
-
-
-def test_trigger_skips_when_claim_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
-    """If _claim_run finds a running run (None) → trigger returns None, spawns nothing."""
-    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: None)
-
-    async def _go() -> int | None:
-        return await scheduler.trigger_house_imv_backfill_run(
-            _FakeDB(),  # type: ignore[arg-type]
-            {"source": "house_imv_backfill", "default_params": {}},
-        )
-
-    assert asyncio.run(_go()) is None
-
-
-def test_trigger_claims_runs_backfill_and_marks_done(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Happy path: claim run → backfill_house_imv awaited with params → mark_done(counters)."""
-    captured: dict[str, Any] = {}
-    marked: dict[str, Any] = {}
-
-    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 77)
-    monkeypatch.setattr(scheduler, "SessionLocal", _FakeDB)
-    monkeypatch.setattr(scheduler.runs_mod, "update_heartbeat", lambda *a, **k: None)
-    monkeypatch.setattr(
-        scheduler.runs_mod,
-        "mark_done",
-        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
-    )
-    monkeypatch.setattr(scheduler.runs_mod, "mark_failed", lambda *a, **k: None)
-
-    class _Result:
-        checked, saved, skipped, errors, duration_sec = 10, 6, 2, 2, 42.0
-
-    async def _fake_backfill(_db: Any, **kwargs: Any) -> _Result:
-        captured.update(kwargs)
-        return _Result()
-
-    import app.services.house_imv_backfill as _svc
-
-    monkeypatch.setattr(_svc, "backfill_house_imv", _fake_backfill)
-
-    async def _go() -> int | None:
-        run_id = await scheduler.trigger_house_imv_backfill_run(
-            _FakeDB(),  # type: ignore[arg-type]
-            {
-                "source": "house_imv_backfill",
-                "default_params": {
-                    "batch_size": 25,
-                    "request_delay_sec": 3,
-                    "only_status": "transient_error",
-                },
-            },
-        )
-        # let the spawned _run() task finish.
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
-        return run_id
-
-    run_id = asyncio.run(_go())
-    assert run_id == 77
-    assert captured["batch_size"] == 25
-    assert captured["request_delay_sec"] == 3.0
-    assert captured["only_status"] == "transient_error"
-    assert callable(captured["heartbeat"])
-    assert marked["run_id"] == 77
-    assert marked["counters"]["checked"] == 10
-    assert marked["counters"]["saved"] == 6
-
-
-def test_trigger_marks_failed_on_backfill_crash(monkeypatch: pytest.MonkeyPatch) -> None:
-    """If backfill_house_imv raises → trigger marks the run failed (does not bubble)."""
-    failed: dict[str, Any] = {}
-
-    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 88)
-    monkeypatch.setattr(scheduler, "SessionLocal", _FakeDB)
-    monkeypatch.setattr(scheduler.runs_mod, "update_heartbeat", lambda *a, **k: None)
-    monkeypatch.setattr(scheduler.runs_mod, "mark_done", lambda *a, **k: None)
-    monkeypatch.setattr(
-        scheduler.runs_mod,
-        "mark_failed",
-        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
-    )
-
-    async def _boom(_db: Any, **kwargs: Any) -> None:
-        raise RuntimeError("imv 403")
-
-    import app.services.house_imv_backfill as _svc
-
-    monkeypatch.setattr(_svc, "backfill_house_imv", _boom)
-
-    async def _go() -> None:
-        await scheduler.trigger_house_imv_backfill_run(
-            _FakeDB(),  # type: ignore[arg-type]
-            {"source": "house_imv_backfill", "default_params": {}},
-        )
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
-
-    asyncio.run(_go())
-    assert failed["run_id"] == 88
-    assert "imv 403" in failed["err"]

tradein-mvp/backend/tests/test_kit_registry_completeness.py

@@ -1,87 +0,0 @@
-"""Полнота kit-registry vs боевой scheduler-dispatch (#2192).
-
-Гарантирует ship-dark инвариант: КАЖДЫЙ source, который обрабатывает боевой
-`app.services.scheduler.scheduler_loop` (if/elif + startswith), резолвится в
-kit-registry (`build_registry(product_handlers=build_product_handlers(ctx))`).
-Тест ПАДАЕТ, если source добавили в боевой dispatch, но забыли в registry —
-иначе после переключения USE_KIT_SCHEDULER=true такой source молча не запустится.
-
-Источник истины — сам файл боевого scheduler'а: парсим `source == "X"` (exact)
-и `source.startswith("X")` (wildcard), чтобы при дрейфе dispatch тест ловил
-рассинхрон автоматически, а не по захардкоженному списку.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-from pathlib import Path
-
-os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
-
-from scraper_kit.orchestration.scheduler import (
-    SchedulerContext,
-    build_registry,
-    resolve_handler,
-)
-
-import app.services.scheduler as _sched_mod
-from app.services.product_handlers import build_product_handlers
-from app.services.scraper_adapters import (
-    RealEnrichmentJobs,
-    RealMatcherAdapter,
-    RealScraperConfig,
-    RealSessionFactory,
-)
-
-
-def _production_sources() -> tuple[set[str], set[str]]:
-    """(exact, prefixes) source'ов из боевого scheduler_loop dispatch."""
-    src = Path(_sched_mod.__file__).read_text(encoding="utf-8")
-    exact = set(re.findall(r'source == "([^"]+)"', src))
-    prefixes = set(re.findall(r'source\.startswith\("([^"]+)"\)', src))
-    return exact, prefixes
-
-
-def _build_registry() -> dict:
-    ctx = SchedulerContext(
-        config=RealScraperConfig(),
-        matcher=RealMatcherAdapter(),
-        enrichment=RealEnrichmentJobs(),
-        session_factory=RealSessionFactory(),
-    )
-    return build_registry(product_handlers=build_product_handlers(ctx))
-
-
-def test_production_dispatch_parse_sane() -> None:
-    """Sanity: парсер видит непустой набор source'ов (защита от regex-дрейфа)."""
-    exact, prefixes = _production_sources()
-    # На момент #2192 боевой dispatch держит 25 exact + 1 wildcard-семейство.
-    assert len(exact) >= 25, f"parse regressed — got {len(exact)} exact sources"
-    assert "deactivate_stale_" in prefixes
-
-
-def test_every_production_source_resolves_in_kit_registry() -> None:
-    """Каждый боевой exact-source резолвится в kit-registry."""
-    registry = _build_registry()
-    exact, _ = _production_sources()
-    missing = [s for s in sorted(exact) if resolve_handler(s, registry) is None]
-    assert not missing, f"sources missing from kit registry: {missing}"
-
-
-def test_wildcard_family_resolves() -> None:
-    """Каждый deactivate_stale_* член семейства резолвится через wildcard-Handler."""
-    registry = _build_registry()
-    _, prefixes = _production_sources()
-    for prefix in sorted(prefixes):
-        for member in ("avito", "yandex", "cian"):
-            source = f"{prefix}{member}"
-            assert (
-                resolve_handler(source, registry) is not None
-            ), f"wildcard source {source} did not resolve"
-
-
-def test_unknown_source_does_not_resolve() -> None:
-    """Негативный контроль: несуществующий source → None (тест реально дискриминирует)."""
-    registry = _build_registry()
-    assert resolve_handler("totally_unknown_source_xyz", registry) is None

tradein-mvp/backend/tests/test_listing_source_snapshot.py

@@ -3,7 +3,7 @@
 Writer (snapshot_listing_sources) is SQL-heavy, so most assertions are static: we read
 the emitted SQL via inspect/text-attr and check the upsert + event-diff CTE shape, table/
 column names, and the psycopg-v3 cast discipline (no :param::type). We also assert the
-scheduler wiring (trigger fn + dispatch branch) and the migration 079 contents.
+migration 079 contents.
 
 Plus one cheap behavioural test: a fake db (monkeypatched .execute) drives the counter
 logic without a real Postgres.
@@ -24,7 +24,6 @@ import pytest
 # static / fake-db; no live database is touched.
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
-from app.services import scheduler
 from app.tasks import listing_source_snapshot as snap_mod
 
 _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
@@ -133,29 +132,6 @@ def test_writer_returns_counters_and_finalises_run() -> None:
     assert "mark_failed" in _WRITER_SRC
 
 
-# ── Scheduler wiring ──────────────────────────────────────────────────────────
-
-
-def test_trigger_fn_exists_and_is_async() -> None:
-    fn = getattr(scheduler, "trigger_listing_source_snapshot_run", None)
-    assert fn is not None, "trigger_listing_source_snapshot_run missing from scheduler"
-    assert inspect.iscoroutinefunction(fn)
-
-
-def test_dispatch_branch_wired() -> None:
-    """scheduler_loop dispatches source='listing_source_snapshot' to the trigger."""
-    loop_src = inspect.getsource(scheduler.scheduler_loop)
-    assert 'source == "listing_source_snapshot"' in loop_src
-    assert "trigger_listing_source_snapshot_run(db, sch)" in loop_src
-
-
-def test_trigger_runs_sync_task_in_executor() -> None:
-    """Sync DB-only task → run_in_executor (mirror rosreestr), not a bare await."""
-    trig_src = inspect.getsource(scheduler.trigger_listing_source_snapshot_run)
-    assert "run_in_executor(None, snapshot_listing_sources" in trig_src
-    assert "_claim_run(db, schedule_row)" in trig_src
-
-
 # ── Migration 079 ─────────────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_scheduler.py

@@ -1,29 +1,19 @@
-"""Offline smoke for scheduler logic."""
+"""Offline smoke for compute_next_run_at (app/services/scheduler.py).
+
+#2397 Part C removed the legacy scheduler_loop + trigger_*/get_due_schedules/_claim_run/
+reap_zombies/_spawn_tracked/_drain_inflight machinery from app.services.scheduler — the kit
+scheduler (scraper_kit.orchestration.scheduler, covered by test_scraper_kit_scheduler_parity.py)
+is now the only scheduling path. compute_next_run_at survives as a shared helper (still used
+by admin.py for the operator-facing "next run" preview), so this file now covers only it.
+"""
 
-import asyncio
 import os
-from unittest.mock import AsyncMock, MagicMock
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
 from datetime import UTC, datetime
 
-import pytest
+from app.services.scheduler import compute_next_run_at
-
-import app.services.scheduler as sched
-from app.core import shutdown as sd
-from app.services.scheduler import ZOMBIE_THRESHOLD_HOURS, compute_next_run_at
-
-
-@pytest.fixture(autouse=True)
-def _reset_drain_state() -> None:
-    """#1182 P2: shutdown-флаг и registry детей — module-global; чистим вокруг каждого
-    теста (изоляция от detached-задач, что триггеры спавнят в _inflight_tasks)."""
-    sd.reset_shutdown()
-    sched._inflight_tasks.clear()
-    yield
-    sched._inflight_tasks.clear()
-    sd.reset_shutdown()
 
 
 def test_compute_next_run_in_normal_window() -> None:
@@ -49,10 +39,6 @@ def test_compute_next_run_window_1_hour() -> None:
     assert next_at.hour == 3
 
 
-def test_zombie_threshold_constant() -> None:
-    assert ZOMBIE_THRESHOLD_HOURS > 0
-
-
 def test_compute_next_run_is_future() -> None:
     """next_run_at всегда в будущем (завтра) — не в прошлом."""
     now = datetime(2026, 5, 23, 3, 30, tzinfo=UTC)
@@ -101,725 +87,3 @@ def test_compute_next_run_interval_days_cross_midnight_weekly() -> None:
     next_at = compute_next_run_at(22, 3, now=now, interval_days=7)
     assert next_at > now
     assert next_at.hour >= 22 or next_at.hour < 3
-
-
-# ── #750: атомарный _claim_run (advisory lock — anti-double-sweep) ──────────
-
-from collections import deque  # noqa: E402
-
-from app.services import scheduler as _sched  # noqa: E402
-
-
-class _Scalar:
-    def __init__(self, v: object) -> None:
-        self._v = v
-
-    def scalar(self) -> object:
-        return self._v
-
-
-class _Fetchone:
-    def __init__(self, row: object) -> None:
-        self._row = row
-
-    def fetchone(self) -> object:
-        return self._row
-
-
-class _FakeSchedulerDB:
-    """Stateful fake моделирующий advisory-lock claim (#750), без реальной БД.
-
-    - SELECT pg_try_advisory_xact_lock → lock_available.
-    - SELECT 1 FROM scrape_runs (has_running_run) → из running_results (если задано),
-      иначе по флагу self.running[source].
-    - UPDATE scrape_schedules → no-op.
-    create_run (монкипатчится) выставляет self.running[source]=True.
-    """
-
-    def __init__(self, *, lock_available: bool = True, running_results: list[bool] | None = None):
-        self.lock_available = lock_available
-        self.running: dict[str, bool] = {}
-        self._running_results: deque[bool] = deque(running_results or [])
-        self.commits = 0
-        self.rollbacks = 0
-
-    def execute(self, stmt: object, params: dict | None = None):
-        sql = str(stmt)
-        if "pg_try_advisory_xact_lock" in sql:
-            return _Scalar(self.lock_available)
-        if "SELECT 1 FROM scrape_runs" in sql:
-            if self._running_results:
-                return _Fetchone(object() if self._running_results.popleft() else None)
-            src = (params or {}).get("source")
-            return _Fetchone(object() if self.running.get(src) else None)
-        return _Fetchone(None)  # UPDATE scrape_schedules etc.
-
-    def commit(self) -> None:
-        self.commits += 1
-
-    def rollback(self) -> None:
-        self.rollbacks += 1
-
-
-_SCHED_ROW = {
-    "source": "avito_city_sweep",
-    "window_start_hour": 2,
-    "window_end_hour": 5,
-    "default_params": {},
-}
-
-
-def test_claim_run_double_call_creates_one_run(monkeypatch: pytest.MonkeyPatch) -> None:
-    """#750 DoD: два последовательных _claim_run одного source (нет running) →
-    первый run_id, второй None (create_run вызван РОВНО один раз)."""
-    db = _FakeSchedulerDB()
-    calls = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        db.running[source] = True
-        return 100 + calls["n"]
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    rid1 = _sched._claim_run(db, _SCHED_ROW)
-    rid2 = _sched._claim_run(db, _SCHED_ROW)
-
-    assert rid1 == 101
-    assert rid2 is None
-    assert calls["n"] == 1  # второй НЕ создал run
-
-
-def test_claim_run_skips_when_advisory_lock_unavailable(monkeypatch: pytest.MonkeyPatch) -> None:
-    """#750: lock занят конкурентным тиком → return None, create_run НЕ вызван."""
-    db = _FakeSchedulerDB(lock_available=False)
-    calls = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        return 999
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    assert _sched._claim_run(db, _SCHED_ROW) is None
-    assert calls["n"] == 0
-
-
-def test_claim_run_rechecks_running_under_lock(monkeypatch: pytest.MonkeyPatch) -> None:
-    """#750: running появился МЕЖДУ pre-check и взятием лока (double-checked) →
-    return None + rollback (освобождение лока), create_run НЕ вызван."""
-    # pre-check → False (нет running), re-check под локом → True (появился).
-    db = _FakeSchedulerDB(lock_available=True, running_results=[False, True])
-    calls = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        return 999
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    assert _sched._claim_run(db, _SCHED_ROW) is None
-    assert calls["n"] == 0
-    assert db.rollbacks == 1  # лок освобождён
-
-
-# ── #974: trigger_yandex_newbuilding_sweep_run + dispatch ────────────────────
-
-
-_YANDEX_NB_SWEEP_ROW = {
-    "source": "yandex_newbuilding_sweep",
-    "window_start_hour": 2,
-    "window_end_hour": 5,
-    "default_params": {
-        "limit": 5,
-        "request_delay_sec": 8.0,
-        "city": "ekaterinburg",
-    },
-}
-
-
-@pytest.mark.asyncio
-async def test_trigger_yandex_newbuilding_sweep_creates_run(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """trigger_yandex_newbuilding_sweep_run claims run + launches create_task (#974)."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"create_run": [], "sweep": []}
-
-    def fake_create_run(_d, *, source, params):
-        calls["create_run"].append(source)
-        db.running[source] = True
-        return 301
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    # Patch enrich_yandex_newbuilding_sweep to a coroutine that records call
-    sweep_result = MagicMock()
-    sweep_result.to_dict.return_value = {}
-
-    async def fake_sweep(run_db, *, limit, **kwargs):
-        calls["sweep"].append(limit)
-        return sweep_result
-
-    # Patch inside the _run closure (lazy import)
-    monkeypatch.setattr(
-        "app.tasks.yandex_newbuilding_sweep.enrich_yandex_newbuilding_sweep",
-        fake_sweep,
-    )
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    # mark_done is called inside _run; mock it so it doesn't fail on fake db
-    monkeypatch.setattr(_sched.runs_mod, "mark_done", lambda *a, **kw: None)
-    monkeypatch.setattr(_sched.runs_mod, "mark_failed", lambda *a, **kw: None)
-
-    run_id = await _sched.trigger_yandex_newbuilding_sweep_run(db, _YANDEX_NB_SWEEP_ROW)
-
-    assert run_id == 301
-    assert calls["create_run"] == ["yandex_newbuilding_sweep"]
-    # Drain event-loop so the asyncio.create_task runs
-    await asyncio.sleep(0)
-    assert calls["sweep"] == [5]
-
-
-@pytest.mark.asyncio
-async def test_trigger_yandex_newbuilding_sweep_skips_when_running(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """trigger_yandex_newbuilding_sweep_run returns None when source already running."""
-    db = _FakeSchedulerDB()
-    db.running["yandex_newbuilding_sweep"] = True
-
-    calls: dict[str, int] = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        return 999
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    result = await _sched.trigger_yandex_newbuilding_sweep_run(db, _YANDEX_NB_SWEEP_ROW)
-
-    assert result is None
-    assert calls["n"] == 0
-
-
-@pytest.mark.asyncio
-async def test_scheduler_dispatch_routes_yandex_newbuilding_sweep(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Dispatch block в scheduler_loop роутит yandex_newbuilding_sweep → trigger."""
-    triggered: list[str] = []
-
-    async def fake_trigger(db, sch):
-        triggered.append(sch["source"])
-        return 1
-
-    monkeypatch.setattr(_sched, "trigger_yandex_newbuilding_sweep_run", fake_trigger)
-
-    db = object()
-    sch = _YANDEX_NB_SWEEP_ROW
-    source = sch["source"]
-    # Replicate the elif chain from scheduler_loop for yandex_newbuilding_sweep
-    if source == "yandex_newbuilding_sweep":
-        await _sched.trigger_yandex_newbuilding_sweep_run(db, sch)
-
-    assert "yandex_newbuilding_sweep" in triggered
-
-
-# ── #973: trigger_cian_city_sweep_run + dispatch ─────────────────────────────
-
-
-_CIAN_SWEEP_ROW = {
-    "source": "cian_city_sweep",
-    "window_start_hour": 2,
-    "window_end_hour": 5,
-    "default_params": {
-        "pages_per_anchor": 3,
-        "request_delay_sec": 5.0,
-        "radius_m": 1500,
-        "detail_top_n": 10,
-        "enrich_houses": True,
-    },
-}
-
-
-@pytest.mark.asyncio
-async def test_trigger_cian_city_sweep_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
-    """trigger_cian_city_sweep_run claims run + launches create_task (#973)."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"create_run": [], "sweep": []}
-
-    def fake_create_run(_d, *, source, params):
-        calls["create_run"].append(source)
-        db.running[source] = True
-        return 201
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    # Patch run_cian_city_sweep to a coroutine that records call
-    async def fake_sweep(run_db, *, run_id, **kwargs):
-        calls["sweep"].append(run_id)
-
-    monkeypatch.setattr(_sched, "run_cian_city_sweep", fake_sweep)
-    # Patch SessionLocal to return the same db
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    run_id = await _sched.trigger_cian_city_sweep_run(db, _CIAN_SWEEP_ROW)
-
-    assert run_id == 201
-    assert calls["create_run"] == ["cian_city_sweep"]
-    # Drain event-loop so the asyncio.create_task runs
-    await asyncio.sleep(0)
-    assert calls["sweep"] == [201]
-
-
-@pytest.mark.asyncio
-async def test_trigger_cian_city_sweep_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
-    """trigger_cian_city_sweep_run returns None when source already running."""
-    db = _FakeSchedulerDB()
-    db.running["cian_city_sweep"] = True
-
-    calls: dict[str, int] = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        return 999
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    result = await _sched.trigger_cian_city_sweep_run(db, _CIAN_SWEEP_ROW)
-
-    assert result is None
-    assert calls["n"] == 0
-
-
-@pytest.mark.asyncio
-async def test_scheduler_dispatch_routes_cian_city_sweep(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Dispatch block в scheduler_loop роутит cian_city_sweep → trigger_cian_city_sweep_run."""
-    triggered: list[str] = []
-
-    async def fake_trigger(db, sch):
-        triggered.append(sch["source"])
-        return 1
-
-    monkeypatch.setattr(_sched, "trigger_cian_city_sweep_run", fake_trigger)
-
-    # Simulate the dispatch block for one due schedule
-    db = object()
-    sch = _CIAN_SWEEP_ROW
-    source = sch["source"]
-    # Replicate the elif chain from scheduler_loop for cian_city_sweep
-    if source == "cian_city_sweep":
-        await _sched.trigger_cian_city_sweep_run(db, sch)
-
-    assert "cian_city_sweep" in triggered
-
-
-# ── cian_full_load: trigger_cian_full_load_run + dispatch ────────────────────
-
-
-_CIAN_FULL_LOAD_ROW = {
-    "source": "cian_full_load",
-    "window_start_hour": 20,
-    "window_end_hour": 22,
-    "default_params": {
-        "price_cap_per_bucket": 1400,
-        "concurrency": 5,
-        "request_delay_sec": 4.0,
-        "enrich_detail": False,
-        "detail_top_n": 0,
-    },
-}
-
-
-@pytest.mark.asyncio
-async def test_trigger_cian_full_load_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
-    """trigger_cian_full_load_run claims run + launches run_cian_full_load via create_task."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"create_run": [], "load": []}
-
-    def fake_create_run(_d, *, source, params):
-        calls["create_run"].append(source)
-        db.running[source] = True
-        return 301
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    async def fake_full_load(run_db, *, run_id, **kwargs):
-        calls["load"].append((run_id, kwargs))
-
-    monkeypatch.setattr(_sched, "run_cian_full_load", fake_full_load)
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    run_id = await _sched.trigger_cian_full_load_run(db, _CIAN_FULL_LOAD_ROW)
-
-    assert run_id == 301
-    assert calls["create_run"] == ["cian_full_load"]
-    # Drain event-loop so the asyncio.create_task runs
-    await asyncio.sleep(0)
-    assert len(calls["load"]) == 1
-    loaded_run_id, kwargs = calls["load"][0]
-    assert loaded_run_id == 301
-    # Params propagated from default_params; resume_run_id forced to None
-    assert kwargs["price_cap_per_bucket"] == 1400
-    assert kwargs["concurrency"] == 5
-    assert kwargs["request_delay_sec"] == 4.0
-    assert kwargs["enrich_detail"] is False
-    assert kwargs["detail_top_n"] == 0
-    assert kwargs["resume_run_id"] is None
-
-
-@pytest.mark.asyncio
-async def test_trigger_cian_full_load_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
-    """trigger_cian_full_load_run returns None when source already running."""
-    db = _FakeSchedulerDB()
-    db.running["cian_full_load"] = True
-
-    calls: dict[str, int] = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        return 999
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    result = await _sched.trigger_cian_full_load_run(db, _CIAN_FULL_LOAD_ROW)
-
-    assert result is None
-    assert calls["n"] == 0
-
-
-@pytest.mark.asyncio
-async def test_scheduler_dispatch_routes_cian_full_load(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Dispatch block в scheduler_loop роутит cian_full_load → trigger_cian_full_load_run."""
-    triggered: list[str] = []
-
-    async def fake_trigger(db, sch):
-        triggered.append(sch["source"])
-        return 1
-
-    monkeypatch.setattr(_sched, "trigger_cian_full_load_run", fake_trigger)
-
-    db = object()
-    sch = _CIAN_FULL_LOAD_ROW
-    source = sch["source"]
-    # Replicate the elif chain from scheduler_loop for cian_full_load
-    if source == "cian_full_load":
-        await _sched.trigger_cian_full_load_run(db, sch)
-
-    assert "cian_full_load" in triggered
-
-
-# ── avito_full_load: trigger_avito_full_load_run + dispatch ──────────────────
-
-
-_AVITO_FULL_LOAD_ROW = {
-    "source": "avito_full_load",
-    "window_start_hour": 13,
-    "window_end_hour": 15,
-    "default_params": {
-        "price_cap_per_bucket": 1400,
-        "concurrency": 5,
-        "request_delay_sec": 7.0,
-        "secondary_only": True,
-    },
-}
-
-
-@pytest.mark.asyncio
-async def test_trigger_avito_full_load_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
-    """trigger_avito_full_load_run claims run + launches run_avito_full_load via create_task."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"create_run": [], "load": []}
-
-    def fake_create_run(_d, *, source, params):
-        calls["create_run"].append(source)
-        db.running[source] = True
-        return 401
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    async def fake_full_load(run_db, *, run_id, **kwargs):
-        calls["load"].append((run_id, kwargs))
-
-    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    run_id = await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)
-
-    assert run_id == 401
-    assert calls["create_run"] == ["avito_full_load"]
-    # Drain event-loop so the asyncio.create_task runs
-    await asyncio.sleep(0)
-    assert len(calls["load"]) == 1
-    loaded_run_id, kwargs = calls["load"][0]
-    assert loaded_run_id == 401
-    # Params propagated from default_params; resume_run_id forced to None
-    assert kwargs["price_cap_per_bucket"] == 1400
-    assert kwargs["concurrency"] == 5
-    assert kwargs["request_delay_sec"] == 7.0
-    assert kwargs["secondary_only"] is True
-    assert kwargs["resume_run_id"] is None
-
-
-@pytest.mark.asyncio
-async def test_trigger_avito_full_load_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
-    """trigger_avito_full_load_run returns None when source already running."""
-    db = _FakeSchedulerDB()
-    db.running["avito_full_load"] = True
-
-    calls: dict[str, int] = {"n": 0}
-
-    def fake_create_run(_d, *, source, params):
-        calls["n"] += 1
-        return 999
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    result = await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)
-
-    assert result is None
-    assert calls["n"] == 0
-
-
-@pytest.mark.asyncio
-async def test_scheduler_dispatch_routes_avito_full_load(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Dispatch block в scheduler_loop роутит avito_full_load → trigger_avito_full_load_run."""
-    triggered: list[str] = []
-
-    async def fake_trigger(db, sch):
-        triggered.append(sch["source"])
-        return 1
-
-    monkeypatch.setattr(_sched, "trigger_avito_full_load_run", fake_trigger)
-
-    db = object()
-    sch = _AVITO_FULL_LOAD_ROW
-    source = sch["source"]
-    # Replicate the elif chain from scheduler_loop for avito_full_load
-    if source == "avito_city_sweep":
-        pass
-    elif source == "avito_full_load":
-        await _sched.trigger_avito_full_load_run(db, sch)
-
-    assert "avito_full_load" in triggered
-
-
-# ── avito_full_load incremental_days + avito_full_load_exhaustive split ───────
-
-
-@pytest.mark.asyncio
-async def test_trigger_avito_full_load_reads_incremental_days(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """incremental_days из default_params прокидывается в run_avito_full_load."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"load": []}
-
-    def fake_create_run(_d, *, source, params):
-        db.running[source] = True
-        return 410
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    async def fake_full_load(run_db, *, run_id, **kwargs):
-        calls["load"].append(kwargs)
-
-    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    row = {
-        **_AVITO_FULL_LOAD_ROW,
-        "default_params": {**_AVITO_FULL_LOAD_ROW["default_params"], "incremental_days": 2},
-    }
-    await _sched.trigger_avito_full_load_run(db, row)
-    await asyncio.sleep(0)
-
-    assert len(calls["load"]) == 1
-    assert calls["load"][0]["incremental_days"] == 2
-
-
-@pytest.mark.asyncio
-async def test_trigger_avito_full_load_no_incremental_days_is_none(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Без incremental_days в params → run_avito_full_load(incremental_days=None) (exhaustive)."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"load": []}
-
-    def fake_create_run(_d, *, source, params):
-        db.running[source] = True
-        return 411
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    async def fake_full_load(run_db, *, run_id, **kwargs):
-        calls["load"].append(kwargs)
-
-    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)
-    await asyncio.sleep(0)
-
-    assert len(calls["load"]) == 1
-    assert calls["load"][0]["incremental_days"] is None
-
-
-_AVITO_FULL_LOAD_EXHAUSTIVE_ROW = {
-    "source": "avito_full_load_exhaustive",
-    "window_start_hour": 13,
-    "window_end_hour": 15,
-    "default_params": {
-        "price_cap_per_bucket": 1400,
-        "concurrency": 1,
-        "request_delay_sec": 7.0,
-        "secondary_only": True,
-    },
-}
-
-
-@pytest.mark.asyncio
-async def test_trigger_avito_full_load_exhaustive_forces_none(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """avito_full_load_exhaustive форсит incremental_days=None даже если в params задан."""
-    db = _FakeSchedulerDB()
-    calls: dict[str, list] = {"load": []}
-
-    def fake_create_run(_d, *, source, params):
-        db.running[source] = True
-        return 420
-
-    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
-
-    async def fake_full_load(run_db, *, run_id, **kwargs):
-        calls["load"].append((run_id, kwargs))
-
-    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
-    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
-
-    # Даже если оператор случайно положит incremental_days в params — exhaustive игнорит.
-    row = {
-        **_AVITO_FULL_LOAD_EXHAUSTIVE_ROW,
-        "default_params": {
-            **_AVITO_FULL_LOAD_EXHAUSTIVE_ROW["default_params"],
-            "incremental_days": 5,
-        },
-    }
-    run_id = await _sched.trigger_avito_full_load_exhaustive_run(db, row)
-    await asyncio.sleep(0)
-
-    assert run_id == 420
-    assert len(calls["load"]) == 1
-    loaded_run_id, kwargs = calls["load"][0]
-    assert loaded_run_id == 420
-    assert kwargs["incremental_days"] is None
-    assert kwargs["concurrency"] == 1
-    assert kwargs["price_cap_per_bucket"] == 1400
-    assert kwargs["resume_run_id"] is None
-
-
-@pytest.mark.asyncio
-async def test_scheduler_dispatch_routes_avito_full_load_exhaustive(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Dispatch роутит avito_full_load_exhaustive → trigger_avito_full_load_exhaustive_run."""
-    triggered: list[str] = []
-
-    async def fake_trigger(db, sch):
-        triggered.append(sch["source"])
-        return 1
-
-    monkeypatch.setattr(_sched, "trigger_avito_full_load_exhaustive_run", fake_trigger)
-
-    db = object()
-    sch = _AVITO_FULL_LOAD_EXHAUSTIVE_ROW
-    source = sch["source"]
-    # Replicate the elif chain from scheduler_loop
-    if source == "avito_full_load":
-        pass
-    elif source == "avito_full_load_exhaustive":
-        await _sched.trigger_avito_full_load_exhaustive_run(db, sch)
-
-    assert "avito_full_load_exhaustive" in triggered
-
-
-# ---------------------------------------------------------------------------
-# #1182 Phase 2 — graceful SIGTERM-drain детей (detached run-задач)
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_drain_inflight_awaits_detached_cooperative_child() -> None:
-    """Гвоздь Phase 2: _drain_inflight ДОЖИДАЕТСЯ detached-ребёнка (как scrape-run,
-    отвязанный через _spawn_tracked) до его clean-exit, а НЕ хард-кансельит его.
-
-    Регрессия gap'а: coordinator раньше ждал только себя → asyncio.run teardown
-    хард-кансельил живых детей mid-await (#1182 failure mode).
-    """
-    clean_exit = asyncio.Event()  # == mark_done-эквивалент ребёнка
-
-    async def _child() -> None:
-        # Кооперативный ребёнок: крутится, пока не запрошен shutdown, потом выходит чисто.
-        while not sd.shutdown_requested():
-            await asyncio.sleep(0.01)
-        clean_exit.set()
-
-    child = sched._spawn_tracked(_child())  # detached, попадает в _inflight_tasks
-    await asyncio.sleep(0.02)
-    assert not child.done(), "ребёнок ещё бежит (shutdown не запрошен)"
-
-    sd.request_shutdown()
-    await asyncio.wait_for(sched._drain_inflight(), timeout=2.0)
-
-    assert clean_exit.is_set(), "drain дождался clean-exit ребёнка"
-    assert child.done()
-    assert not child.cancelled(), "ребёнок вышел сам, НЕ хард-cancel"
-
-
-@pytest.mark.asyncio
-async def test_drain_inflight_times_out_on_noncooperative_child(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Некооперативный ребёнок (игнорит shutdown) упирается в _CHILD_DRAIN_TIMEOUT_S и
-    остаётся бежать — отдаётся внешнему hard-cancel fallback (scheduler_main/docker)."""
-    monkeypatch.setattr(sched, "_CHILD_DRAIN_TIMEOUT_S", 0.05)
-
-    async def _stuck() -> None:
-        await asyncio.Event().wait()  # никогда не проверяет shutdown_requested()
-
-    child = sched._spawn_tracked(_stuck())
-    sd.request_shutdown()
-
-    await asyncio.wait_for(sched._drain_inflight(), timeout=2.0)
-
-    assert not child.done(), "drain НЕ повесил процесс — оставил ребёнка под hard-cancel"
-
-    # cleanup
-    child.cancel()
-    with pytest.raises(asyncio.CancelledError):
-        await child
-
-
-@pytest.mark.asyncio
-async def test_drain_inflight_noop_when_no_children() -> None:
-    """Без in-flight детей _drain_inflight() — мгновенный no-op (без сна/ожидания)."""
-    assert not sched._inflight_tasks
-    await asyncio.wait_for(sched._drain_inflight(), timeout=1.0)
-
-
-@pytest.mark.asyncio
-async def test_scheduler_loop_calls_drain_on_shutdown(monkeypatch: pytest.MonkeyPatch) -> None:
-    """scheduler_loop при выходе по SIGTERM-drain'у вызывает _drain_inflight (проводка
-    coordinator → дренаж детей). Initial sleep(30) и tick-sleep занулены, чтобы тест
-    был детерминированным и быстрым."""
-    drain_spy = AsyncMock()
-    monkeypatch.setattr(sched, "_drain_inflight", drain_spy)
-    monkeypatch.setattr(sched.asyncio, "sleep", AsyncMock())  # без реальных 30s/60s
-
-    sd.request_shutdown()  # первый же top-of-tick check → break
-
-    await asyncio.wait_for(sched.scheduler_loop(), timeout=2.0)
-
-    drain_spy.assert_awaited_once()

tradein-mvp/backend/tests/test_scheduler_main.py

@@ -56,15 +56,13 @@ async def test_run_cancels_cleanly(monkeypatch: pytest.MonkeyPatch) -> None:
     cancelled = asyncio.Event()
 
     async def _stub_loop() -> None:
-        """Короткая заглушка: ждёт отмены, имитируя бесконечный scheduler_loop."""
+        """Короткая заглушка: ждёт отмены, имитируя бесконечный kit scheduler_loop."""
         cancelled.set()
         await asyncio.sleep(999)
 
-    # Патчим источник импорта: _run() делает `from app.services.scheduler import scheduler_loop`
+    # #2397 Part C: _run() безусловно зовёт _run_kit_scheduler() (legacy fallback удалён) —
-    # на каждый вызов, поэтому патч модуля-источника достаточен.
+    # патчим саму функцию на модуле scheduler_main.
-    import app.services.scheduler as _sched_mod
+    monkeypatch.setattr(sm, "_run_kit_scheduler", _stub_loop)
-
-    monkeypatch.setattr(_sched_mod, "scheduler_loop", _stub_loop)
 
     task = asyncio.create_task(sm._run())
 
@@ -139,9 +137,8 @@ async def test_run_uses_request_shutdown_not_cancel(monkeypatch: pytest.MonkeyPa
         while not sd.shutdown_requested():
             await asyncio.sleep(0.01)
 
-    import app.services.scheduler as _sched_mod
+    # #2397 Part C: _run() безусловно зовёт _run_kit_scheduler() (legacy fallback удалён).
-
+    monkeypatch.setattr(sm, "_run_kit_scheduler", _coop_loop)
-    monkeypatch.setattr(_sched_mod, "scheduler_loop", _coop_loop)
 
     run_task = asyncio.create_task(sm._run())
     await asyncio.wait_for(started.wait(), timeout=2.0)

tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py

@@ -1,16 +1,13 @@
-"""Golden-parity: `scraper_kit.orchestration.scheduler` registry-dispatch ≡ старый if/elif.
+"""Smoke/shape tests for `scraper_kit.orchestration.scheduler` (#2136, post-strangler).
 
-Strangler-инвариант (#2136): registry-реестр (`build_registry` + `resolve_handler` +
+Kit — единственный scheduler-путь (#2397 Part C убрал legacy
-`_dispatch`) должен маршрутизировать те же source'ы к тем же обработчикам, что и боевой
+`app.services.scheduler.scheduler_loop` + trigger_*/get_due_schedules/_claim_run/
-27-веточный `if/elif` в `app.services.scheduler.scheduler_loop`, и сохранить критичную
+reap_zombies machinery; сравнивать «golden-parity» больше не с чем). Этот файл проверяет
-concurrency-логику (`_claim_run` advisory-lock + double-check, `reap_zombies` порог,
+форму и инварианты kit-реестра самого по себе:
-proxy_healthcheck sub-hourly post-claim).
 
-Метод:
+  1. REGISTRY: реальный `build_product_handlers()` (НЕ stub) + `build_registry` покрывает
-  1. ROUTING: для каждого source прогоняем ОДИН тик боевого `scheduler_loop` (все
+     каждый canonical scheduled source (kit-native + продуктовые + deactivate_stale_*
-     `trigger_*` замоканы) → фиксируем какой trigger сработал; параллельно `resolve_handler`
+     wildcard-члены); неизвестный source → None.
-     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.
@@ -25,12 +22,9 @@ 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
@@ -45,45 +39,41 @@ from scraper_kit.orchestration.scheduler import (
     resolve_handler,
 )
 
-import app.services.scheduler as old_sched
+# ── продуктовые source'ы (НЕ kit-native) — CANONICAL, поддерживается ВРУЧНУЮ ─────────
-
+# Источник истины: app.services.product_handlers.build_product_handlers() keys (минус
-# ── таблица маршрутизации: source → имя боевого trigger_* ─────────────────────
+# wildcard "deactivate_stale_*", раскрытый в 3 конкретных member-source'а) + семейство
-# (то, что `if/elif` в старом scheduler_loop выбирает для каждого source).
+# scrape_schedules-сидов (data/sql/*scrape_schedules*seed*.sql, 158_seed_proxy_healthcheck_
-SOURCE_TO_OLD_TRIGGER: dict[str, str] = {
+# schedule.sql, 162_seed_deals_freshness_monitor.sql). Этот список НЕ выводится из кода —
-    "avito_city_sweep": "trigger_avito_city_sweep_run",
+# добавил/убрал источник в product_handlers.py? Обнови и здесь, иначе
-    "avito_full_load": "trigger_avito_full_load_run",
+# test_real_build_product_handlers_covers_all_scheduled_sources ничего не поймает.
-    "avito_full_load_exhaustive": "trigger_avito_full_load_exhaustive_run",
+# (2026-07-04 deep-review #2397 Part C: список был устаревшим — не хватало
-    "avito_newbuilding_sweep": "trigger_avito_newbuilding_sweep_run",
+# cian_history_backfill/deals_freshness_monitor/osm_poi_ekb_refresh; исправлено.)
-    "yandex_city_sweep": "trigger_yandex_city_sweep_run",
+_PRODUCT_SOURCES: set[str] = {
-    "cian_history_backfill": "trigger_cian_backfill_run",
+    "cian_history_backfill",
-    "cian_city_sweep": "trigger_cian_city_sweep_run",
+    "rosreestr_dkp_import",
-    "cian_full_load": "trigger_cian_full_load_run",
+    "listing_source_snapshot",
-    "domclick_city_sweep": "trigger_domclick_city_sweep_run",
+    "asking_to_sold_ratio_refresh",
-    "rosreestr_dkp_import": "trigger_rosreestr_dkp_run",
+    "refresh_search_matview",
-    "listing_source_snapshot": "trigger_listing_source_snapshot_run",
+    "yandex_address_backfill",
-    "asking_to_sold_ratio_refresh": "trigger_asking_to_sold_ratio_run",
+    "deactivate_stale_avito",
-    "refresh_search_matview": "trigger_refresh_search_matview_run",
+    "deactivate_stale_yandex",
-    "yandex_address_backfill": "trigger_yandex_address_backfill_run",
+    "deactivate_stale_cian",
-    "deactivate_stale_avito": "trigger_deactivate_stale_run",
+    "sber_index_pull",
-    "deactivate_stale_yandex": "trigger_deactivate_stale_run",
+    "rosreestr_quarter_poll",
-    "deactivate_stale_cian": "trigger_deactivate_stale_run",
+    "deals_freshness_monitor",
-    "sber_index_pull": "trigger_sber_index_pull_run",
+    "newbuilding_enrich",
-    "rosreestr_quarter_poll": "trigger_rosreestr_quarter_poll_run",
+    "yandex_newbuilding_sweep",
-    "newbuilding_enrich": "trigger_newbuilding_enrich_run",
+    "geoportal_coords_backfill",
-    "yandex_newbuilding_sweep": "trigger_yandex_newbuilding_sweep_run",
+    "geocode_missing_listings",
-    "geoportal_coords_backfill": "trigger_geoportal_coords_backfill_run",
+    "avito_detail_backfill",
-    "geocode_missing_listings": "trigger_geocode_missing_listings_run",
+    "yandex_detail_backfill",
-    "avito_detail_backfill": "trigger_avito_detail_backfill_run",
+    "cadastral_geo_match",
-    "yandex_detail_backfill": "trigger_yandex_detail_backfill_run",
+    "osm_poi_ekb_refresh",
-    "cadastral_geo_match": "trigger_cadastral_geo_match_run",
+    "house_imv_backfill",
-    "house_imv_backfill": "trigger_house_imv_backfill_run",
+    "house_dedup_merge",
-    "house_dedup_merge": "trigger_house_dedup_merge_run",
+    "proxy_healthcheck",
-    "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",
@@ -129,7 +119,7 @@ def _build_recording_registry() -> tuple[dict[str, Handler], dict[str, MagicMock
 
     product: dict[str, Handler] = {}
     # продуктовые exact-match источники
-    for src in SOURCE_TO_OLD_TRIGGER:
+    for src in _PRODUCT_SOURCES:
         if src in _KIT_NATIVE_SOURCES or src.startswith("deactivate_stale_"):
             continue
         product[src] = Handler(_make_job(src), src)
@@ -139,56 +129,30 @@ def _build_recording_registry() -> tuple[dict[str, Handler], dict[str, MagicMock
     return build_registry(product), fired
 
 
-# ── 1. ROUTING parity ────────────────────────────────────────────────────────
+# ── 1. REGISTRY shape ────────────────────────────────────────────────────────
 
 
-async def _drive_old_one_tick(source: str) -> list[str]:
+def test_real_build_product_handlers_covers_all_scheduled_sources() -> None:
-    """Прогнать ОДИН тик боевого scheduler_loop для одного source, вернуть сработавшие trigger'ы."""
+    """Реальный (не stub!) build_product_handlers покрывает КАЖДЫЙ canonical source.
-    sch = _make_sched(source)
-    triggers = {name: AsyncMock(return_value=1) for name in _ALL_TRIGGER_NAMES}
 
-    with ExitStack() as es:
+    2026-07-04 deep-review #2397 Part C: старая версия этого теста строила registry ИЗ
-        es.enter_context(patch.object(old_sched, "reap_zombies", MagicMock()))
+    `_PRODUCT_SOURCES` через recording-stub (`_build_recording_registry`) и проверяла, что
-        es.enter_context(
+    тот же набор резолвится — тавтология, ничего не гоняющая через реальный код. Эта версия
-            patch.object(old_sched, "get_due_schedules", MagicMock(return_value=[sch]))
+    зовёт настоящий `product_handlers.build_product_handlers(ctx)` — если кто-то уронит
-        )
+    handler-entry в product_handlers.py (или source добавили в scrape_schedules seed, но
-        es.enter_context(patch.object(old_sched, "SessionLocal", MagicMock()))
+    забыли завести handler), тест падает.
-        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]
+    ctx=None безопасен: build_product_handlers его не замыкает при сборке dict (сами job'ы
+    получают ctx только во время dispatch) — см. её докстринг + test_deals_freshness_monitor.
+    """
+    from app.services.product_handlers import build_product_handlers
 
+    real_registry = build_registry(build_product_handlers(ctx=None))  # type: ignore[arg-type]
 
-@pytest.mark.parametrize("source", sorted(SOURCE_TO_OLD_TRIGGER))
+    for source in _PRODUCT_SOURCES | _KIT_NATIVE_SOURCES:
-async def test_routing_parity_per_source(source: str) -> None:
+        assert (
-    """Для каждого source: старый if/elif зовёт ожидаемый trigger, kit-реестр резолвит handler."""
+            resolve_handler(source, real_registry) is not None
-    expected_trigger = SOURCE_TO_OLD_TRIGGER[source]
+        ), f"real build_product_handlers()/build_registry() misses source={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:
@@ -342,7 +306,7 @@ def test_reap_zombies_counts_and_commits() -> None:
 
 
 def test_reap_zombies_threshold_constant() -> None:
-    assert kit_sched.ZOMBIE_THRESHOLD_HOURS == old_sched.ZOMBIE_THRESHOLD_HOURS == 6
+    assert kit_sched.ZOMBIE_THRESHOLD_HOURS == 6
 
 
 # ── 4. proxy_healthcheck sub-hourly post_claim ───────────────────────────────