fix(scrapers): isolate flats from poison-extras to restore KN flats_count (#1945)

ГЛАВНЫЙ unblock. Эмпирически (prod worker):
  • flats endpoint /portal-kn/api/sales/portal/table @ concurrency=2 — 0 WAF-бан
    на 250 объектах / 251 req, 77163 flat-строк (vs текущие 9 во всём sweep);
  • extras /сервисы/api/object/{id}/* отдают 403 СРАЗУ (volume-independent, мертвы
    с 2026-06-03) И травят cookies сессии → последующие flats на ней тоже 403.
flats_count рухнул 3670→9 именно из-за shared-session отравления.

Fix: flats и extras больше НИКОГДА не делят сессию.
- run_region_sweep(extras_isolated=True, дефолт): warm flats-сессия тянет ТОЛЬКО
  flats (extras на ней не вызываются) → flats_count восстанавливается; extras
  уходят отдельным best-effort проходом _run_extras_pass с recycle BrowserSession
  на каждый WAF-403 (яд не накапливается, flats-сессия уже закрыта).
- _run_extras_pass: ранний обрыв после N подряд полностью-WAF объектов
  (_EXTRAS_ABORT_AFTER_CONSEC_WAF=5) — extras мертвы целиком, нет смысла жечь
  ~1500 релончей браузера. Любой сбой паса → log, run НЕ падает.
- extras_isolated=False → старое combined-поведение (poison-prone).
- config: scrape_kn_extras_isolated=True (ENV SCRAPE_KN_EXTRAS_ISOLATED);
  прокинут через оба worker call-site.
- рефактор: _process_flats_result / _process_extras_result / _extras_coros
  делят upsert-логику между combined-loop и isolated-pass.
- tests: _extras_coros без flats, _is_waf_poisoned, recycle-на-poison,
  early-abort на wholesale-WAF.

Сохраняет throttle + optional-proxy из предыдущего коммита.

backend/app/core/config.py

@@ -140,6 +140,19 @@ class Settings(BaseSettings):
     # (SCRAPER_PROXY_URL). Заводится ТОЛЬКО через окружение прод-контейнера.
     # ENV: SCRAPE_KN_PROXY_URL.
     scrape_kn_proxy_url: str | None = None
+    # Рычаг 3 (ГЛАВНЫЙ unblock #1945) — изоляция flats от extras.
+    # Эмпирически (prod, 2026-06-27): flats endpoint /portal-kn/api/sales/portal/table
+    # на concurrency=2 = 0 WAF-бан на 100+ объектах; extras /сервисы/api/object/{id}/*
+    # отдают 403 СРАЗУ (volume-independent, мертвы с 2026-06-03) И ТРАВЯТ cookies
+    # сессии → последующие flats на той же сессии тоже 403. flats_count (метрика
+    # #1945) рухнул 3670→9 именно из-за этого.
+    #
+    # True (дефолт, NEW): flats тянутся в ЧИСТОЙ flats-only сессии (extras на ней
+    # НЕ дёргаются НИКОГДА) → flats_count восстанавливается; extras идут отдельным
+    # best-effort проходом с recycle сессии на каждый WAF-403 (яд не накапливается
+    # и НЕ касается flats). False: старое поведение (flats+extras в одной сессии,
+    # poison-prone). ENV: SCRAPE_KN_EXTRAS_ISOLATED.
+    scrape_kn_extras_isolated: bool = True
 
     # NSPD-scraper (Playwright) УДАЛЁН 2026-05-11. Сменён на bulk geo-fetcher
     # через rosreestr2coord — запускается вручную через /admin/scrape/geo.

backend/app/services/scrapers/domrf_kn.py

@@ -33,7 +33,7 @@ from app.services.scrapers.documents import extract_documents, upsert_documents
 # obj_checks import temporarily disabled — endpoint /checks returns 404 (run #19).
 # Re-enable with _fetch_obj_checks_safe when endpoint is found (see TODO in Phase B/C).
 # from app.services.scrapers.obj_checks import extract_obj_checks, upsert_obj_checks
-from app.services.scrapers.stealth import BASE_URL, BrowserSession
+from app.services.scrapers.stealth import BASE_URL, BrowserSession, WafBlockedError
 
 logger = logging.getLogger(__name__)
 
@@ -1450,6 +1450,247 @@ async def download_photos(
     return locals_out, thumbs_out
 
 
+# ── result processing (shared by combined + isolated #1945 passes) ────────────
+
+# #1945 isolated extras pass: после стольких подряд полностью-WAF объектов
+# обрываем пас (extras заблокированы целиком — нет смысла жечь ~1500 recycle).
+_EXTRAS_ABORT_AFTER_CONSEC_WAF = 5
+
+# 5 doc-endpoint kinds, accumulated into a single per-object documents upsert.
+_DOC_KINDS = frozenset(
+    (
+        "doc_rpd",
+        "doc_developer_report",
+        "doc_project_documentation",
+        "doc_documentation_other",
+        "doc_permits",
+    )
+)
+
+
+def _process_flats_result(
+    db: Session,
+    run_id: int,
+    obj_id: int,
+    full_url: str,
+    result: list[dict[str, Any]] | Exception,
+    snapshot_date: date,
+    region_code: int,
+) -> int:
+    """Upsert flats from a _fetch_flats_safe result. Returns flats inserted.
+
+    Errors → kn_scrape_failures (via _classify_and_log), 0 returned. Вынесено из
+    Phase-B/C loop чтобы isolated-flats-pass (#1945) и combined-loop делили один код.
+    """
+    if isinstance(result, Exception):
+        _classify_and_log(db, run_id, obj_id, "flats", full_url, result)
+        return 0
+    if result:
+        return upsert_flats(db, result, snapshot_date, region_code)
+    return 0
+
+
+async def _process_extras_result(
+    sess: BrowserSession,
+    db: Session,
+    run_id: int,
+    obj_id: int,
+    kind_tag: str,
+    full_url: str,
+    result: Any,
+    snapshot_date: date,
+    extras_counts: dict[str, int],
+    all_docs: list[dict[str, Any]],
+    *,
+    download_photos_binary: bool,
+    pdir: Path,
+) -> None:
+    """Handle one extras endpoint result (sale_graph/sales_agg/infra/photos/docs).
+
+    Mutates extras_counts / all_docs in place. Errors → kn_scrape_failures.
+    Doc-kinds накапливаются в all_docs (единый upsert делает вызывающий код).
+    Вынесено из loop чтобы combined- и isolated-extras-pass (#1945) делили код.
+    """
+    if isinstance(result, Exception):
+        _classify_and_log(db, run_id, obj_id, kind_tag, full_url, result)
+        return
+
+    if kind_tag in ("sale_graph_apartments", "sale_graph_parking"):
+        sg_type = kind_tag.replace("sale_graph_", "")
+        rows_sg, _ = result
+        extras_counts["sale_graph_rows"] += upsert_sale_graph(
+            db, obj_id, sg_type, rows_sg, snapshot_date
+        )
+    elif kind_tag == "sales_agg":
+        agg_data, _ = result
+        extras_counts["sales_agg_rows"] += upsert_sales_agg(db, obj_id, agg_data, snapshot_date)
+    elif kind_tag == "infrastructure":
+        pois_data, _ = result
+        extras_counts["infra_rows"] += upsert_infrastructure(db, obj_id, pois_data, snapshot_date)
+    elif kind_tag == "photos":
+        photos_data, _ = result
+        local_paths: dict[str, str] = {}
+        thumb_paths: dict[str, str] = {}
+        if download_photos_binary and photos_data:
+            local_paths, thumb_paths = await download_photos(sess, obj_id, photos_data, pdir)
+            extras_counts["photos_downloaded"] += len(local_paths)
+        extras_counts["photos_rows"] += upsert_photos(
+            db, obj_id, photos_data, local_paths, thumb_paths
+        )
+    elif kind_tag in _DOC_KINDS:
+        doc_items, _ = result
+        all_docs.extend(extract_documents(doc_items or []))
+
+
+def _extras_coros(sess: BrowserSession, obj_id: int) -> list[Any]:
+    """The 10 extras-endpoint coroutines for one object (flats NOT included)."""
+    return [
+        _fetch_sale_graph_safe(sess, obj_id, "apartments"),
+        _fetch_sale_graph_safe(sess, obj_id, "parking"),
+        _fetch_sales_agg_safe(sess, obj_id),
+        _fetch_infrastructure_safe(sess, obj_id),
+        _fetch_photos_safe(sess, obj_id),
+        _fetch_doc_rpd_safe(sess, obj_id),
+        _fetch_doc_developer_report_safe(sess, obj_id),
+        _fetch_doc_project_documentation_safe(sess, obj_id),
+        _fetch_doc_documentation_other_safe(sess, obj_id),
+        _fetch_doc_permits_safe(sess, obj_id),
+    ]
+
+
+def _is_waf_poisoned(results: list[tuple[str, str, Any]]) -> bool:
+    """Любой WafBlockedError среди результатов = сессия отравлена (cookies флагнуты).
+
+    После этого ВСЕ запросы на этой сессии (incl. следующего объекта) пойдут в 403,
+    поэтому вызывающий код должен пересоздать BrowserSession перед продолжением.
+    """
+    return any(isinstance(r, WafBlockedError) for _kind, _url, r in results)
+
+
+async def _run_extras_pass(
+    *,
+    db: Session,
+    run_id: int,
+    region_code: int,
+    all_objects: list[dict[str, Any]],
+    start_index: int,
+    snapshot_date: date,
+    extras_counts: dict[str, int],
+    pdir: Path,
+    download_photos_binary: bool,
+    load_state: str | None,
+    headed: bool,
+    browser_concurrency: int | None,
+    request_jitter_min_ms: int | None,
+    request_jitter_max_ms: int | None,
+    proxy_url: str | None,
+) -> int:
+    """#1945 isolated extras pass: extras в ОТДЕЛЬНОЙ BrowserSession от flats.
+
+    Ключ изоляции: на ЛЮБОЙ WafBlockedError (extras-403 травит cookies сессии)
+    закрываем сессию и открываем НОВУЮ (свежие cookies/контекст) перед следующим
+    объектом — яд не накапливается и НИКОГДА не касается flats (flats-сессия уже
+    закрыта). Best-effort: каждый сбой → kn_scrape_failures, run не падает.
+
+    Возвращает суммарный request_count всех extras-сессий (для метрики run'а).
+    """
+    total = len(all_objects)
+    total_requests = 0
+    log_progress(
+        db, run_id, f"extras-пас (isolated): старт {total - start_index} объектов", stage="extras"
+    )
+
+    def _new_session() -> BrowserSession:
+        return BrowserSession(
+            region_code=region_code,
+            headed=headed,
+            load_state=load_state,
+            concurrency=browser_concurrency,
+            jitter_min_ms=request_jitter_min_ms,
+            jitter_max_ms=request_jitter_max_ms,
+            proxy_url=proxy_url,
+        )
+
+    # extras сейчас мертвы (403 с 2026-06-03). Recycle на КАЖДЫЙ объект (полный
+    # перезапуск браузера ~1500×) бессмыслен, если extras заблокированы целиком.
+    # Поэтому считаем подряд идущие полностью-WAF объекты и обрываем пас после
+    # порога — extras остаются best-effort opt-in, но не жгут час впустую.
+    consecutive_all_waf = 0
+    aborted = False
+    sess = _new_session()
+    await sess.__aenter__()
+    try:
+        await sess.warm_up()
+        for i in range(start_index, total):
+            obj = all_objects[i]
+            obj_id = obj.get("objId") or obj.get("obj_id")
+            if not obj_id:
+                continue
+
+            results = await asyncio.gather(*_extras_coros(sess, obj_id), return_exceptions=False)
+            all_docs: list[dict[str, Any]] = []
+            for kind_tag, full_url, result in results:
+                await _process_extras_result(
+                    sess,
+                    db,
+                    run_id,
+                    obj_id,
+                    kind_tag,
+                    full_url,
+                    result,
+                    snapshot_date,
+                    extras_counts,
+                    all_docs,
+                    download_photos_binary=download_photos_binary,
+                    pdir=pdir,
+                )
+            if all_docs:
+                ins, _skip = upsert_documents(db, obj_id, all_docs)
+                extras_counts["documents_rows"] += ins
+
+            poisoned = _is_waf_poisoned(results)
+            # Полностью-WAF объект = ни один extras не отдал JSON (все 403).
+            all_waf = all(isinstance(r, WafBlockedError) for _k, _u, r in results)
+            consecutive_all_waf = consecutive_all_waf + 1 if all_waf else 0
+            if consecutive_all_waf >= _EXTRAS_ABORT_AFTER_CONSEC_WAF:
+                aborted = True
+                log_progress(
+                    db,
+                    run_id,
+                    f"extras-пас оборван: {consecutive_all_waf} объектов подряд"
+                    " полностью WAF-403 (extras заблокированы целиком)",
+                    level="warn",
+                    stage="extras",
+                )
+                break
+
+            # RECYCLE: если сессия отравлена WAF-403 — пересоздаём перед next obj.
+            if poisoned:
+                total_requests += sess.request_count
+                await sess.__aexit__(None, None, None)
+                sess = _new_session()
+                await sess.__aenter__()
+                await sess.warm_up()
+
+            if (i + 1) % 10 == 0:
+                _checkpoint(db, run_id, i + 1)
+    finally:
+        total_requests += sess.request_count
+        await sess.__aexit__(None, None, None)
+    if aborted:
+        logger.info("extras isolated pass aborted early — extras WAF-blocked wholesale")
+
+    log_progress(
+        db,
+        run_id,
+        f"extras-пас done: sale_graph={extras_counts['sale_graph_rows']}"
+        f" agg={extras_counts['sales_agg_rows']} infra={extras_counts['infra_rows']}"
+        f" photos={extras_counts['photos_rows']} docs={extras_counts['documents_rows']}",
+        stage="extras",
+    )
+    return total_requests
+
+
 # ── orchestrator ─────────────────────────────────────────────────────────────
 
 
@@ -1664,6 +1905,7 @@ async def run_region_sweep(
     request_jitter_min_ms: int | None = None,
     request_jitter_max_ms: int | None = None,
     proxy_url: str | None = None,
+    extras_isolated: bool = True,
 ) -> dict[str, Any]:
     """Full sweep for one region: bootstrap → fetch objects → per-object processing.
 
@@ -1672,6 +1914,12 @@ async def run_region_sweep(
     прогон (другие скраперы используют модульные дефолты stealth.py без изменений).
     None у каждого = дефолт stealth.py (concurrency=8, jitter 600–1500, без прокси).
 
+    ``extras_isolated`` (#1945 ГЛАВНЫЙ unblock): True (дефолт) → flats тянутся в
+    ЧИСТОЙ flats-only сессии (extras на ней НЕ дёргаются НИКОГДА → 403-яд extras не
+    травит cookies flats-сессии → flats_count восстанавливается), а extras идут
+    ОТДЕЛЬНЫМ best-effort проходом с recycle сессии на WAF-403. False → старое
+    поведение (flats+extras в одной сессии, poison-prone).
+
     Resume-aware: если `resume_from_run_id` задан, читаем objects_snapshot и
     progress_obj_index из исходного run и продолжаем с этого индекса. Новый
     run_id создаётся (с resumed_from_run_id ссылкой), исходный остаётся
@@ -1898,6 +2146,12 @@ async def run_region_sweep(
             # ограничивает запросы.
             # DB upserts выполняются последовательно после gather — один db Session
             # не thread-safe для параллельной записи.
+            #
+            # #1945 ИЗОЛЯЦИЯ: при extras_isolated=True эта (warm) сессия делает
+            # ТОЛЬКО flats — extras на ней НЕ дёргаются (их 403-яд не травит cookies
+            # flats-сессии → flats_count восстанавливается). extras уходят отдельным
+            # best-effort проходом ниже (после закрытия flats-сессии) с recycle на
+            # WAF-403. При extras_isolated=False — старое combined-поведение.
             pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT
             total = len(all_objects)
             for i in range(start_index, total):
@@ -1912,17 +2166,12 @@ async def run_region_sweep(
                 if fetch_flats:
                     coros.append(_fetch_flats_safe(sess, obj_id))
 
-                if extras:
-                    coros.append(_fetch_sale_graph_safe(sess, obj_id, "apartments"))
-                    coros.append(_fetch_sale_graph_safe(sess, obj_id, "parking"))
-                    coros.append(_fetch_sales_agg_safe(sess, obj_id))
-                    coros.append(_fetch_infrastructure_safe(sess, obj_id))
-                    coros.append(_fetch_photos_safe(sess, obj_id))
-                    coros.append(_fetch_doc_rpd_safe(sess, obj_id))
-                    coros.append(_fetch_doc_developer_report_safe(sess, obj_id))
-                    coros.append(_fetch_doc_project_documentation_safe(sess, obj_id))
-                    coros.append(_fetch_doc_documentation_other_safe(sess, obj_id))
-                    coros.append(_fetch_doc_permits_safe(sess, obj_id))
+                # extras в combined-сессии ТОЛЬКО когда изоляция выключена —
+                # иначе они отравят flats (#1945). В isolated-режиме extras
+                # обрабатываются отдельным проходом ниже.
+                combined_extras = extras and not extras_isolated
+                if combined_extras:
+                    coros.extend(_extras_coros(sess, obj_id))
                     # TODO: obj_checks endpoint not found at /api/object/{id}/checks (404).
                     # 6 чек-боксов "Проверено на наш.дом.рф" вероятно inline в kn/object payload.
                     # Re-enable после investigation структуры объекта (separate PR).
@@ -1936,64 +2185,28 @@ async def run_region_sweep(
 
                 # Sequential upsert — DB session не thread-safe
                 all_docs: list[dict[str, Any]] = []
-                _doc_kinds = frozenset(
-                    (
-                        "doc_rpd",
-                        "doc_developer_report",
-                        "doc_project_documentation",
-                        "doc_documentation_other",
-                        "doc_permits",
-                    )
-                )
                 for kind_tag, full_url, result in results:
-                    if isinstance(result, Exception):
-                        _classify_and_log(db, run_id, obj_id, kind_tag, full_url, result)
-                        continue
-
                     if kind_tag == "flats":
-                        flats_list: list[dict[str, Any]] = result  # type: ignore[assignment]
-                        if flats_list:
-                            total_flats += upsert_flats(db, flats_list, snapshot_date, region_code)
-
-                    elif kind_tag in ("sale_graph_apartments", "sale_graph_parking"):
-                        sg_type = kind_tag.replace("sale_graph_", "")
-                        rows_sg, _ = result  # type: ignore[misc]
-                        extras_counts["sale_graph_rows"] += upsert_sale_graph(
-                            db, obj_id, sg_type, rows_sg, snapshot_date
+                        total_flats += _process_flats_result(
+                            db, run_id, obj_id, full_url, result, snapshot_date, region_code
+                        )
+                    else:
+                        await _process_extras_result(
+                            sess,
+                            db,
+                            run_id,
+                            obj_id,
+                            kind_tag,
+                            full_url,
+                            result,
+                            snapshot_date,
+                            extras_counts,
+                            all_docs,
+                            download_photos_binary=download_photos_binary,
+                            pdir=pdir,
                         )
 
-                    elif kind_tag == "sales_agg":
-                        agg_data, _ = result  # type: ignore[misc]
-                        extras_counts["sales_agg_rows"] += upsert_sales_agg(
-                            db, obj_id, agg_data, snapshot_date
-                        )
-
-                    elif kind_tag == "infrastructure":
-                        pois_data, _ = result  # type: ignore[misc]
-                        extras_counts["infra_rows"] += upsert_infrastructure(
-                            db, obj_id, pois_data, snapshot_date
-                        )
-
-                    elif kind_tag == "photos":
-                        photos_data, _ = result  # type: ignore[misc]
-                        local_paths: dict[str, str] = {}
-                        thumb_paths: dict[str, str] = {}
-                        if download_photos_binary and photos_data:
-                            local_paths, thumb_paths = await download_photos(
-                                sess, obj_id, photos_data, pdir
-                            )
-                            extras_counts["photos_downloaded"] += len(local_paths)
-                        extras_counts["photos_rows"] += upsert_photos(
-                            db, obj_id, photos_data, local_paths, thumb_paths
-                        )
-
-                    elif kind_tag in _doc_kinds:
-                        # Каждый из 5 doc-endpoint'ов отдаёт свой список документов.
-                        # Накапливаем в all_docs — единый upsert после цикла.
-                        doc_items, _ = result  # type: ignore[misc]
-                        all_docs.extend(extract_documents(doc_items or []))
-
-                # Единый upsert всех документов объекта после обработки 5 endpoint'ов.
+                # Единый upsert всех документов объекта после обработки doc-endpoint'ов.
                 if all_docs:
                     ins, _skip = upsert_documents(db, obj_id, all_docs)
                     extras_counts["documents_rows"] += ins
@@ -2012,12 +2225,49 @@ async def run_region_sweep(
                         f" downloaded={extras_counts['photos_downloaded']}"
                         f" docs={extras_counts['documents_rows']}"
                         f" checks={extras_counts['checks_rows']}",
-                        stage="extras" if extras else "fetch_flats",
+                        stage="extras" if combined_extras else "fetch_flats",
                     )
 
             _checkpoint(db, run_id, total)
             request_count = sess.request_count
 
+        # ── Phase C (isolated) — extras best-effort пас в ОТДЕЛЬНОЙ сессии ──────
+        # flats-сессия закрыта. Теперь (только если extras=True И extras_isolated)
+        # тянем extras в disposable-сессиях: на каждый WAF-403 recycle всю
+        # BrowserSession (свежие cookies/контекст) перед следующим объектом, чтобы
+        # яд НЕ накапливался. extras сейчас мертвы (403 с 2026-06-03) — пас целиком
+        # best-effort: любые сбои уходят в kn_scrape_failures и НЕ роняют run.
+        if extras and extras_isolated:
+            try:
+                ex_req = await _run_extras_pass(
+                    db=db,
+                    run_id=run_id,
+                    region_code=region_code,
+                    all_objects=all_objects,
+                    start_index=start_index,
+                    snapshot_date=snapshot_date,
+                    extras_counts=extras_counts,
+                    pdir=pdir,
+                    download_photos_binary=download_photos_binary,
+                    load_state=load_state,
+                    headed=headed,
+                    browser_concurrency=browser_concurrency,
+                    request_jitter_min_ms=request_jitter_min_ms,
+                    request_jitter_max_ms=request_jitter_max_ms,
+                    proxy_url=proxy_url,
+                )
+                request_count += ex_req
+            except Exception as e:
+                # best-effort: extras-пас НИКОГДА не должен ронять уже собранные flats.
+                logger.warning("extras isolated pass failed (best-effort): %s", e)
+                log_progress(
+                    db,
+                    run_id,
+                    f"extras-пас (best-effort) упал: {type(e).__name__}: {str(e)[:200]}",
+                    level="warn",
+                    stage="extras",
+                )
+
         # ── Phase D — finalize ─────────────────────────────────────────────
         db.execute(
             text(

backend/app/workers/tasks/scrape_kn.py

@@ -210,11 +210,12 @@ def scrape_kn_region(
                 fetch_flats=fetch_flats,
                 extras=extras,
                 download_photos_binary=download_photos,
-                # #1945 anti-ban: throttle + optional proxy для KN-sweep.
+                # #1945 anti-ban: throttle + optional proxy + flats/extras isolation.
                 browser_concurrency=settings.scrape_kn_browser_concurrency,
                 request_jitter_min_ms=settings.scrape_kn_request_jitter_min_ms,
                 request_jitter_max_ms=settings.scrape_kn_request_jitter_max_ms,
                 proxy_url=settings.scrape_kn_proxy_url,
+                extras_isolated=settings.scrape_kn_extras_isolated,
             )
         )
 
@@ -294,10 +295,11 @@ def resume_kn_run(self: Any, run_id: int) -> dict[str, Any]:
                 developers=developers,
                 load_state=state_path,
                 resume_from_run_id=run_id,
-                # #1945 anti-ban: throttle + optional proxy для KN-sweep.
+                # #1945 anti-ban: throttle + optional proxy + flats/extras isolation.
                 browser_concurrency=settings.scrape_kn_browser_concurrency,
                 request_jitter_min_ms=settings.scrape_kn_request_jitter_min_ms,
                 request_jitter_max_ms=settings.scrape_kn_request_jitter_max_ms,
                 proxy_url=settings.scrape_kn_proxy_url,
+                extras_isolated=settings.scrape_kn_extras_isolated,
             )
         )

backend/tests/services/scrapers/test_domrf_kn_isolation.py

@@ -0,0 +1,200 @@
+"""#1945 flats/extras isolation tests (scrapers/domrf_kn).
+
+PROVEN mechanism: extras /сервисы/api/object/{id}/* return 403 (volume-independent,
+dead since 2026-06-03) AND poison the session cookies → subsequent FLATS on the same
+session flip to 403. flats_count collapsed 3670→9. Fix: flats never share a session
+with extras; extras run as a separate best-effort pass that RECYCLES the session on
+any WAF-403 so poison never accumulates (and never touches flats).
+
+These tests assert:
+  • _extras_coros builds ONLY extras endpoints (never flats);
+  • _is_waf_poisoned flips on any WafBlockedError;
+  • _run_extras_pass recycles the BrowserSession on a poisoned object;
+  • _run_extras_pass aborts early when extras are wholesale-403 (no ~1500 relaunches).
+"""
+
+from __future__ import annotations
+
+import datetime
+from typing import Any, ClassVar
+
+import pytest
+
+import app.services.scrapers.domrf_kn as kn
+from app.services.scrapers.stealth import WafBlockedError
+
+
+class TestExtrasCorosPurity:
+    """_extras_coros НИКОГДА не включает flats — иначе изоляция бессмысленна."""
+
+    def test_extras_coros_excludes_flats(self) -> None:
+        coros = kn._extras_coros(sess=object(), obj_id=123)  # type: ignore[arg-type]
+        try:
+            names = {c.__qualname__ for c in coros}
+            assert not any("flats" in n for n in names)
+            # 10 extras: 2 sale_graph + sales_agg + infra + photos + 5 docs.
+            assert len(coros) == 10
+        finally:
+            for c in coros:
+                c.close()  # avoid un-awaited coroutine warnings
+
+
+class TestWafPoisonDetection:
+    def test_poisoned_on_any_waf(self) -> None:
+        results = [
+            ("sales_agg", "u", {"ok": 1}),
+            ("infrastructure", "u", WafBlockedError("403 waf")),
+        ]
+        assert kn._is_waf_poisoned(results) is True
+
+    def test_not_poisoned_without_waf(self) -> None:
+        results = [
+            ("sales_agg", "u", {"ok": 1}),
+            ("infrastructure", "u", RuntimeError("http 500")),  # transient, not WAF
+        ]
+        assert kn._is_waf_poisoned(results) is False
+
+    def test_empty_not_poisoned(self) -> None:
+        assert kn._is_waf_poisoned([]) is False
+
+
+# ── _run_extras_pass session-lifecycle tests ─────────────────────────────────
+
+
+class _FakeSession:
+    """BrowserSession stand-in: tracks bootstrap/warm/exit, no real browser."""
+
+    instances: ClassVar[list[_FakeSession]] = []
+
+    def __init__(self, **_kw: Any) -> None:
+        self.request_count = 7
+        self.warmed = 0
+        self.exited = False
+        _FakeSession.instances.append(self)
+
+    async def __aenter__(self) -> _FakeSession:
+        return self
+
+    async def __aexit__(self, *_exc: Any) -> None:
+        self.exited = True
+
+    async def warm_up(self, force: bool = False) -> None:
+        self.warmed += 1
+
+
+def _objs(n: int) -> list[dict[str, Any]]:
+    return [{"objId": 1000 + i} for i in range(n)]
+
+
+@pytest.fixture(autouse=True)
+def _patch(monkeypatch: pytest.MonkeyPatch) -> None:
+    _FakeSession.instances = []
+    monkeypatch.setattr(kn, "BrowserSession", _FakeSession)
+    monkeypatch.setattr(kn, "log_progress", lambda *a, **k: None)
+    monkeypatch.setattr(kn, "_checkpoint", lambda *a, **k: None)
+    monkeypatch.setattr(kn, "upsert_documents", lambda *a, **k: (0, 0))
+
+    # _process_extras_result is real upsert plumbing — no-op it (we test lifecycle).
+    async def _noop_process(*_a: Any, **_k: Any) -> None:
+        return None
+
+    monkeypatch.setattr(kn, "_process_extras_result", _noop_process)
+
+
+def _make_coros_factory(per_obj_results: Any) -> Any:
+    """Return a _extras_coros replacement: per-object it yields coroutines that each
+    resolve to one staged (kind, url, result) tuple, so real asyncio.gather is used."""
+    state = {"i": 0}
+
+    def _coros(_sess: Any, _oid: int) -> list[Any]:
+        results = per_obj_results(state["i"])
+        state["i"] += 1
+
+        async def _wrap(res: tuple[str, str, Any]) -> tuple[str, str, Any]:
+            return res
+
+        return [_wrap(r) for r in results]
+
+    return _coros
+
+
+async def _drive(objs: list[dict[str, Any]]) -> int:
+    extras_counts = dict.fromkeys(
+        (
+            "sale_graph_rows",
+            "sales_agg_rows",
+            "infra_rows",
+            "photos_rows",
+            "photos_downloaded",
+            "documents_rows",
+            "checks_rows",
+        ),
+        0,
+    )
+    return await kn._run_extras_pass(
+        db=object(),  # type: ignore[arg-type]
+        run_id=1,
+        region_code=66,
+        all_objects=objs,
+        start_index=0,
+        snapshot_date=datetime.date(2026, 6, 28),
+        extras_counts=extras_counts,
+        pdir=kn.PHOTOS_DIR_DEFAULT,
+        download_photos_binary=False,
+        load_state=None,
+        headed=False,
+        browser_concurrency=2,
+        request_jitter_min_ms=1200,
+        request_jitter_max_ms=3000,
+        proxy_url=None,
+    )
+
+
+class TestRunExtrasPassLifecycle:
+    @pytest.mark.asyncio
+    async def test_clean_results_use_single_session(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Every object returns clean JSON → no poison → one session for the whole pass.
+        def clean(_i: int) -> list[tuple[str, str, Any]]:
+            return [("sales_agg", "u", ({"x": 1}, "u")), ("infrastructure", "u", ([], "u"))]
+
+        monkeypatch.setattr(kn, "_extras_coros", _make_coros_factory(clean))
+        req = await _drive(_objs(4))
+        # Exactly 1 BrowserSession constructed for a fully-clean pass.
+        assert len(_FakeSession.instances) == 1
+        assert _FakeSession.instances[0].exited is True
+        assert req == 7  # single session.request_count
+
+    @pytest.mark.asyncio
+    async def test_recycle_on_poison(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Object 0 poisoned (1 WAF among results) but NOT all-WAF (so no early abort);
+        # objects 1+ clean → pass recycles once after object 0, then continues clean.
+        def mixed(i: int) -> list[tuple[str, str, Any]]:
+            if i == 0:
+                return [
+                    ("sales_agg", "u", ({"x": 1}, "u")),  # one ok → not all-WAF
+                    ("infrastructure", "u", WafBlockedError("403")),  # poison
+                ]
+            return [("sales_agg", "u", ({"x": 1}, "u")), ("infrastructure", "u", ([], "u"))]
+
+        monkeypatch.setattr(kn, "_extras_coros", _make_coros_factory(mixed))
+        await _drive(_objs(3))
+        # 1 initial + 1 recycle after the poisoned object 0 = 2 sessions.
+        assert len(_FakeSession.instances) == 2
+        assert all(s.exited for s in _FakeSession.instances)
+
+    @pytest.mark.asyncio
+    async def test_early_abort_on_wholesale_waf(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Every object fully WAF-403 (extras dead) → abort after the threshold,
+        # NOT grinding through all 1000 objects with a relaunch each time.
+        def all_waf(_i: int) -> list[tuple[str, str, Any]]:
+            return [
+                ("sales_agg", "u", WafBlockedError("403")),
+                ("infrastructure", "u", WafBlockedError("403")),
+            ]
+
+        monkeypatch.setattr(kn, "_extras_coros", _make_coros_factory(all_waf))
+        await _drive(_objs(1000))
+        # Abort kicks in at _EXTRAS_ABORT_AFTER_CONSEC_WAF consecutive all-WAF objects.
+        # Sessions: each poisoned object recycles UNLESS it's the abort object.
+        # Bounded well under 1000 — proves we don't relaunch per object forever.
+        assert len(_FakeSession.instances) <= kn._EXTRAS_ABORT_AFTER_CONSEC_WAF + 1