feat(scheduler): add dormant Yandex Realty city-sweep (#561)

Wire a full Yandex.Недвижимость periodic city-sweep into the in-app
scheduler, shipped DISABLED (enabled=false) by design — live behavior is
intentionally unverified pending egress-IP rotation (#623). Mirrors the
existing avito_city_sweep (also currently disabled in prod).

- run_yandex_city_sweep in scrape_pipeline.py: mirrors run_avito_city_sweep
  but simpler (no houses/detail enrich) — search + save_listings per anchor,
  cooperative cancel, per-anchor exception isolation, defensive abort after
  3 consecutive anchor failures (mark_banned; YandexRealtyScraper has no
  block/rate exception classes — fetch_around swallows errors → []).
- trigger_yandex_city_sweep_run + dispatch branch in scheduler.py (faithful
  mirror of the avito trigger: claim run, fresh SessionLocal, create_task).
- Migration 078 seeds scrape_schedules row enabled=false, window 02:00-05:00
  UTC, conservative params {pages_per_anchor:2, request_delay_sec:9,
  radius_m:1500}, ON CONFLICT (source) DO NOTHING (idempotent).
- 8 unit tests (mocked scraper + save_listings, no live network/DB):
  anchor iteration, cancel mid-sweep, per-anchor isolation, consecutive-
  failure abort + reset-on-success, empty-page pagination break.

Enable later (after #623): UPDATE scrape_schedules SET enabled=true
WHERE source='yandex_city_sweep';

Closes #561

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

@@ -9,6 +9,7 @@
 
 Sources:
   - avito_city_sweep        → run_avito_city_sweep (scrape_pipeline.py)
+  - yandex_city_sweep       → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT)
   - cian_history_backfill   → run_cian_backfill (this module, #560)
   - rosreestr_dkp_import    → import_rosreestr_dkp (this module, #563)
 """
@@ -26,7 +27,7 @@ from sqlalchemy.orm import Session
 
 from app.core.db import SessionLocal
 from app.services import scrape_runs as runs_mod
-from app.services.scrape_pipeline import run_avito_city_sweep
+from app.services.scrape_pipeline import run_avito_city_sweep, run_yandex_city_sweep
 
 logger = logging.getLogger(__name__)
 
@@ -186,6 +187,44 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]
     return run_id
 
 
+async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать новый scrape_runs + launch run_yandex_city_sweep в asyncio.create_task.
+
+    Зеркало trigger_avito_city_sweep_run, но проще (#561): у Yandex sweep нет
+    houses/detail-параметров. Shipped DORMANT — schedule seed enabled=false (078),
+    включается оператором вручную после #623 (egress-IP rotation).
+
+    Returns run_id (или None если skip — есть running run).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            await run_yandex_city_sweep(
+                run_db,
+                run_id=run_id,
+                pages_per_anchor=int(params.get("pages_per_anchor", 2)),
+                request_delay_sec=float(params.get("request_delay_sec", 9.0)),
+                radius_m=int(params.get("radius_m", 1500)),
+            )
+        except Exception:
+            logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    # Keep reference to avoid GC before task completes (RUF006)
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered yandex_city_sweep run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch run_cian_backfill в asyncio.create_task.
 
@@ -592,6 +631,8 @@ async def scheduler_loop() -> None:
                     source = sch["source"]
                     if source == "avito_city_sweep":
                         await trigger_avito_city_sweep_run(db, sch)
+                    elif source == "yandex_city_sweep":
+                        await trigger_yandex_city_sweep_run(db, sch)
                     elif source == "cian_history_backfill":
                         await trigger_cian_backfill_run(db, sch)
                     elif source == "rosreestr_dkp_import":

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

@@ -40,6 +40,7 @@ from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichm
 from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
 from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment
 from app.services.scrapers.base import ScrapedLot, save_listings
+from app.services.scrapers.yandex_realty import YandexRealtyScraper
 
 logger = logging.getLogger(__name__)
 
@@ -424,12 +425,165 @@ async def run_avito_city_sweep(
             raise
 
 
+# ── Yandex city sweep ───────────────────────────────────────────
+
+
+@dataclass
+class YandexCitySweepCounters:
+    """Aggregate counters для Yandex.Недвижимость city sweep run.
+
+    Проще avito CitySweepCounters: у Yandex SERP нет house-catalog / detail-enrich
+    шага (только search → save), поэтому houses/detail-полей нет.
+    """
+
+    anchors_total: int = 0
+    anchors_done: int = 0
+    pages_fetched: int = 0
+    lots_fetched: int = 0
+    lots_inserted: int = 0
+    lots_updated: int = 0
+    errors_count: int = 0
+
+    def to_dict(self) -> dict[str, int]:
+        return {f.name: getattr(self, f.name) for f in fields(self)}
+
+
+# Defensive abort threshold: у YandexRealtyScraper нет block/rate-exception классов
+# (fetch_around глотает ошибки и возвращает []). Чтобы не молотить Yandex впустую при
+# системном блоке/бане IP, прерываем sweep после N подряд anchor'ов, упавших с
+# исключением. NB: пустой результат (0 lots) — это НЕ ошибка, счётчик не растёт.
+YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES = 3
+
+
+async def run_yandex_city_sweep(
+    db: Session,
+    *,
+    run_id: int,
+    anchors: list[tuple[float, float, str]] | None = None,
+    radius_m: int = 1500,
+    pages_per_anchor: int = 2,
+    request_delay_sec: float | None = None,
+) -> YandexCitySweepCounters:
+    """Yandex.Недвижимость city sweep: iterate anchors × pages → save.
+
+    Проще avito sweep — у Yandex SERP нет house-catalog / detail-enrich шага:
+      per anchor: YandexRealtyScraper().fetch_around(lat, lon, radius_m, page=p)
+      → save_listings(db, lots) → accumulate counters.
+
+    Anti-bot / cancel семантика (зеркало avito sweep, см. #561):
+      - Cooperative cancel: is_cancelled(db, run_id) проверяется перед каждым anchor;
+        при cancel — update_heartbeat + return (mark_cancelled делает caller-flow по статусу).
+      - YandexRealtyScraper НЕ определяет block/rate exception классов (fetch_around
+        ловит исключения внутри и возвращает []). Поэтому per-anchor ловим broad
+        Exception, логируем и continue; защитно прерываем sweep (early abort + mark_banned),
+        если N anchor'ов ПОДРЯД упали с исключением — это сигнал системного блока IP.
+      - request_delay_sec: YandexRealtyScraper.fetch_around НЕ принимает delay-override
+        (он сам спит self.request_delay_sec между страницами через sleep_between_requests).
+        Поэтому request_delay_sec применяется здесь как доп. asyncio.sleep МЕЖДУ anchor'ами.
+
+    Возвращает YandexCitySweepCounters (агрегат по всем anchor'ам).
+    """
+    _anchors = anchors if anchors is not None else EKB_ANCHORS
+    counters = YandexCitySweepCounters(anchors_total=len(_anchors))
+    inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
+    consecutive_failures = 0
+
+    try:
+        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
+            if scrape_runs.is_cancelled(db, run_id):
+                logger.info(
+                    "yandex-sweep run_id=%d: cancelled at anchor #%d/%d (%s)",
+                    run_id, idx, len(_anchors), name,
+                )
+                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                return counters
+
+            logger.info(
+                "yandex-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
+                run_id, idx, len(_anchors), name, lat, lon,
+            )
+            try:
+                # Fresh scraper per anchor — uses its own httpx.AsyncClient via the
+                # async-context-manager. fetch_around fetches ONE SERP page; loop pages.
+                anchor_lots: list[ScrapedLot] = []
+                async with YandexRealtyScraper() as scraper:
+                    for page in range(pages_per_anchor):
+                        page_lots = await scraper.fetch_around(
+                            lat, lon, radius_m, page=page
+                        )
+                        counters.pages_fetched += 1
+                        if not page_lots:
+                            # Пустая страница → дальше результатов нет, прекращаем пагинацию.
+                            break
+                        anchor_lots.extend(page_lots)
+
+                counters.lots_fetched += len(anchor_lots)
+                if anchor_lots:
+                    inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
+                    counters.lots_inserted += inserted
+                    counters.lots_updated += updated
+                consecutive_failures = 0
+            except Exception as e:
+                # YandexRealtyScraper не propagate'ит block/rate exceptions, поэтому здесь
+                # ловим broad Exception per-anchor: логируем и continue (изоляция anchor'а).
+                logger.exception(
+                    "yandex-sweep run_id=%d: anchor %s failed", run_id, name
+                )
+                counters.errors_count += 1
+                consecutive_failures += 1
+                if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES:
+                    logger.error(
+                        "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
+                        "%d consecutive failures (IP likely blocked): %s",
+                        run_id, idx, len(_anchors), name, consecutive_failures, e,
+                    )
+                    counters.anchors_done = idx
+                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                    scrape_runs.mark_banned(
+                        db,
+                        run_id,
+                        f"yandex sweep aborted: {consecutive_failures} consecutive "
+                        f"anchor failures (last: {e})",
+                        counters.to_dict(),
+                    )
+                    return counters
+
+            counters.anchors_done = idx
+            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+            # Доп. пауза между anchor'ами (поверх per-page sleep внутри scraper'а).
+            if idx < len(_anchors):
+                await asyncio.sleep(inter_anchor_delay)
+
+        scrape_runs.mark_done(db, run_id, counters.to_dict())
+        logger.info(
+            "yandex-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d "
+            "(ins=%d/upd=%d) errors=%d",
+            run_id,
+            counters.anchors_done,
+            counters.anchors_total,
+            counters.pages_fetched,
+            counters.lots_fetched,
+            counters.lots_inserted,
+            counters.lots_updated,
+            counters.errors_count,
+        )
+        return counters
+
+    except Exception as exc:
+        logger.exception("yandex-sweep run_id=%d: fatal error", run_id)
+        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+        raise
+
+
 # ── Public re-exports ───────────────────────────────────────────
 __all__ = [
     "EKB_ANCHORS",
     "CitySweepCounters",
     "PipelineCounters",
     "PipelineResult",
+    "YandexCitySweepCounters",
     "run_avito_city_sweep",
     "run_avito_pipeline",
+    "run_yandex_city_sweep",
 ]

tradein-mvp/backend/data/sql/078_scrape_schedules_seed_yandex_sweep.sql

@@ -0,0 +1,55 @@
+-- 078_scrape_schedules_seed_yandex_sweep.sql
+-- Seed row для Yandex.Недвижимость периодического city-sweep (#561).
+--
+-- !!! DORMANT BY DESIGN !!! enabled = false.
+--   Sweep будет бить Yandex с prod-IP → anti-bot риск ("осторожнее с лимитами").
+--   Capability полностью wired (run_yandex_city_sweep + trigger + dispatch),
+--   но schedule засеян ВЫКЛЮЧЕННЫМ. Live-поведение намеренно НЕ verified.
+--   ВКЛЮЧАТЬ ВРУЧНУЮ оператором ПОСЛЕ #623 (egress-IP rotation):
+--     UPDATE scrape_schedules SET enabled = true WHERE source = 'yandex_city_sweep';
+--   Зеркалит existing avito_city_sweep (тоже сейчас enabled = f в prod).
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)).
+-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно.
+--
+-- yandex_city_sweep — окно 02:00-05:00 UTC (05:00-08:00 МСК), как avito_city_sweep.
+--   default_params (консервативно, под анти-бот):
+--     pages_per_anchor — страниц SERP на anchor (мало, чтобы не отсвечивать).
+--     request_delay_sec — пауза между anchor'ами (поверх per-page sleep scraper'а).
+--     radius_m — прокидывается в fetch_around (Yandex его игнорит: path-based URL,
+--                держим для совместимости сигнатуры avito-shape sweep).
+--   У Yandex sweep НЕТ detail/houses-enrich шага — поэтому detail_top_n/enrich_houses
+--   в params отсутствуют (в отличие от avito_city_sweep).
+
+BEGIN;
+
+-- next_run_at = следующее наступление night-window (tomorrow + window_start_hour).
+-- При enabled=false get_due_schedules() всё равно не подхватит строку, но заполняем
+-- по образцу 072: когда оператор включит schedule, next_run_at уже валидный и
+-- запуск не выстрелит мгновенно (вне окна). AT TIME ZONE 'UTC' фиксирует wall-clock
+-- час в UTC независимо от session TimeZone GUC.
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'yandex_city_sweep',
+    false,                                       -- DORMANT (#561) — enable manually after #623
+    2,
+    5,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC',
+    '{"pages_per_anchor": 2, "request_delay_sec": 9, "radius_m": 1500}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет cron-script setup). '
+    'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
+    'cian_history_backfill, rosreestr_dkp_import.';
+
+COMMIT;

tradein-mvp/backend/tests/test_yandex_city_sweep.py

@@ -0,0 +1,372 @@
+"""Offline unit tests for run_yandex_city_sweep (#561, shipped DORMANT).
+
+Pure & fast: NO live network, NO DB. We monkeypatch:
+  - YandexRealtyScraper.__init__ / __aenter__ / __aexit__ — avoid get_scraper_delay
+    DB read + real httpx.AsyncClient creation.
+  - YandexRealtyScraper.fetch_around — return a fixed list of fake ScrapedLot.
+  - scrape_pipeline.save_listings — counter/no-op (no real listings table).
+  - scrape_runs.* (is_cancelled / update_heartbeat / mark_*) — in-memory fakes.
+
+These tests assert orchestration semantics only; live Yandex behavior is intentionally
+unverified (sweep is dormant until egress-IP rotation #623).
+"""
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+
+import pytest
+
+from app.services import scrape_pipeline
+from app.services.scrapers.base import ScrapedLot
+from app.services.scrapers.yandex_realty import YandexRealtyScraper
+
+# 3 anchors keep tests fast and let us probe "anchor k cancels/fails, others proceed".
+TEST_ANCHORS: list[tuple[float, float, str]] = [
+    (56.8400, 60.6050, "A1"),
+    (56.7950, 60.5300, "A2"),
+    (56.8970, 60.6100, "A3"),
+]
+
+
+def _fake_lot(offer_id: str) -> ScrapedLot:
+    """Minimal valid ScrapedLot (source/source_url required, price_rub > 0)."""
+    return ScrapedLot(
+        source="yandex",
+        source_url=f"https://realty.yandex.ru/offer/{offer_id}/",
+        source_id=offer_id,
+        address="Екатеринбург, улица Тестовая, 1",
+        price_rub=5_000_000,
+        area_m2=50.0,
+        rooms=2,
+    )
+
+
+class _FakeRuns:
+    """In-memory stand-in for app.services.scrape_runs (no DB)."""
+
+    def __init__(self, *, cancel_at_anchor: int | None = None) -> None:
+        # cancel_at_anchor: 1-based anchor index at which is_cancelled flips True.
+        self.cancel_at_anchor = cancel_at_anchor
+        self.is_cancelled_calls = 0
+        self.heartbeats: list[dict[str, int]] = []
+        self.done: dict[str, int] | None = None
+        self.failed: tuple[str, dict[str, int]] | None = None
+        self.banned: tuple[str, dict[str, int]] | None = None
+
+    def is_cancelled(self, _db: Any, _run_id: int) -> bool:
+        self.is_cancelled_calls += 1
+        if self.cancel_at_anchor is None:
+            return False
+        return self.is_cancelled_calls >= self.cancel_at_anchor
+
+    def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.heartbeats.append(dict(counters))
+
+    def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.done = dict(counters)
+
+    def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.failed = (error, dict(counters))
+
+    def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.banned = (error, dict(counters))
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kill inter-anchor asyncio.sleep so tests are instant."""
+    async def _instant(_secs: float) -> None:
+        return None
+
+    monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant)
+
+
+@pytest.fixture(autouse=True)
+def _stub_scraper_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Neutralize __init__/__aenter__/__aexit__ — no get_scraper_delay DB, no httpx."""
+
+    def _init(self: YandexRealtyScraper, city: str = "ekaterinburg") -> None:
+        self.city = city
+        self.request_delay_sec = 0.0
+        self._client = None
+
+    async def _aenter(self: YandexRealtyScraper) -> YandexRealtyScraper:
+        return self
+
+    async def _aexit(self: YandexRealtyScraper, *_args: Any) -> None:
+        return None
+
+    monkeypatch.setattr(YandexRealtyScraper, "__init__", _init)
+    monkeypatch.setattr(YandexRealtyScraper, "__aenter__", _aenter)
+    monkeypatch.setattr(YandexRealtyScraper, "__aexit__", _aexit)
+
+
+def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None:
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned)
+
+
+# ── Happy path: all anchors iterated, counters aggregated ───────────────────
+
+
+async def test_sweep_iterates_all_anchors_and_aggregates(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fetch_calls: list[tuple[float, float, int]] = []
+    save_calls: list[int] = []
+
+    async def fake_fetch(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        page: int = 0,
+    ) -> list[ScrapedLot]:
+        fetch_calls.append((lat, lon, page))
+        # 2 lots per page; let pagination run the full pages_per_anchor.
+        return [_fake_lot(f"{lat}-{page}-{i}") for i in range(2)]
+
+    def fake_save(
+        _db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
+    ) -> tuple[int, int]:
+        save_calls.append(len(lots))
+        # Pretend everything is a fresh insert.
+        return len(lots), 0
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=object(),
+        run_id=1,
+        anchors=TEST_ANCHORS,
+        pages_per_anchor=2,
+        request_delay_sec=0.0,
+    )
+
+    # fetch_around called pages_per_anchor times per anchor (no empty-page short-circuit).
+    assert len(fetch_calls) == len(TEST_ANCHORS) * 2
+    # save_listings called once per anchor (anchor-level batch).
+    assert len(save_calls) == len(TEST_ANCHORS)
+    # Counters: 3 anchors × 2 pages × 2 lots = 12.
+    assert counters.anchors_total == len(TEST_ANCHORS)
+    assert counters.anchors_done == len(TEST_ANCHORS)
+    assert counters.pages_fetched == len(TEST_ANCHORS) * 2
+    assert counters.lots_fetched == 12
+    assert counters.lots_inserted == 12
+    assert counters.lots_updated == 0
+    assert counters.errors_count == 0
+    # mark_done called with final counters; no failure/ban.
+    assert fake.done is not None
+    assert fake.done["lots_fetched"] == 12
+    assert fake.failed is None
+    assert fake.banned is None
+
+
+async def test_save_listings_receives_run_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    """run_id must be threaded into save_listings for snapshot provenance."""
+    seen_run_ids: list[int | None] = []
+
+    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+        return [_fake_lot(f"{lat}-{page}")]
+
+    def fake_save(_db, lots, *, run_id=None):
+        seen_run_ids.append(run_id)
+        return len(lots), 0
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+    _install_runs(monkeypatch, _FakeRuns())
+
+    await scrape_pipeline.run_yandex_city_sweep(
+        db=object(), run_id=777, anchors=TEST_ANCHORS, pages_per_anchor=1,
+    )
+    assert seen_run_ids == [777, 777, 777]
+
+
+# ── Empty page short-circuits pagination ────────────────────────────────────
+
+
+async def test_empty_page_stops_pagination(monkeypatch: pytest.MonkeyPatch) -> None:
+    """An empty SERP page ends pagination for that anchor (no further pages)."""
+    pages_seen: list[int] = []
+
+    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+        pages_seen.append(page)
+        if page == 0:
+            return [_fake_lot(f"{lat}-0")]
+        return []  # page 1 empty → stop
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=object(), run_id=2, anchors=TEST_ANCHORS, pages_per_anchor=5,
+    )
+    # Per anchor: page 0 (1 lot) then page 1 (empty, break) → 2 pages each.
+    assert pages_seen == [0, 1, 0, 1, 0, 1]
+    assert counters.pages_fetched == 6  # 2 per anchor × 3
+    assert counters.lots_fetched == 3  # 1 per anchor
+    assert counters.anchors_done == 3
+
+
+# ── Mid-sweep cancel stops further anchors ──────────────────────────────────
+
+
+async def test_cancel_midway_stops_remaining_anchors(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fetch_calls: list[str] = []
+
+    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+        fetch_calls.append(f"{lat}")
+        return [_fake_lot(f"{lat}-{page}")]
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    # is_cancelled flips True on the 2nd check (before anchor #2) → only anchor #1 runs.
+    fake = _FakeRuns(cancel_at_anchor=2)
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=object(), run_id=3, anchors=TEST_ANCHORS, pages_per_anchor=1,
+    )
+    # Only the first anchor fetched before cancel.
+    assert len(fetch_calls) == 1
+    assert counters.anchors_done == 1
+    # Cancel path returns early: no mark_done, no ban/fail.
+    assert fake.done is None
+    assert fake.failed is None
+    assert fake.banned is None
+
+
+# ── Per-anchor exception is isolated (single failure → continue) ────────────
+
+
+async def test_single_anchor_failure_is_isolated(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Anchor #2 raises → anchors #1 and #3 still processed; sweep completes."""
+    processed: list[str] = []
+
+    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+        if name_of(lat) == "A2":
+            raise RuntimeError("boom on anchor 2")
+        processed.append(name_of(lat))
+        return [_fake_lot(f"{lat}-{page}")]
+
+    def name_of(lat: float) -> str:
+        for a_lat, _lon, nm in TEST_ANCHORS:
+            if a_lat == lat:
+                return nm
+        return "?"
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=object(), run_id=4, anchors=TEST_ANCHORS, pages_per_anchor=1,
+    )
+    # A1 and A3 processed; A2 raised and was skipped.
+    assert processed == ["A1", "A3"]
+    assert counters.errors_count == 1
+    assert counters.anchors_done == 3  # loop advanced past all 3 anchors
+    assert counters.lots_fetched == 2  # only A1 + A3 contributed lots
+    # Not an abort — sweep marked done despite the isolated failure.
+    assert fake.done is not None
+    assert fake.banned is None
+
+
+# ── Defensive abort: N consecutive failures → mark_banned, stop early ───────
+
+
+async def test_consecutive_failures_abort_sweep(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """All anchors raise → after MAX_CONSECUTIVE_FAILURES the sweep aborts + bans."""
+    attempts: list[str] = []
+
+    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+        attempts.append(f"{lat}")
+        raise RuntimeError("yandex blocked")
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    # 5 anchors so the 3-consecutive-failure threshold trips before the list ends.
+    anchors = [
+        *TEST_ANCHORS,
+        (56.7700, 60.5500, "A4"),
+        (56.8650, 60.6200, "A5"),
+    ]
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=object(), run_id=5, anchors=anchors, pages_per_anchor=1,
+    )
+    # Abort exactly at the threshold (3 consecutive failures), not all 5 anchors.
+    assert len(attempts) == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES
+    assert counters.errors_count == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES
+    assert counters.anchors_done == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES
+    assert fake.banned is not None
+    assert "consecutive" in fake.banned[0]
+    assert fake.done is None  # aborted, never marked done
+
+
+async def test_non_consecutive_failures_do_not_abort(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A success between two failures resets the consecutive counter → no abort."""
+    def name_of(lat: float, anchors: list[tuple[float, float, str]]) -> str:
+        for a_lat, _lon, nm in anchors:
+            if a_lat == lat:
+                return nm
+        return "?"
+
+    anchors = [
+        *TEST_ANCHORS,
+        (56.7700, 60.5500, "A4"),
+        (56.8650, 60.6200, "A5"),
+    ]
+    # Fail A1, succeed A2, fail A3, succeed A4, fail A5 → never 3-in-a-row.
+    failing = {"A1", "A3", "A5"}
+
+    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+        if name_of(lat, anchors) in failing:
+            raise RuntimeError("transient")
+        return [_fake_lot(f"{lat}-{page}")]
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=object(), run_id=6, anchors=anchors, pages_per_anchor=1,
+    )
+    assert counters.errors_count == 3  # A1, A3, A5
+    assert counters.anchors_done == 5  # walked the whole list, no early abort
+    assert counters.lots_fetched == 2  # A2 + A4 contributed
+    assert fake.banned is None
+    assert fake.done is not None