fix(scraper-kit/yandex): обрезанный потолком страниц leaf не идёт в чекпоинт

После #3362 degraded-ветка отмечается complete=False, а `_leaf` писал бакет как
полный, даже когда его пагинация упиралась в max_pages_per_bucket/_GATE_MAX_PAGES_CAP.
С containment-гейтом (#3358/#3359) такой ключ покрывает свой интервал целиком, и
резюм больше не заходит в полосу, чей хвост не читали ни разу.

Флаг полноты — как у cian: complete = pages_needed <= max_pages, передаётся в
on_bucket на обоих выходах leaf'а (_mark_bucket кладёт в done только complete).
Переполненный leaf возможен только там, где бисекции дробить нечем (размах <
min_bracket, открытый верхний брекет, потолок глубины) — это честный исход
«бакет неполон по построению», поэтому он ещё и считается отдельно
(scraper.capped_buckets → counters.capped_buckets): лечится не повтором прогона,
а порогами бисекции.

Closes #3368

tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py

@@ -0,0 +1,135 @@
+"""Leaf-бакет яндекса, обрезанный потолком страниц, не идёт в чекпоинт (#3368).
+
+Довесок к #3362: там честной сделали degraded-ветку (`on_bucket(..., complete=False)`),
+а `_leaf` продолжал писать бакет как полный, даже когда его пагинация упиралась в
+`max_pages_per_bucket`/`_GATE_MAX_PAGES_CAP`. С containment-гейтом (#3358/#3359) такой
+ключ покрывает СВОЙ интервал целиком → резюм больше не заходит в полосу, чей хвост не
+читали ни разу. У cian тот же случай считается честно (`complete = pages_needed <=
+max_pages`, cian/serp.py).
+
+Переполненный leaf возможен только там, где бисекции дробить нечем: размах меньше
+`_YANDEX_SPLIT_MIN_BRACKET`, открытый верхний брекет или потолок глубины. Здесь берётся
+первый случай: плотная выдача (totalItems=2000 > cap=500) делится до размаха 499 999 и
+дальше делиться не может.
+
+Сеть не нужна: `_fetch_page_json` подменяется счётчиком. Проверки ПО ЗНАЧЕНИЮ — что
+попало в done-леджер и сколько запросов сделал резюм.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import Any
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from scraper_kit.base import ScrapedLot
+from scraper_kit.providers.yandex.serp import YandexRealtyScraper
+
+from app.services.scraper_adapters import RealScraperConfig
+
+_ROOMS = "2"
+
+
+def _page(total_items: int) -> dict[str, Any]:
+    return {
+        "response": {
+            "search": {
+                "offers": {
+                    "entities": [],
+                    "pager": {
+                        "totalItems": total_items,
+                        "totalPages": max(1, total_items // 20),
+                        "page": 0,
+                    },
+                }
+            }
+        }
+    }
+
+
+def _walk(
+    total_items: int,
+    max_pages_per_bucket: int,
+    skip_buckets: set[str] | None = None,
+) -> tuple[list[tuple[str, bool]], int, int]:
+    """Прогнать бисекцию [4M, 5M) и вернуть (отметки бакетов, запросы, capped_buckets).
+
+    Колбэк — той же формы, что pipeline._on_bucket: третий позиционный аргумент =
+    признак полноты; в done-леджер `_mark_bucket` кладёт только complete=True.
+    """
+    s = YandexRealtyScraper(RealScraperConfig())
+    s.request_delay_sec = 0.0
+    calls = [0]
+
+    async def fake_fetch(
+        rooms: str | None,
+        page: int,
+        price_min: int | None,
+        price_max: int | None,
+        new_flat: str = "NO",
+    ) -> dict[str, Any]:
+        calls[0] += 1
+        return _page(total_items)
+
+    s._fetch_page_json = fake_fetch  # type: ignore[method-assign]
+    marked: list[tuple[str, bool]] = []
+
+    def on_bucket(key: str, _count: int, complete: bool = True) -> None:
+        marked.append((key, complete))
+
+    seen: dict[str, ScrapedLot] = {}
+    asyncio.run(
+        s._walk_price_range(
+            rooms=_ROOMS,
+            lo=4_000_000,
+            hi=4_999_999,
+            seen=seen,
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=max_pages_per_bucket,
+            on_bucket=on_bucket,
+            skip_buckets=skip_buckets,
+        )
+    )
+    # getattr, а не атрибут напрямую: без счётчика тест обязан краснеть НЕВЕРНЫМ
+    # ЗНАЧЕНИЕМ (ключ в леджере / ноль запросов на резюме), а не AttributeError'ом —
+    # «возможности нет» неотличимо от «проверка не проведена».
+    return marked, calls[0], getattr(s, "capped_buckets", 0)
+
+
+def test_capped_leaf_stays_out_of_done_ledger() -> None:
+    """Приёмка: leaf, которому нужно больше страниц, чем потолок, — не в леджере."""
+    marked, _, capped = _walk(total_items=2000, max_pages_per_bucket=1)
+    assert marked, "leaf не вызвал on_bucket — тест ничего не проверяет"
+    ledger = {key for key, complete in marked if complete}
+    assert not ledger, (
+        f"бакеты {sorted(ledger)} прочитаны на 1 страницу из 100 (totalItems=2000), "
+        "но помечены complete — их интервал зачтётся containment-гейтом целиком"
+    )
+    assert capped == len(marked), (
+        f"capped_buckets={capped} при {len(marked)} обрезанных бакетах — "
+        "счётчик прогона не покажет, что полоса недобрана по построению"
+    )
+
+
+def test_fully_read_leaf_goes_into_done_ledger() -> None:
+    """Контроль честности: дочитанный до конца бакет по-прежнему чекпоинтится."""
+    marked, _, capped = _walk(total_items=15, max_pages_per_bucket=1)
+    ledger = {key for key, complete in marked if complete}
+    assert ledger, (
+        "бакет из одной страницы (totalItems=15) не попал в леджер — резюм будет "
+        "перечитывать уже собранную территорию"
+    )
+    assert capped == 0, f"полный бакет посчитан обрезанным (capped_buckets={capped})"
+
+
+def test_capped_band_is_rewalked_on_resume() -> None:
+    """По значению: полоса с обрезанным leaf'ом на резюме снова обходится."""
+    marked, _, _ = _walk(total_items=2000, max_pages_per_bucket=1)
+    ledger = {key for key, complete in marked if complete}
+    _, calls, _ = _walk(total_items=2000, max_pages_per_bucket=1, skip_buckets=ledger or None)
+    assert calls > 0, (
+        "резюм не сделал ни одного запроса по полосе, у которой прочитана 1 страница "
+        "из 100 — недобранный хвост потерян навсегда"
+    )

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

@@ -3807,6 +3807,11 @@ class YandexFullLoadCounters:
     # Бакеты, собранные ЧАСТИЧНО (probe провалился → degraded-пагинация): в
     # чекпоинт не пишутся, следующий прогон перечитает их целиком.
     partial_buckets: int = 0
+    # Подмножество partial_buckets: бакет неполон ПО ПОСТРОЕНИЮ (#3368) — total
+    # требует больше страниц, чем потолок, а бисекции делить его уже нечем.
+    # Отдельный счётчик, потому что лечение другое: не «повторить прогон», а
+    # снизить min_bracket / поднять потолок страниц.
+    capped_buckets: int = 0
 
     def to_dict(self) -> dict[str, int]:
         return {f.name: getattr(self, f.name) for f in fields(self)}
@@ -3950,6 +3955,8 @@ async def run_yandex_full_load(
                 on_progress=_on_progress,
                 skip_buckets=skip_set if skip_set else None,
             )
+            # #3368: leaf'ы, обрезанные потолком страниц (в чекпоинт не попали).
+            counters.capped_buckets = getattr(scraper, "capped_buckets", 0)
 
         logger.info(
             "yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py

@@ -531,6 +531,11 @@ class YandexRealtyScraper(BaseScraper):
         # блокировки — см. _track_gate_result.
         self.gate_fetch_attempts: int = 0
         self.gate_fetch_failures: int = 0
+        # #3368: leaf-бакеты, чья пагинация упёрлась в потолок страниц (дробить
+        # бисекции уже нечем). Не пишутся в чекпоинт → читаются вызывающим для
+        # counters прогона. Атрибут, а не аргумент on_bucket — тот же довод, что у
+        # cian.last_dropped_nb: у колбэка есть внешние реализации.
+        self.capped_buckets: int = 0
 
     def _track_gate_result(self, ok: bool) -> None:
         """Учёт исхода одного top-level gate-API запроса (#2625).
@@ -1220,13 +1225,35 @@ class YandexRealtyScraper(BaseScraper):
                 logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key)
                 return
 
-            total_pages = min(
-                math.ceil(total / 20),
-                _GATE_MAX_PAGES_CAP,
-                max_pages_per_bucket,
-            )
+            pages_needed = math.ceil(total / 20)
+            max_pages = min(_GATE_MAX_PAGES_CAP, max_pages_per_bucket)
+            total_pages = min(pages_needed, max_pages)
+            # Полнота бакета (#3368): у обрезанного потолком страниц ключ ТОТ ЖЕ, что у
+            # собранного целиком, — как у cian (`complete = pages_needed <= max_pages`).
+            # complete=False → бакет не идёт в done-леджер, иначе его интервал в
+            # containment-гейте (#3359) склеился бы с соседними и резюм пропустил бы
+            # полосу, чей хвост никогда не читали. Бакет доходит сюда переполненным
+            # только там, где бисекции делить больше нечем (размах < min_bracket,
+            # открытый верхний брекет, потолок глубины) — см. walk_price_range.
+            complete = pages_needed <= max_pages
+            if not complete:
+                self.capped_buckets += 1
+                logger.warning(
+                    "yandex gate: leaf bucket %s НЕПОЛОН по построению — total=%d требует "
+                    "%d страниц при потолке %d; дробить дальше нечем, в чекпоинт не пишем "
+                    "(capped_buckets=%d)",
+                    bucket_key,
+                    total,
+                    pages_needed,
+                    max_pages,
+                    self.capped_buckets,
+                )
             logger.info(
-                "yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages
+                "yandex gate: leaf bucket %s total=%d pages=%d complete=%s",
+                bucket_key,
+                total,
+                total_pages,
+                complete,
             )
 
             # Add probe lots (page 1 already fetched)
@@ -1236,7 +1263,7 @@ class YandexRealtyScraper(BaseScraper):
 
             if total_pages <= 1:
                 if on_bucket is not None:
-                    on_bucket(bucket_key, len(seen))
+                    on_bucket(bucket_key, len(seen), complete)
                 return
 
             # Paginate pages 2..total_pages with concurrency
@@ -1258,7 +1285,7 @@ class YandexRealtyScraper(BaseScraper):
                         seen[lot.source_id] = lot
 
             if on_bucket is not None:
-                on_bucket(bucket_key, len(seen))
+                on_bucket(bucket_key, len(seen), complete)
 
         # #3359: гейт ПЕРЕД probe (как у avito, #3315). Ключи yandex'а — `_combo_label`,
         # т.е. «rooms:lo-hi» с «None» вместо открытого потолка: другой разделитель и