Merge pull request 'fix(tradein/avito): брейкер добора считает долю только по отказам площадки — сайдкар/пустой пул не рвут прогон и не красят его banned' (#3367) from fix/3288-breaker-block-kind into main

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

@@ -56,6 +56,8 @@ from __future__ import annotations
 from collections import Counter, deque
 from dataclasses import dataclass, field
 
+from scraper_kit.orchestration.runs import BAN_KIND_PLATFORM
+
 
 @dataclass
 class BlockRatioBreaker:
@@ -88,7 +90,26 @@ class BlockRatioBreaker:
         """Знаменатель ratio-критерия: сколько попыток уже влезло в окно."""
         return len(self._window)
 
-    def record_block(self) -> None:
+    def record_block(self, kind: str = BAN_KIND_PLATFORM) -> None:
+        """Блок в числитель ratio-критерия — ТОЛЬКО отказ площадки (#3288).
+
+        `kind` — диагноз того же исключения, что уходит в counters.ban_kinds
+        (ban_kind_of_exception). Всё, что не 'platform' (в первую очередь
+        BAN_KIND_INFRA: отказ нашего сайдкара, пустой пул), — отказ НАШЕЙ стороны:
+        площадка его не показывала, и рвать по нему прогон значит объявить бан там,
+        где площадка молчала. Прогон 5425 оборвался по доле на 41 infra из 48
+        «блоков» при 41 успешно обогащённой карточке.
+
+        Такой отказ идёт ровно туда же, куда record_failure(): в ЗНАМЕНАТЕЛЬ окна
+        (доля платформенных блоков от него честно падает), мимо серии и safety-net.
+        Отдельно важно, что серию он не двигает: иначе safety-net на снапшоте
+        короче окна рвал бы прогон по тем же infra-отказам, только другим
+        критерием. Дефолт 'platform' сохраняет поведение вызывающих, которые вид
+        не считают.
+        """
+        if kind != BAN_KIND_PLATFORM:
+            self.record_failure()
+            return
         self._consecutive_blocks += 1
         self._window.append(True)
 

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

@@ -911,23 +911,48 @@ def mark_backfill_finished(
         return
 
     if blocked and (aborted_by_blocks or produced == 0):
-        reason = (
-            f"backfill-honest-status: {source} остановлен блоками источника — "
-            f"blocked={blocked}, обогащено {enriched} из {attempted} попыток{hint} (#2674)"
-        )
-        logger.error("%s run_id=%d", reason, run_id)
         # Counter() принимает и Collection (считает элементы — старые set/list-вызовы),
         # и Mapping (копирует кратности как есть — census от вызывающего) одним и тем
         # же конструктором.
         census = Counter(ban_kinds)
         if census:
             counters["ban_kinds"] = dict(census)  # type: ignore[assignment]
+        dominant = _dominant_ban_kind(census)
+
+        if dominant == BAN_KIND_INFRA and produced > 0:
+            # #3288: 'banned' означает «площадка нас заблокировала» — и читается так
+            # же (триггер ротации IP, алерты, разбор простоя). Прогон 5425 при 41
+            # infra из 48 «блоков» честно обогатил 41 карточку — его оборвал брейкер
+            # по доле, а не площадка, — и всё равно рапортовал «остановлен блоками
+            # источника». Понижаем ровно этот случай: диагноз infra И прогон работу
+            # сделал → 'done'.
+            #
+            # Нулевой прогон с infra остаётся 'banned' — контракт #2764/#3196:
+            # там диагноз несёт ban_kind строки ('infra'), а статус говорит «прогон
+            # оборван отказами». Понижать его до 'failed' по одному лишь диагнозу
+            # опаснее исходного дефекта: при пустом/отсутствующем census (источник
+            # видов не различает) dominant='unknown', а настоящий бан площадки,
+            # опознанный как infra по 5xx, спрятался бы под «нашей поломкой».
+            reason = (
+                f"backfill-honest-status: {source} оборван брейкером на отказах НАШЕГО "
+                f"тракта — blocked={blocked} (диагноз '{BAN_KIND_INFRA}' у большинства), "
+                f"обогащено {enriched} из {attempted} попыток{hint} (#3288)"
+            )
+            logger.error("%s run_id=%d", reason, run_id)
+            mark_done(db, run_id, counters)
+            return
+
+        reason = (
+            f"backfill-honest-status: {source} остановлен блоками источника — "
+            f"blocked={blocked}, обогащено {enriched} из {attempted} попыток{hint} (#2674)"
+        )
+        logger.error("%s run_id=%d", reason, run_id)
         mark_banned(
             db,
             run_id,
             reason,
             counters,
-            ban_kind=_dominant_ban_kind(census),
+            ban_kind=dominant,
         )
         return
 

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

@@ -80,6 +80,7 @@ from scraper_kit.providers.avito.detail import (
     save_detail_enrichment,
 )
 from scraper_kit.providers.avito.serp import AvitoScraper
+from scraper_kit.proxy_errors import NoProxyAvailableError
 from scraper_kit.snapshot_writer import upsert_listing_snapshot
 from sqlalchemy import text
 from sqlalchemy.orm import Session
@@ -173,6 +174,34 @@ def _top_failure(census: Counter[str]) -> str | None:
     return f"{reason} ({hits} из {sum(census.values())})"
 
 
+def _iter_causes(exc: BaseException) -> list[BaseException]:
+    """Цепочка причин исключения, без зацикливания (копия приёма из
+    domclick_detail_backfill — там же он и обкатан на #3283)."""
+    seen: set[int] = set()
+    out: list[BaseException] = []
+    cur: BaseException | None = exc
+    while cur is not None and id(cur) not in seen:
+        out.append(cur)
+        seen.add(id(cur))
+        cur = cur.__cause__ or cur.__context__
+    return out
+
+
+def _caused_by_empty_pool(exc: BaseException) -> bool:
+    """Прячется ли за этим «блоком» пустой пул прокси (#3288, как #3283 у домклика).
+
+    `NoProxyAvailableError` документирован ровно как «НАША инфраструктура, не
+    внешний блок», и поднимается ДО HTTP-запроса: к площадке мы не ходили вовсе.
+    Сюда он попадает под видом блокировки, потому что fetch_detail заворачивает
+    в `AvitoSidecarUnavailableError` любое исключение фетча.
+
+    Опора — ТИП в цепочке `__cause__`/`__context__`, а не подстрока «no proxy
+    available» в тексте: текст обёртки её действительно содержит, но ровно так
+    #3272 уже один раз объявил блоком пользовательское описание квартиры.
+    """
+    return any(isinstance(c, NoProxyAvailableError) for c in _iter_causes(exc))
+
+
 # Провайдер mobileproxy.space почти всегда возвращает "rt" (секунды на переподключение
 # канала после ротации) в RotationResult.reconnect_delay_s -- см. app.services.
 # proxy_rotation docstring. Дефолт нужен ТОЛЬКО если провайдер его не прислал
@@ -529,6 +558,8 @@ async def run_avito_detail_backfill(
         consecutive_failures = 0
         aborted_by_blocks = False
         abort_reason: str | None = None
+        # #3288: прогон оборван пустым пулом прокси — «нечем ходить», а не бан.
+        no_proxy_stop = False
         do_sleep = False
         items_since_warm = 0
         # Счётчик попыток (успех ИЛИ отказ — оба тратят бюджет IP одинаково, см.
@@ -765,10 +796,33 @@ async def run_avito_detail_backfill(
                 )
 
             except (AvitoBlockedError, AvitoRateLimitedError) as e:
-                breaker.record_block()
+                # #3288: пустой пул — не блок и не отказ площадки: запрос не
+                # уходил вовсе, следующая карточка упрётся ровно в то же самое.
+                # Сюда он приезжает под видом блокировки, потому что fetch_detail
+                # заворачивает в AvitoSidecarUnavailableError любое исключение
+                # фетча. Исход честно failed (тот же разряд, что транспортные
+                # сбои ниже) — тождество attempted = enriched + blocked + gone +
+                # failed остаётся целым (#3338), а причину несёт no_proxy_stop=1
+                # и mark_failed с текстом про пул, как у домклика после #3283.
+                if _caused_by_empty_pool(e):
+                    counters.failed += 1
+                    logger.error(
+                        "avito_detail_backfill: run_id=%d СТОП — пул прокси пуст, "
+                        "к площадке не ходили. enriched=%d attempted=%d",
+                        run_id,
+                        counters.enriched,
+                        counters.attempted,
+                    )
+                    no_proxy_stop = True
+                    break
+
+                ban_kind = ban_kind_of_exception(e)
+                # #3288: в окно доли идёт только 'platform' — infra (отказ нашего
+                # сайдкара) уходит в знаменатель, см. BlockRatioBreaker.record_block.
+                breaker.record_block(ban_kind)
                 counters.blocked += 1
                 failure_census[_failure_signature(e)] += 1
-                block_ban_kinds[ban_kind_of_exception(e)] += 1
+                block_ban_kinds[ban_kind] += 1
                 do_sleep = False
                 # #3251/#3283g: на настоящий бан ПЛОЩАДКОЙ (AvitoBlockedError и подтипы:
                 # AvitoContentBlockedError, AvitoWarmupCookiesMissingError; НЕ на
@@ -982,6 +1036,29 @@ async def run_avito_detail_backfill(
         # и без этого ключа "banned" опять не отличить по причине (#3178).
         if abort_reason is not None:
             current_counters["abort_reason"] = abort_reason  # type: ignore[assignment]
+        if no_proxy_stop:
+            # #3288 (как #3283 у домклика): остановка из-за пустого пула — НЕ блок,
+            # поэтому и не aborted_by_blocks: иначе прогон уйдёт в 'banned' и запись
+            # будет утверждать про площадку то, чего не было. Это отказ нашей стороны.
+            current_counters["no_proxy_stop"] = 1
+            runs_mod.mark_failed(
+                db,
+                run_id,
+                "пул прокси пуст — к площадке не ходили (#3288)",
+                current_counters,
+            )
+            logger.info(
+                "avito_detail_backfill: run_id=%d FINISHED (пул пуст) -- attempted=%d "
+                "enriched=%d blocked=%d gone=%d failed=%d duration=%.1fs",
+                run_id,
+                counters.attempted,
+                counters.enriched,
+                counters.blocked,
+                counters.gone,
+                counters.failed,
+                counters.duration_sec,
+            )
+            return counters
         runs_mod.mark_backfill_finished(
             db,
             run_id,

tradein-mvp/backend/tests/test_3288b_block_kind_ratio.py

@@ -0,0 +1,373 @@
+"""Вид блока решает обрыв по доле и статус прогона (#3288, часть B).
+
+Прогон 5425 оборвался по доле блоков (`abort_reason='ratio'`, статус `banned`) на
+48 «блоках», из которых 41 был `AvitoSidecarUnavailableError` — отказ НАШЕГО тракта,
+а не площадки: та же карточка не доехала до Авито, и 41 других карточек прогон
+успел честно обогатить. `record_block()` вида не принимал вовсе, поэтому в числитель
+скользящего окна (#3184) падало всё подряд.
+
+Проверка ПО ЗНАЧЕНИЮ — сравниваются `abort_reason()`, счётчики и то, какой именно
+финализатор дёрнут (`mark_banned` / `mark_failed` / `mark_done`), а не «не бросило
+исключение». К каждому послаблению идёт контроль на противоположную ошибку: 14
+настоящих `AvitoBlockedError` из 20 обязаны рвать прогон как раньше (#3184 не
+меняется), а пустой пул опознаётся по ЦЕПОЧКЕ `__cause__`, не по подстроке
+«no proxy available» — на подстроке уже обжёгся #3272.
+
+Харнесс бэкфилла зеркалит tests/test_3338_backfill_counter_identity.py, оттуда же
+тождество attempted = enriched + blocked + gone + failed.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+import pytest  # noqa: E402
+from scraper_kit.avito_exceptions import (  # noqa: E402
+    AvitoBlockedError,
+    AvitoSidecarUnavailableError,
+)
+from scraper_kit.proxy_errors import NoProxyAvailableError  # noqa: E402
+
+from app.services import scrape_runs as runs_svc  # noqa: E402
+from app.services.backfill_block_breaker import BlockRatioBreaker  # noqa: E402
+from app.services.scrape_runs import (  # noqa: E402
+    BAN_KIND_INFRA,
+    BAN_KIND_PLATFORM,
+    BAN_KIND_UNKNOWN,
+)
+from app.tasks.avito_detail_backfill import (  # noqa: E402
+    AvitoDetailBackfillResult,
+    run_avito_detail_backfill,
+)
+
+_A_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
+_A_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
+_A_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
+_A_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
+_A_SETTINGS = "app.tasks.avito_detail_backfill.settings"
+_A_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
+_A_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
+_A_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"
+_A_ROTATE_PROXY = "app.tasks.avito_detail_backfill.rotate_proxy"
+
+_WINDOW = 20
+_THRESHOLD = 0.7
+
+
+def _breaker(snapshot_size: int = 100) -> BlockRatioBreaker:
+    return BlockRatioBreaker(
+        window_size=_WINDOW,
+        ratio_threshold=_THRESHOLD,
+        safety_min=5,
+        snapshot_size=snapshot_size,
+    )
+
+
+# ── брейкер: что попадает в числитель доли ────────────────────────────────────
+
+
+def test_twenty_sidecar_blocks_do_not_abort_by_ratio() -> None:
+    """20 подряд отказов сайдкара при снапшоте больше окна — не 'ratio' (приёмка)."""
+    breaker = _breaker()
+    for _ in range(20):
+        breaker.record_block(BAN_KIND_INFRA)
+
+    assert breaker.abort_reason() is None, (
+        f"abort_reason={breaker.abort_reason()!r} при 20 infra-отказах: площадка "
+        "не отказывала ни разу, рвать прогон не по чему"
+    )
+    assert (breaker.window_blocks, breaker.window_len) == (0, 20), (
+        f"окно {breaker.window_blocks}/{breaker.window_len}: infra идёт в ЗНАМЕНАТЕЛЬ "
+        "(как record_failure), а не в числитель"
+    )
+    assert breaker.consecutive_blocks == 0, (
+        f"consecutive_blocks={breaker.consecutive_blocks}: infra не двигает серию, "
+        "иначе на коротком снапшоте прогон рвал бы safety-net по тем же отказам"
+    )
+
+
+def test_fourteen_platform_blocks_of_twenty_still_abort_by_ratio() -> None:
+    """Контроль на противоположную ошибку: #3184 не меняется (14/20 = порог 0.7)."""
+    breaker = _breaker()
+    for _ in range(6):
+        breaker.record_success()
+    for _ in range(14):
+        breaker.record_block(BAN_KIND_PLATFORM)
+
+    assert breaker.abort_reason() == "ratio", (
+        f"abort_reason={breaker.abort_reason()!r} при 14 настоящих блоках из 20 — "
+        "критерий #3184 обязан сработать ровно на пороге"
+    )
+    assert breaker.window_blocks == 14, f"числитель {breaker.window_blocks}, ожидали 14"
+
+
+def test_infra_does_not_pad_the_numerator_to_the_threshold() -> None:
+    """13 platform + 7 infra = 0.65, а не 1.0: infra числитель не добивает."""
+    breaker = _breaker()
+    for _ in range(7):
+        breaker.record_block(BAN_KIND_INFRA)
+    for _ in range(13):
+        breaker.record_block(BAN_KIND_PLATFORM)
+
+    assert breaker.abort_reason() is None, (
+        f"abort_reason={breaker.abort_reason()!r}: 13/20 = 0.65 ниже порога 0.7, "
+        "прогон рвётся только если infra зачли блоками"
+    )
+
+
+# ── бэкфилл: тот же вопрос сквозь задачу ──────────────────────────────────────
+
+
+def _fake_avito_settings() -> MagicMock:
+    return MagicMock(
+        scraper_fetch_mode="browser",
+        avito_detail_backfill_use_curl=False,
+        detail_backfill_block_ratio_window=_WINDOW,
+        detail_backfill_block_ratio_threshold=_THRESHOLD,
+        browser_http_endpoint="http://browser:9000",
+        avito_detail_backfill_rotate_after_attempts=15,
+        avito_detail_backfill_rotate_on_ban_max=0,
+        avito_detail_backfill_rotate_on_ban_min_gap=10,
+    )
+
+
+def _mock_avito_db(n: int) -> MagicMock:
+    snapshot = [
+        {
+            "id": i + 1,
+            "source_url": f"https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_{i + 1}",
+        }
+        for i in range(n)
+    ]
+    db = MagicMock()
+    sel = MagicMock()
+    sel.mappings.return_value.all.return_value = snapshot
+    db.execute.return_value = sel
+    return db
+
+
+def _mock_browser_fetcher_cls() -> MagicMock:
+    instance = AsyncMock()
+    instance.__aenter__ = AsyncMock(return_value=instance)
+    instance.__aexit__ = AsyncMock(return_value=False)
+    instance.request_context_reset = MagicMock()
+    instance.lease_id = 42
+    return MagicMock(return_value=instance)
+
+
+async def _run_avito(
+    fetch_results: list[Any], *, snapshot_size: int
+) -> tuple[AvitoDetailBackfillResult, MagicMock]:
+    runs = MagicMock()
+    db = _mock_avito_db(snapshot_size)
+    with (
+        patch(_A_SETTINGS, _fake_avito_settings()),
+        patch(_A_SESSION),
+        patch(_A_SCRAPER),
+        patch(_A_RUNS, runs),
+        patch(_A_BROWSER_FETCHER, _mock_browser_fetcher_cls()),
+        patch(_A_FETCH, AsyncMock(side_effect=fetch_results)),
+        patch(_A_SAVE, return_value=True),
+        patch(_A_ROTATE_PROXY, AsyncMock()),
+        patch(_A_SLEEP, new_callable=AsyncMock),
+    ):
+        counters = await run_avito_detail_backfill(
+            db,
+            run_id=3288,
+            params={
+                "batch_size": snapshot_size,
+                "budget_sec": 3600,
+                "max_consecutive_blocks": 10,
+            },
+        )
+    return counters, runs
+
+
+def _assert_identity(c: AvitoDetailBackfillResult, *, expected_attempted: int) -> None:
+    outcomes = c.enriched + c.blocked + c.gone + c.failed
+    assert (c.attempted, outcomes) == (expected_attempted, expected_attempted), (
+        f"attempted={c.attempted} vs исходы={outcomes} "
+        f"(enriched={c.enriched} blocked={c.blocked} gone={c.gone} failed={c.failed}), "
+        f"ожидали {expected_attempted} — попытка без исхода (#3338)"
+    )
+
+
+@pytest.mark.asyncio
+async def test_sidecar_run_walks_the_whole_snapshot_and_is_not_banned() -> None:
+    """20 отказов сайдкара + 10 успехов: снапшот пройден целиком, обрыва нет."""
+    fetches: list[Any] = [AvitoSidecarUnavailableError("sidecar 503") for _ in range(20)]
+    fetches += [MagicMock() for _ in range(10)]
+
+    counters, runs = await _run_avito(fetches, snapshot_size=30)
+
+    finished = runs.mark_backfill_finished.call_args
+    assert finished is not None, "прогон не финализирован mark_backfill_finished"
+    # Приёмка #3288 первой строкой: величина, по которой прогон 5425 и оборвался.
+    assert finished.args[2].get("abort_reason") is None, (
+        f"abort_reason={finished.args[2].get('abort_reason')!r} при 20 отказах сайдкара: "
+        "обрыва по доле быть не могло — площадка не отказала ни разу"
+    )
+    _assert_identity(counters, expected_attempted=30)
+    assert (counters.blocked, counters.enriched) == (20, 10), (
+        f"blocked={counters.blocked} enriched={counters.enriched}: прогон обязан дойти "
+        "до конца снапшота, отказ нашего тракта его не рвёт"
+    )
+    assert finished.kwargs["aborted_by_blocks"] is False, (
+        "aborted_by_blocks=True при нулевых отказах площадки — прогон уйдёт в 'banned'"
+    )
+    assert finished.kwargs["ban_kinds"] == {BAN_KIND_INFRA: 20}, (
+        f"ban_kinds={finished.kwargs['ban_kinds']}: перепись диагнозов не должна "
+        "теряться — она и решает статус"
+    )
+
+
+@pytest.mark.asyncio
+async def test_real_platform_bans_still_abort_by_ratio() -> None:
+    """Контроль: 14 настоящих AvitoBlockedError из 20 рвут прогон как раньше."""
+    fetches: list[Any] = [MagicMock() for _ in range(6)]
+    fetches += [AvitoBlockedError("firewall/soft-block") for _ in range(14)]
+
+    counters, runs = await _run_avito(fetches, snapshot_size=30)
+
+    _assert_identity(counters, expected_attempted=20)
+    finished = runs.mark_backfill_finished.call_args
+    assert finished.args[2].get("abort_reason") == "ratio", (
+        f"counters={finished.args[2]}: 14 отказов площадки из 20 — ровно порог #3184"
+    )
+    assert finished.kwargs["aborted_by_blocks"] is True, "обрыв по доле обязан быть помечен"
+
+
+@pytest.mark.asyncio
+async def test_empty_pool_stops_run_with_its_own_reason() -> None:
+    """Пустой пул — не блок: прогон завершается 'нечем ходить', а не баном."""
+    wrapped = AvitoSidecarUnavailableError("Avito detail browser fetch failed for https://...")
+    wrapped.__cause__ = NoProxyAvailableError("avito")
+
+    counters, runs = await _run_avito([wrapped], snapshot_size=30)
+
+    _assert_identity(counters, expected_attempted=1)
+    assert (counters.blocked, counters.failed) == (0, 1), (
+        f"blocked={counters.blocked} failed={counters.failed}: к площадке не ходили — "
+        "это отказ нашей стороны, а не блок"
+    )
+    assert runs.mark_backfill_finished.call_args is None, (
+        "прогон с пустым пулом не должен идти через backfill-honest-status: у него своя причина"
+    )
+    failed = runs.mark_failed.call_args
+    assert failed is not None, "прогон не финализирован mark_failed"
+    assert failed.args[3]["no_proxy_stop"] == 1, f"counters={failed.args[3]}: нет no_proxy_stop"
+    assert "пул прокси пуст" in failed.args[2], f"причина={failed.args[2]!r}"
+
+
+@pytest.mark.asyncio
+async def test_empty_pool_recognised_by_cause_not_by_text() -> None:
+    """#3272: подстрока «no proxy available» в тексте — не доказательство пустого пула."""
+    liar = AvitoBlockedError("firewall page: no proxy available for provider='avito'")
+
+    counters, runs = await _run_avito([liar, MagicMock()], snapshot_size=2)
+
+    _assert_identity(counters, expected_attempted=2)
+    assert (counters.blocked, counters.enriched) == (1, 1), (
+        f"blocked={counters.blocked} enriched={counters.enriched}: отказ площадки с "
+        "«no proxy available» в тексте остаётся блоком — цепочка причин пуста"
+    )
+    assert runs.mark_failed.call_args is None, (
+        "прогон остановлен как «пул пуст» по подстроке в тексте — ровно ошибка #3272"
+    )
+
+
+# ── статус прогона: диагноз infra ≠ 'banned' ──────────────────────────────────
+
+
+def _finish(ban_kinds: dict[str, int], *, enriched: int) -> tuple[MagicMock, ...]:
+    counters = {"attempted": 90, "enriched": enriched, "blocked": 48, "gone": 0, "failed": 1}
+    with (
+        patch.object(runs_svc, "mark_banned") as banned,
+        patch.object(runs_svc, "mark_failed") as failed,
+        patch.object(runs_svc, "mark_done") as done,
+    ):
+        runs_svc.mark_backfill_finished(
+            MagicMock(),
+            5425,
+            counters,
+            source="avito_detail_backfill",
+            aborted_by_blocks=True,
+            ban_kinds=ban_kinds,
+        )
+    return banned, failed, done
+
+
+def test_infra_dominant_run_is_not_marked_banned() -> None:
+    """41 infra из 48 при 41 обогащённой карточке — прогон 'done', не 'banned'."""
+    banned, failed, done = _finish({BAN_KIND_INFRA: 41, BAN_KIND_PLATFORM: 7}, enriched=41)
+
+    assert banned.call_count == 0, (
+        "прогон помечен 'banned' при доминирующем infra — запись утверждает про "
+        "площадку то, чего не было"
+    )
+    assert (failed.call_count, done.call_count) == (0, 1), (
+        f"mark_failed={failed.call_count} mark_done={done.call_count}: 41 карточка "
+        "обогащена, прогон работу сделал"
+    )
+
+
+def test_pure_infra_run_with_result_is_not_banned() -> None:
+    """Основной кейс без примеси: перепись целиком infra, карточки есть → не 'banned'."""
+    banned, failed, done = _finish({BAN_KIND_INFRA: 20}, enriched=10)
+
+    assert (banned.call_count, failed.call_count, done.call_count) == (0, 0, 1), (
+        f"mark_banned={banned.call_count} mark_failed={failed.call_count} "
+        f"mark_done={done.call_count}: 20 отказов сайдкара при 10 обогащённых — 'done'"
+    )
+
+
+def test_infra_dominant_run_without_result_stays_banned() -> None:
+    """Контроль на ОБРАТНУЮ ошибку: ноль результата — по-прежнему 'banned' (#2764/#3196).
+
+    Диагноз infra на нулевом прогоне неотличим от бана площадки, опознанного по 5xx
+    (yandex, #3196): статус остаётся 'banned', честность несёт ban_kind. Понижение
+    до 'failed' прятало бы настоящий бан под «нашу поломку» — падало 2 теста CI.
+    """
+    banned, failed, done = _finish({BAN_KIND_INFRA: 41, BAN_KIND_PLATFORM: 7}, enriched=0)
+
+    assert (banned.call_count, failed.call_count, done.call_count) == (1, 0, 0), (
+        f"mark_banned={banned.call_count} mark_failed={failed.call_count} "
+        f"mark_done={done.call_count}: нулевой прогон обязан остаться 'banned'"
+    )
+    assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_INFRA, (
+        f"ban_kind={banned.call_args.kwargs['ban_kind']!r}: диагноз обязан доехать"
+    )
+
+
+def test_run_without_census_stays_banned() -> None:
+    """Источник видов не различает (ban_kinds пуст) → прежнее поведение: 'banned'."""
+    banned, failed, done = _finish({}, enriched=41)
+
+    assert (banned.call_count, failed.call_count, done.call_count) == (1, 0, 0), (
+        f"mark_banned={banned.call_count} mark_failed={failed.call_count} "
+        f"mark_done={done.call_count}: без переписи диагноза нет — статус не меняем"
+    )
+    assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_UNKNOWN, (
+        f"ban_kind={banned.call_args.kwargs['ban_kind']!r}, ожидали 'unknown'"
+    )
+
+
+def test_platform_dominant_run_is_still_banned() -> None:
+    """Контроль на противоположную ошибку: настоящий бан площадки — по-прежнему 'banned'."""
+    banned, failed, done = _finish({BAN_KIND_PLATFORM: 41, BAN_KIND_INFRA: 7}, enriched=41)
+
+    assert banned.call_count == 1, (
+        f"mark_banned={banned.call_count} (failed={failed.call_count} done={done.call_count}): "
+        "площадка отказала в 41 блоке из 48 — это ровно 'banned'"
+    )
+    assert banned.call_args.kwargs["ban_kind"] == BAN_KIND_PLATFORM, (
+        f"ban_kind={banned.call_args.kwargs['ban_kind']!r}, ожидали 'platform'"
+    )