fix(scraper-kit): yandex on_bucket отдаёт лоты бакета, а не len(seen)

`_leaf`/`_degraded` звали on_bucket(bucket_key, len(seen), complete) — int
уезжал в run_yandex_full_load._on_bucket и дальше в save_listings
(`for lot in lots`) → TypeError, ручной full-load Яндекса не сохранял ничего.
Контракт выровнен по cian/avito: провайдер копит лоты бакета (новые в seen)
и отдаёт список. Parity-фикстура подменяла скрапер целиком и слала list по
построению — добавлен прогон run_yandex_full_load через НАСТОЯЩИЙ
YandexRealtyScraper (замокан только gate-JSON транспорт).

Closes #3375

tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py

@@ -142,7 +142,7 @@ def _walk_degraded(skip_buckets: set[str] | None) -> tuple[list[tuple[str, bool]
     s._rotate_ip = no_rotate  # type: ignore[method-assign]
     marked: list[tuple[str, bool]] = []
 
-    def on_bucket(key: str, _count: int, complete: bool = True) -> None:
+    def on_bucket(key: str, _lots: list[Any], complete: bool = True) -> None:
         marked.append((key, complete))
 
     asyncio.run(

tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py

@@ -84,7 +84,7 @@ def _walk(
     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:
+    def on_bucket(key: str, _lots: list[ScrapedLot], complete: bool = True) -> None:
         marked.append((key, complete))
 
     seen: dict[str, ScrapedLot] = {}

tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py

@@ -30,6 +30,7 @@ import pytest
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
 
+from scraper_kit.base import ScrapedLot
 from scraper_kit.orchestration.pipeline import (
     run_avito_full_load,
     run_avito_newbuilding_sweep,
@@ -40,6 +41,7 @@ from scraper_kit.orchestration.pipeline import (
     run_yandex_full_load,
 )
 from scraper_kit.providers.domclick.serp import ROOM_BUCKETS
+from scraper_kit.providers.yandex.serp import YandexRealtyScraper
 
 PFX = "scraper_kit.orchestration.pipeline"
 
@@ -619,7 +621,12 @@ async def _drive_full_load(*, source: str, capture: dict[str, Any] | None = None
 @pytest.mark.asyncio
 @pytest.mark.parametrize("source", ["avito", "cian", "yandex"])
 async def test_full_load_smoke(source: str) -> None:
-    """Full load: 2 бакета через on_bucket → save + heartbeat → mark_done."""
+    """Full load: 2 бакета через on_bucket → save + heartbeat → mark_done.
+
+    ВНИМАНИЕ: `_full_load_scraper` шлёт в on_bucket список ПО ПОСТРОЕНИЮ — контракт
+    «что провайдер реально кладёт в колбэк» здесь не проверяется вовсе (#3375).
+    Настоящий контракт — в тесте ниже.
+    """
     counters, calls = await _drive_full_load(source=source)
     assert counters["unique_fetched"] == 3
     assert counters["saved_inserted"] == 3
@@ -627,6 +634,89 @@ async def test_full_load_smoke(source: str) -> None:
     assert calls[-1][0] == "mark_done"
 
 
+# ── #3375: yandex full-load — контракт on_bucket через НАСТОЯЩИЙ провайдер ────
+#
+# Фикстура выше подменяет скрапер целиком, поэтому в save_listings всегда приезжал
+# список — а yandex/serp.py звал `on_bucket(bucket_key, len(seen), complete)` (int),
+# и `_on_bucket` отдавал это в `save_listings` (`for lot in lots`) → TypeError:
+# ручной full-load Яндекса не сохранял ничего. Ниже гоняется НАСТОЯЩИЙ
+# YandexRealtyScraper (fetch_all_secondary → _walk_price_range → _leaf), подменён
+# только gate-JSON транспорт. Фальсификация: вернуть в serp.py `len(seen)` — тест
+# краснеет TypeError'ом на save_listings, как прод.
+
+
+class _StubbedYandexScraper(YandexRealtyScraper):
+    """Настоящий скрапер яндекса, у которого замокан ТОЛЬКО gate-JSON транспорт."""
+
+    async def __aenter__(self) -> _StubbedYandexScraper:
+        return self  # без camoufox/BrowserFetcher
+
+    async def __aexit__(self, *_exc: Any) -> bool:
+        return False
+
+    async def fetch_all_secondary(self, **kwargs: Any) -> list[Any]:
+        """Одна комнатность вместо пяти — путь до `_leaf` тот же, прогон короче."""
+        return await super().fetch_all_secondary(rooms_buckets=["2"], **kwargs)
+
+    async def _fetch_page_json(
+        self,
+        rooms: str | None,
+        page: int,
+        price_min: int | None = None,
+        price_max: int | None = None,
+        new_flat: str = "NO",
+    ) -> dict[str, Any]:
+        """totalItems=1 → бисекция не делит, leaf сразу. Оффер уникален на бакет."""
+        offer_id = f"y{price_min or 0}_{price_max or 0}"
+        return {
+            "response": {
+                "search": {
+                    "offers": {
+                        "entities": [{"offerId": offer_id, "price": {"value": 5_000_000}}],
+                        "pager": {"totalItems": 1, "totalPages": 1, "page": 0},
+                    }
+                }
+            }
+        }
+
+
+@pytest.mark.asyncio
+async def test_yandex_full_load_saves_lot_list_from_real_provider() -> None:
+    """#3375: save_listings получает СПИСОК лотов, ровно один вызов на бакет."""
+    recorder = _RunsRecorder()
+    save_mock = MagicMock(return_value=(1, 0))
+    enrichment = MagicMock()
+    enrichment.record_yandex_price_history = MagicMock(return_value=0)
+    with (
+        patch(f"{PFX}.YandexRealtyScraper", _StubbedYandexScraper),
+        patch(f"{PFX}.save_listings", save_mock),
+        patch(f"{PFX}.runs", recorder),
+    ):
+        counters = await run_yandex_full_load(
+            MagicMock(),
+            run_id=1,
+            config=_config(),
+            matcher=MagicMock(),
+            enrichment=enrichment,
+            request_delay_sec=0.0,
+        )
+
+    assert save_mock.call_count > 0, "save_listings не вызван — прогон ничего не сохранил"
+    for call in save_mock.call_args_list:
+        lots = call.args[1]
+        assert isinstance(lots, list), (
+            f"в save_listings уехал {type(lots).__name__} вместо списка лотов — "
+            "провайдер и pipeline разошлись контрактом (#3375)"
+        )
+        assert lots and all(isinstance(lot, ScrapedLot) for lot in lots), (
+            f"бакет отдал {lots!r} — save_listings пишет не лоты"
+        )
+    # По одному новому лоту на бакет (offer_id уникален на ценовой диапазон).
+    assert counters.unique_fetched == save_mock.call_count > 0
+    assert counters.saved_inserted == save_mock.call_count
+    assert _normalize(recorder.calls)[-1][0] == "mark_done"
+
+
 # ── #2616: run_avito_full_load прокидывает proxy_provider в AvitoScraper ─────
 #
 # run_avito_full_load — единственное из мест создания AvitoScraper в pipeline.py, где

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

@@ -1139,7 +1139,8 @@ class YandexRealtyScraper(BaseScraper):
 
         Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Yandex-spec
         (probe gate-JSON page=1, degraded paginate-until-empty, leaf с first-wins
-        дедупом и count-based on_bucket) — в callback'ах; пороги (cap, 500k min,
+        дедупом; on_bucket отдаёт ЛОТЫ бакета — новые в `seen`, — как cian/avito:
+        pipeline._on_bucket кладёт их в save_listings, #3375) — в callback'ах; пороги (cap, 500k min,
         depth<8) и политика DEGRADE — в BisectionConfig.
         """
 
@@ -1194,6 +1195,7 @@ class YandexRealtyScraper(BaseScraper):
                 return
             page = 1
             pages_fetched = 0
+            bucket_lots: list[ScrapedLot] = []
             while pages_fetched < max_pages_per_bucket:
                 payload = await self._fetch_page_json(rooms, page, lo_param, phi)
                 await asyncio.sleep(self.request_delay_sec)
@@ -1205,6 +1207,7 @@ class YandexRealtyScraper(BaseScraper):
                 for lot in lots:
                     if lot.source_id and lot.source_id not in seen:
                         seen[lot.source_id] = lot
+                        bucket_lots.append(lot)
                 page += 1
                 pages_fetched += 1
             if on_bucket is not None:
@@ -1213,13 +1216,14 @@ class YandexRealtyScraper(BaseScraper):
                 # бакет НЕ пишется в done-леджер (как у cian), иначе его интервал
                 # слился бы с соседними в containment-гейте (#3359) и резюм уже не
                 # переобошёл бы недобранную полосу.
-                on_bucket(bucket_key, len(seen), False)
+                on_bucket(bucket_key, bucket_lots, False)
 
         async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None:
             total = result.count
             assert total is not None  # DEGRADE-политика уводит None в _degraded
             lo_param = plo if plo and plo > 0 else None
             probe_lots: list[ScrapedLot] = result.payload or []
+            bucket_lots: list[ScrapedLot] = []
             bucket_key = _combo_label(rooms, plo, phi)
             if skip_buckets and bucket_key in skip_buckets:
                 logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key)
@@ -1262,10 +1266,11 @@ class YandexRealtyScraper(BaseScraper):
             for lot in probe_lots:
                 if lot.source_id and lot.source_id not in seen:
                     seen[lot.source_id] = lot
+                    bucket_lots.append(lot)
 
             if total_pages <= 1:
                 if on_bucket is not None:
-                    on_bucket(bucket_key, len(seen), not capped)
+                    on_bucket(bucket_key, bucket_lots, not capped)
                 return
 
             # Paginate pages 2..total_pages with concurrency
@@ -1288,6 +1293,7 @@ class YandexRealtyScraper(BaseScraper):
                 for lot in page_lots:
                     if lot.source_id and lot.source_id not in seen:
                         seen[lot.source_id] = lot
+                        bucket_lots.append(lot)
 
             complete = not capped and dropped_pages == 0
             if dropped_pages:
@@ -1299,13 +1305,15 @@ class YandexRealtyScraper(BaseScraper):
                     total_pages - 1,
                 )
             if on_bucket is not None:
-                on_bucket(bucket_key, len(seen), complete)
+                on_bucket(bucket_key, bucket_lots, complete)
 
         # #3359: гейт ПЕРЕД probe (как у avito, #3315). Ключи yandex'а — `_combo_label`,
         # т.е. «rooms:lo-hi» с «None» вместо открытого потолка: другой разделитель и
         # другой open-токен, чем у avito/cian, поэтому парсер параметризуется, а ключи
         # остаются как есть (живые чекпоинты не ломаем).
-        _covered = done_range_skipper(skip_buckets, rooms or "any", range_sep="-", open_token="None")
+        _covered = done_range_skipper(
+            skip_buckets, rooms or "any", range_sep="-", open_token="None"
+        )
 
         def _skip_done(plo: int | None, phi: int | None) -> bool:
             if _covered is None or not _covered(plo, phi):