Merge pull request 'fix(tradein/scrapers): прогон больше не рапортует «done» поверх провала и нуля' (#2892) from fix/tradein-honest-run-status into main

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

@@ -143,6 +143,28 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
 #   unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
 #                    сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
 #                    реально дали ноль.
+#   succeeded      — yandex_newbuilding_sweep (42 прогона/90д) и newbuilding_enrich
+#                    (65 прогонов/90д, единственные два писателя ключа на проде,
+#                    проверено 2026-08-15). НЕ 'rows_inserted': тот ключ пишет ЕЩЁ и
+#                    rosreestr_dkp_import (67 прогонов/90д) — у него rows_inserted=0 в
+#                    66 из 67 это ЗДОРОВЫЙ ответ догнавшего инкрементального импорта
+#                    (rows_fetched=rows_skipped=96974, last_id не двигается неделями),
+#                    а не отказ; если бы 'rows_inserted' попал в этот список, сторож
+#                    зачитывал бы этот здоровый ноль как измеренный провал и копил бы
+#                    практически непрерываемый стрик (rosreestr_dkp_import не
+#                    прерывается другим статусом — импорт либо 'done', либо не бежал).
+#                    НЕ 'processed' по той же причине с другой стороны: это счётчик
+#                    ПОПЫТОК (у newbuilding_enrich processed==attempted==limit даже
+#                    когда succeeded меньше — прод-факт 09.08: processed=25 succeeded=14,
+#                    44% отказов замаскировались бы под measured-25) — сторож нулевого
+#                    результата на нём молчал бы ровно там, где должен сработать, а на
+#                    будущем опустении очереди домов (cian_houses_pending) создал бы
+#                    свой вечный ложный zero-стрик. 'succeeded' у yandex_newbuilding_sweep
+#                    численно совпадает с 'rows_inserted' на всех 42/42 прод-прогонах —
+#                    замена не теряет исходную цель (десять прогонов подряд 26.07-10.08,
+#                    все 'done', succeeded=0 rows_inserted=0 failed_resolve=4-5 — раньше
+#                    ни total_seen/lots_fetched/unique_fetched не было, и
+#                    _run_result_count всегда возвращал None (honest-run-status)).
 # Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
 # прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
 # (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
@@ -150,7 +172,12 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
 # трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
 # (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
 # словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
-_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
+_RESULT_COUNTER_KEYS = (
+    "total_seen",
+    "lots_fetched",
+    "unique_fetched",
+    "succeeded",
+)
 
 
 def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
@@ -282,6 +309,63 @@ def _phase_totally_failed(counters: Mapping[str, Any]) -> str | None:
     return None
 
 
+# honest-run-status (2026-08-15): доля отказов, которая обесценивает формально ненулевой
+# сбор. Прод-факт avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,
+# "blocked":1} — 89% попыток отказали, а mark_backfill_finished всё равно звал mark_done,
+# потому что "produced != 0" (6 обогащено). Ни _sweep_run_did_nothing (нужны
+# anchors_total/errors_count, у backfill'ов их нет), ни _phase_totally_failed (нужна пара
+# "<phase>_attempted"/"<phase>_failed" — здесь голые "attempted"/"failed" без фазового
+# префикса, `"attempted".endswith("_attempted")` не матчит) эту форму counters не ловят —
+# обе проверки написаны под СВОИ формы, а не под backfill'овскую.
+#
+# Порог 'failed' — половина и больше отказов: сбор для практических целей провалился,
+# даже если несколько записей всё же обогатились. Порог 'partial' НЕ заведён отдельным
+# статусом scrape_runs.status — это потребовало бы миграции (DROP+ADD CHECK constraint,
+# 051_scrape_runs_extend.sql) и обучило бы новому значению ещё 4 места (Literal-фильтр
+# admin API, хардкод статусов фронта, оба IN-списка сторожей) — тот же класс "оборванной
+# проводки", из-за которого заведён #2686/ban_kind. Вместо статуса — тот же диагноз, что и
+# у ban_kind: causa в тексте `error`, терминальный статус один ('failed'). 0.15..0.5 —
+# та же 'failed', но с другой формулировкой причины ("деградировал", не "провалился"), чтобы
+# оператор видел разницу читая error, не только status.
+FAILED_RATIO_FAILED_THRESHOLD = 0.5
+FAILED_RATIO_DEGRADED_THRESHOLD = 0.15
+# Минимум попыток, при котором доля вообще что-то значит — иначе 1 отказ из 2 (=0.5)
+# палит статус на шуме единичного случая. То же рассуждение и то же число, что у
+# _PHASE_MIN_ATTEMPTS (см. выше).
+_FAILED_RATIO_MIN_ATTEMPTS = _PHASE_MIN_ATTEMPTS
+
+
+def _failed_ratio_too_high(counters: Mapping[str, Any]) -> str | None:
+    """Прогон, у которого доля отказов слишком велика, даже если что-то собрано.
+
+    Возвращает текст причины (для error) либо None. Читает ГОЛЫЕ ключи "attempted"/
+    "failed" (без фазового префикса) — сейчас это словарь только у четырёх
+    detail-backfill'ов (avito/yandex/domclick/newbuilding_enrich), все идут через
+    mark_backfill_finished → mark_done. `attempted < _FAILED_RATIO_MIN_ATTEMPTS` или
+    отсутствие любого из ключей → None (нечем/не о чём судить — счётчики либо не
+    заполнены, либо принадлежат другому источнику со своим словарём).
+
+    Что признак НЕ доказывает: КТО виноват (площадка, наш прокси, наш парсер) — поэтому
+    'failed' без диагноза, как и у #2625/#2700/#2764.
+    """
+    attempted = _pick_int(counters, "attempted")
+    failed = _pick_int(counters, "failed")
+    if attempted is None or failed is None or attempted < _FAILED_RATIO_MIN_ATTEMPTS:
+        return None
+    ratio = failed / max(attempted, 1)
+    if ratio >= FAILED_RATIO_FAILED_THRESHOLD:
+        verb = "провалился"
+    elif ratio >= FAILED_RATIO_DEGRADED_THRESHOLD:
+        verb = "деградировал"
+    else:
+        return None
+    return (
+        f"failed-ratio-honest-status: сбор {verb} — {failed} из {attempted} попыток "
+        f"отказали (доля {ratio:.0%}); формально ненулевой результат этого не искупает. "
+        f"Причина НЕ установлена — статус 'failed' без диагноза"
+    )
+
+
 def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
 
@@ -292,13 +376,24 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
 
     Приоритет ключей:
-      - total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
-      - new_count  ← 'new_count' (если уже есть) иначе 'lots_inserted'
+      - total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched /
+        succeeded)
+      - new_count  ← 'new_count' / 'lots_inserted' / 'saved_inserted' / 'rows_inserted'
+        (первый присутствующий). 'saved_inserted' — full-load'ы (cian/avito/yandex,
+        CianFullLoadCounters и аналоги в pipeline.py): на проде витрина показывала
+        new_count=0 у трёх подряд cian_full_load при реально сохранённых
+        saved_inserted=482/214/239 (honest-run-status) — ключ 'new_count'/'lots_inserted'
+        у full-load'ов в counters не пишется вовсе. 'rows_inserted' — тот же ключ,
+        которым yandex_newbuilding_sweep и rosreestr_dkp_import сообщают число upsert'ов;
+        здесь (для витринной колонки new_count) это безопасно — в отличие от
+        _RESULT_COUNTER_KEYS этот список не участвует в подсчёте zero-result-стрика.
 
     Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
     тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
     """
-    return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
+    return _run_result_count(counters), _pick_int(
+        counters, "new_count", "lots_inserted", "saved_inserted", "rows_inserted"
+    )
 
 
 def _alert_if_consecutive_failures(db: Session, source: str) -> None:
@@ -558,6 +653,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
     #2700: там же — отказ называть успехом прогон, у которого отказала КАЖДАЯ попытка
     целой фазы (см. _phase_totally_failed). Отличие от #2625: тот случай про «не сделано
     ничего», этот — про «одно направление работы мертво, а суммарный сбор это прячет».
+
+    honest-run-status: там же — отказ называть успехом прогон с высокой долей отказов,
+    даже если собрано > 0 (см. _failed_ratio_too_high). Отличие от #2625/#2700: те два
+    смотрят на «всё или ничего» (все якоря / вся фаза), этот — на ДОЛЮ отказов у
+    detail-backfill'ов, где ни один из первых двух признаков не матчит форму counters.
     """
     did_nothing = _sweep_run_did_nothing(counters)
     if did_nothing is not None:
@@ -569,6 +669,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
         logger.error("%s run_id=%d", phase_dead, run_id)
         mark_failed(db, run_id, phase_dead, counters)
         return
+    ratio_bad = _failed_ratio_too_high(counters)
+    if ratio_bad is not None:
+        logger.error("%s run_id=%d", ratio_bad, run_id)
+        mark_failed(db, run_id, ratio_bad, counters)
+        return
     total_seen, new_count = _column_counts(counters)
     row = db.execute(
         text(

tradein-mvp/backend/tests/test_backfill_honest_status.py

@@ -5,7 +5,17 @@
 1500-1600 попыток без единого обогащения), yandex 31/52, domclick 24/30
 (494 попытки → 0 обогащено, 63 блока, 431 fail — и все 30 'done').
 
-Проверяем ровно ветвление mark_backfill_finished — БД замокана.
+Проверяем ровно ветвление mark_backfill_finished — БД замокана (mark_done/mark_failed/
+mark_banned здесь fake-заглушки, регистрирующие ТОЛЬКО факт вызова). Это значит: кейсы
+ниже с высокой долей отказов (attempted=50, failed=38 или 36 — 76%/72%), ожидающие
+'done', проверяют лишь то, КАКОЙ финализатор ВЫБРАЛ mark_backfill_finished (#2674:
+"обогатили хоть что-то — успех"), а НЕ то, что реально запишет в БД mark_done. С
+honest-run-status (2026-08-15) mark_done САМ переквалифицирует такой прогон в 'failed'
+через _failed_ratio_too_high (доля отказов >= 0.5) — реальный терминальный статус
+для этих двух кейсов на проде теперь 'failed', не 'done'. Это намеренно проверяется
+отдельно, БЕЗ мока mark_done, в tests/test_honest_run_status_failed_ratio.py
+(test_prod_fact_avito_15_08_no_longer_done и соседние) — не читай эти два кейса как
+"76%/72% отказов = 'done' в проде".
 """
 
 from __future__ import annotations
@@ -59,9 +69,15 @@ def _finish(counters: dict[str, int], *, aborted: bool = False) -> tuple[str, st
         ({"attempted": 5, "enriched": 0, "failed": 5}, False, "failed"),
         # Кандидатов не было — честная пустота, это успех.
         ({"attempted": 0, "enriched": 0, "blocked": 0, "failed": 0}, False, "done"),
-        # Частичный прогон: обогатили хоть что-то → успех.
+        # Частичный прогон: обогатили хоть что-то → mark_backfill_finished ВЫБИРАЕТ
+        # mark_done как финализатор (#2674). 76% отказов (38 из 50) — здесь mark_done
+        # замокан, поэтому статус остаётся 'done'; в реальном mark_done с
+        # honest-run-status (2026-08-15) это переквалифицируется в 'failed'
+        # (_failed_ratio_too_high, доля >= 0.5) — см. докстринг модуля.
         ({"attempted": 50, "enriched": 12, "blocked": 0, "failed": 38}, False, "done"),
-        # Блоки были, но прогон доработал и обогатил — не бан.
+        # Блоки были, но прогон доработал и обогатил — mark_backfill_finished выбирает
+        # НЕ 'banned'. 72% отказов (36 из 50) — та же оговорка: реальный mark_done
+        # переквалифицирует в 'failed', см. докстринг модуля выше.
         ({"attempted": 50, "enriched": 12, "blocked": 2, "failed": 36}, False, "done"),
         # Блок оборвал прогон, хотя часть успели обогатить — работа не доделана.
         ({"attempted": 50, "enriched": 12, "blocked": 5, "failed": 33}, True, "banned"),

tradein-mvp/backend/tests/test_honest_run_status_failed_ratio.py

@@ -0,0 +1,336 @@
+"""honest-run-status (2026-08-15): статус прогона не должен рапортовать 'done' поверх
+провала или нуля. Три прод-факта закрыты этой правкой:
+
+  (a) avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,"blocked":1}
+      -> status='done' — 89% отказов, статус зелёный. mark_backfill_finished звал
+      mark_done, потому что produced=6 (>0); ни _sweep_run_did_nothing (нет
+      anchors_total/errors_count у backfill'ов), ни _phase_totally_failed (ключи
+      "attempted"/"failed" без фазового префикса) эту форму counters не ловили.
+      Фикс: _failed_ratio_too_high внутри mark_done.
+
+  (b) yandex_newbuilding_sweep 26.07-10.08: десять прогонов подряд 'done' при
+      processed=5, succeeded=0, rows_inserted=0, failed_resolve=4-5 — сторож нулевого
+      результата (_alert_if_consecutive_zero_results) слеп, т.к. _RESULT_COUNTER_KEYS
+      не знал ни одного ключа этого sweep'а (total_seen/lots_fetched/unique_fetched).
+      Фикс: _RESULT_COUNTER_KEYS дополнен 'succeeded'. Первая версия правки добавляла
+      голые 'rows_inserted'/'processed' — ревью нашло, что 'rows_inserted' пишет ЕЩЁ
+      rosreestr_dkp_import (66/67 прод-прогонов, здоровый ноль догнавшего импорта, а не
+      отказ) и завёл бы непрерываемый ложный zero-стрик, а 'processed' — счётчик
+      попыток (==limit даже при частичном провале у newbuilding_enrich) и маскирует
+      реальные отказы. 'succeeded' пишут только yandex_newbuilding_sweep и
+      newbuilding_enrich, численно совпадает с прежним 'rows_inserted' на всех
+      прод-прогонах sweep'а — см. test_rosreestr_dkp_import_healthy_zero_stays_unmeasured
+      и test_newbuilding_enrich_partial_failure_not_masked_by_processed ниже.
+
+  (c) admin-витрина показывала new_count=0 у трёх подряд cian_full_load, хотя реально
+      сохранено saved_inserted=482/214/239 — full-load'ы не пишут ни 'new_count', ни
+      'lots_inserted'. Фикс: _column_counts дополнен saved_inserted/rows_inserted.
+
+Проверяем на обоих модулях (kit-копия и app-копия — байт-эквивалентны по докстрингу
+runs.py), тем же паттерном, что test_2625_run_that_did_nothing.py.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from scraper_kit.orchestration import runs as kit_runs
+
+from app.services import scrape_runs as app_runs
+
+_MODULES = {"kit": kit_runs, "app": app_runs}
+
+
+def _capture_status(mod: Any, counters: dict[str, Any]) -> list[str]:
+    """Прогнать mark_done на фейковой сессии, вернуть статусы всех UPDATE'ов.
+
+    Тот же helper, что в test_2625_run_that_did_nothing.py — читаем СТАТУС В SQL, а не
+    имя вызванной функции.
+    """
+    statuses: list[str] = []
+
+    def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
+        sql = str(stmt)
+        for status in ("done", "failed", "banned"):
+            if f"status = '{status}'" in sql:
+                statuses.append(status)
+        return MagicMock()
+
+    db = MagicMock()
+    db.execute.side_effect = _execute
+    with patch.object(mod, "sentry_sdk", MagicMock()):
+        mod.mark_done(db, 1, dict(counters))
+    return statuses
+
+
+def _capture_backfill_status(
+    counters: dict[str, Any], *, source: str = "avito_detail_backfill", aborted: bool = False
+) -> list[str]:
+    """Прогнать app_runs.mark_backfill_finished на фейковой сессии (mark_done НЕ мокан —
+    в отличие от test_backfill_honest_status.py, здесь важно именно его РЕАЛЬНОЕ
+    поведение: mark_backfill_finished решает вызвать mark_done, а решает ли mark_done
+    остаться 'done' или сам себя переквалифицировать в 'failed' — предмет этого теста).
+
+    mark_backfill_finished есть только в app_runs (kit-копия его не держит — см.
+    docstring модуля runs.py, "mark_skipped есть только здесь" — тот же принцип
+    относится к продуктовым финализаторам detail-backfill'ов).
+    """
+    statuses: list[str] = []
+
+    def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
+        sql = str(stmt)
+        for status in ("done", "failed", "banned"):
+            if f"status = '{status}'" in sql:
+                statuses.append(status)
+        return MagicMock()
+
+    db = MagicMock()
+    db.execute.side_effect = _execute
+    with patch.object(app_runs, "sentry_sdk", MagicMock()):
+        app_runs.mark_backfill_finished(
+            db, 1, dict(counters), source=source, aborted_by_blocks=aborted
+        )
+    return statuses
+
+
+# ── (a) failed_ratio: прод-факт avito_detail_backfill 15.08 ─────────────────────────
+
+
+def test_prod_fact_avito_15_08_no_longer_done() -> None:
+    """{"attempted":64,"failed":57,"enriched":6,"blocked":1} — 89% отказов — 'failed',
+    НЕ 'done'. Красный на старом коде (produced=6 != 0 -> mark_done -> 'done')."""
+    counters = {"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
+    assert _capture_backfill_status(counters) == ["failed"]
+
+
+def test_prod_fact_avito_reason_names_the_ratio() -> None:
+    reason = app_runs._failed_ratio_too_high(
+        {"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
+    )
+    assert reason is not None
+    assert "failed-ratio-honest-status" in reason
+    assert "57 из 64" in reason
+    assert "89%" in reason
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+@pytest.mark.parametrize(
+    ("counters", "flagged", "why"),
+    [
+        ({"attempted": 64, "failed": 57}, True, "прод-факт: 89% отказов"),
+        ({"attempted": 10, "failed": 5}, True, "ровно порог failed (0.5)"),
+        ({"attempted": 20, "failed": 3}, True, "ровно порог degraded (0.15)"),
+        ({"attempted": 20, "failed": 2}, False, "ниже порога degraded (0.10)"),
+        ({"attempted": 2, "failed": 2}, False, "ratio=1.0, но < _FAILED_RATIO_MIN_ATTEMPTS"),
+        ({"attempted": 0, "failed": 0}, False, "нет попыток вовсе"),
+        ({"failed": 5}, False, "нет attempted — чужой словарь"),
+        ({"attempted": 50}, False, "нет failed — чужой словарь"),
+        ({}, False, "пустые counters"),
+        (
+            {"anchors_total": 5, "errors_count": 5, "lots_fetched": 0},
+            False,
+            "sweep-словарь (anchors_total), не detail-backfill",
+        ),
+    ],
+)
+def test_failed_ratio_classifier_boundaries(
+    name: str, counters: dict[str, Any], flagged: bool, why: str
+) -> None:
+    reason = _MODULES[name]._failed_ratio_too_high(counters)
+    assert (reason is not None) is flagged, why
+
+
+# ── (5) не должен палить прогоны с малой/умеренной долей отказов ────────────────────
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_low_failure_ratio_stays_done(name: str) -> None:
+    """Штатный шум (10% отказов) не становится 'failed' — не каждый отказ диагноз."""
+    counters = {"attempted": 50, "enriched": 45, "failed": 5}
+    assert _capture_status(_MODULES[name], counters) == ["done"]
+
+
+def test_tiny_batch_zero_produced_fails_via_old_rule_not_ratio() -> None:
+    """2 попытки, обе отказали, produced=0 — доля тут не при чём (attempted < floor
+    _FAILED_RATIO_MIN_ATTEMPTS, _failed_ratio_too_high вернул бы None); статус всё
+    равно 'failed', но по СТАРОМУ правилу #2674 (produced==0), внутри
+    mark_backfill_finished — mark_done/_failed_ratio_too_high тут не вызываются вовсе.
+    Показывает, что новая проверка не дублирует и не подменяет старую."""
+    counters = {"attempted": 2, "enriched": 0, "failed": 2}
+    assert _capture_backfill_status(counters) == ["failed"]
+
+
+def test_tiny_batch_with_partial_success_stays_done() -> None:
+    """2 попытки, 1 успех, 1 отказ (ratio=0.5, но attempted < floor=3) — стрик слишком
+    короткий, чтобы доля что-то значила -> остаётся 'done'."""
+    counters = {"attempted": 2, "enriched": 1, "failed": 1}
+    assert _capture_backfill_status(counters) == ["done"]
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_honest_empty_sweep_unaffected_by_failed_ratio(name: str) -> None:
+    """Сознательно спящее расписание (город без новостроек): sweep-словарь без
+    attempted/failed вовсе -> failed_ratio не о чем судить, честная пустота остаётся
+    'done' (см. также test_2625_run_that_did_nothing.py::test_honest_empty_stays_done)."""
+    counters = {"anchors_total": 1, "errors_count": 0, "lots_fetched": 0}
+    assert _capture_status(_MODULES[name], counters) == ["done"]
+
+
+# ── (b) _RESULT_COUNTER_KEYS: прод-факт yandex_newbuilding_sweep 26.07-10.08 ─────────
+
+
+def test_prod_fact_yandex_newbuilding_sweep_measured_as_zero() -> None:
+    """processed=5, succeeded=0, rows_inserted=0, failed_resolve=4 — раньше
+    _run_result_count возвращал None ("не измерено"); теперь — измеренный 0 (через
+    'succeeded', не 'rows_inserted' — см. ниже, почему ключ переигран ревью)."""
+    counters = {
+        "total": 309,
+        "fetchable": 200,
+        "pending": 50,
+        "processed": 5,
+        "skipped_already_enriched": 0,
+        "succeeded": 0,
+        "resolved_slug": 1,
+        "failed_resolve": 4,
+        "failed_fetch": 0,
+        "rows_inserted": 0,
+        "duration_sec": 42.0,
+    }
+    assert app_runs._run_result_count(counters) == 0
+    assert kit_runs._run_result_count(counters) == 0
+
+
+def test_succeeded_is_the_measured_key_not_rows_inserted_or_processed() -> None:
+    """'succeeded' читается как результат; голые 'rows_inserted'/'processed' в
+    _RESULT_COUNTER_KEYS больше не участвуют (были в первой версии правки, снято
+    ревью — см. test_rosreestr_dkp_import_healthy_zero_stays_unmeasured и
+    test_newbuilding_enrich_partial_failure_not_masked_by_processed ниже)."""
+    counters = {"processed": 5, "rows_inserted": 0}
+    assert app_runs._run_result_count(counters) is None
+    assert kit_runs._run_result_count(counters) is None
+
+
+def test_rosreestr_dkp_import_healthy_zero_stays_unmeasured() -> None:
+    """Прод-факт rosreestr_dkp_import (2026-08-15, 66 из 67 прогонов за 90д): инкрементальный
+    импорт догнал источник — rows_fetched==rows_skipped, rows_inserted=0. Это ЗДОРОВЫЙ
+    ответ (нечего вставлять), а не отказ; словарь не содержит 'succeeded' вовсе.
+
+    Первая версия правки добавляла голый 'rows_inserted' в _RESULT_COUNTER_KEYS — тогда
+    этот прод-факт читался бы как "измеренный провал" и копил бы практически
+    непрерываемый zero-стрик (rosreestr_dkp_import не прерывается другим статусом:
+    он либо 'done' с этим же нулём, либо не бежал). Ревью поймало это до деплоя —
+    правильный ответ: "не измерено" (None), стрик не копится."""
+    counters = {
+        "last_id": 6829903,
+        "batches_done": 49,
+        "rows_errored": 0,
+        "rows_fetched": 96974,
+        "rows_skipped": 96974,
+        "rows_updated": 0,
+        "rows_inserted": 0,
+    }
+    assert app_runs._run_result_count(counters) is None
+    assert kit_runs._run_result_count(counters) is None
+
+
+def test_newbuilding_enrich_partial_failure_not_masked_by_processed() -> None:
+    """Прод-факт newbuilding_enrich (09.08): processed=25 (счётчик ПОПЫТОК, ==limit),
+    succeeded=14 — 44% отказов. Если бы сторож читал 'processed' как результат, партиальный
+    провал замаскировался бы под measured-25 (сторож нулевого результата промолчал бы
+    ровно там, где должен был сработать при полном провале). 'succeeded' даёт честные 14."""
+    counters = {
+        "failed": 11,
+        "enriched": 14,
+        "attempted": 25,
+        "processed": 25,
+        "succeeded": 14,
+        "failed_fetch": 11,
+    }
+    assert app_runs._run_result_count(counters) == 14
+    assert kit_runs._run_result_count(counters) == 14
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_zero_result_watchdog_now_fires_for_newbuilding_sweep_streak(name: str) -> None:
+    """(b) integration: 3 подряд yandex_newbuilding_sweep-подобных 'done' с succeeded=0
+    -> алерт срабатывает. До фикса _RESULT_COUNTER_KEYS сторож считал результат "не
+    измеренным" и молчал бы вечно (см. #2703 в docstring модуля)."""
+    mod = _MODULES[name]
+    row = MagicMock()
+    row.status = "done"
+    row.counters = {"processed": 5, "succeeded": 0, "rows_inserted": 0, "failed_resolve": 4}
+    db = MagicMock()
+    result = MagicMock()
+    result.fetchall.return_value = [row, row, row]
+    db.execute.return_value = result
+    with patch.object(mod, "sentry_sdk") as mock_sentry:
+        mod._alert_if_consecutive_zero_results(db, "yandex_newbuilding_sweep")
+    mock_sentry.capture_message.assert_called_once()
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_zero_result_watchdog_silent_on_rosreestr_dkp_import_streak(name: str) -> None:
+    """Негативный аналог теста выше: та же лестница из 3 подряд 'done', но словарь
+    rosreestr_dkp_import (нет 'succeeded') -> сторож не считает результат измеренным
+    и НЕ шлёт алерт — регрессионный тест на замечание ревью (HIGH #1)."""
+    mod = _MODULES[name]
+    row = MagicMock()
+    row.status = "done"
+    row.counters = {
+        "last_id": 6829903,
+        "rows_fetched": 96974,
+        "rows_skipped": 96974,
+        "rows_inserted": 0,
+    }
+    db = MagicMock()
+    result = MagicMock()
+    result.fetchall.return_value = [row, row, row]
+    db.execute.return_value = result
+    with patch.object(mod, "sentry_sdk") as mock_sentry:
+        mod._alert_if_consecutive_zero_results(db, "rosreestr_dkp_import")
+    mock_sentry.capture_message.assert_not_called()
+
+
+# ── (c) _column_counts: прод-факт cian_full_load new_count=0 при saved_inserted>0 ───
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_prod_fact_cian_full_load_saved_inserted_surfaces_as_new_count(name: str) -> None:
+    """saved_inserted=482 (прод-факт: три подряд прогона 482/214/239) — new_count
+    больше не 0, хотя ключей 'new_count'/'lots_inserted' в counters нет вовсе."""
+    counters = {"unique_fetched": 1200, "saved_inserted": 482, "saved_updated": 30}
+    total_seen, new_count = _MODULES[name]._column_counts(counters)
+    assert total_seen == 1200
+    assert new_count == 482
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_yandex_newbuilding_rows_inserted_surfaces_as_new_count(name: str) -> None:
+    counters = {"rows_inserted": 7}
+    _, new_count = _MODULES[name]._column_counts(counters)
+    assert new_count == 7
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_new_count_priority_unchanged_by_new_keys(name: str) -> None:
+    """'new_count' явный ключ всё ещё побеждает 'lots_inserted'/'saved_inserted' —
+    расширение списка не меняет приоритет уже существующих ключей."""
+    counters = {"new_count": 5, "lots_inserted": 99, "saved_inserted": 1}
+    _, new_count = _MODULES[name]._column_counts(counters)
+    assert new_count == 5
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_lots_inserted_still_beats_saved_inserted(name: str) -> None:
+    """Порядок пикулярно НЕ переставлен для уже существующей пары — 'lots_inserted'
+    (city/newbuilding-sweep'ы) проверяется раньше 'saved_inserted' (full-load'ы),
+    т.к. это разные, непересекающиеся семейства источников."""
+    counters = {"lots_inserted": 12, "saved_inserted": 999}
+    _, new_count = _MODULES[name]._column_counts(counters)
+    assert new_count == 12

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

@@ -138,6 +138,28 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
 #   unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
 #                    сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
 #                    реально дали ноль.
+#   succeeded      — yandex_newbuilding_sweep (42 прогона/90д) и newbuilding_enrich
+#                    (65 прогонов/90д, единственные два писателя ключа на проде,
+#                    проверено 2026-08-15). НЕ 'rows_inserted': тот ключ пишет ЕЩЁ и
+#                    rosreestr_dkp_import (67 прогонов/90д) — у него rows_inserted=0 в
+#                    66 из 67 это ЗДОРОВЫЙ ответ догнавшего инкрементального импорта
+#                    (rows_fetched=rows_skipped=96974, last_id не двигается неделями),
+#                    а не отказ; если бы 'rows_inserted' попал в этот список, сторож
+#                    зачитывал бы этот здоровый ноль как измеренный провал и копил бы
+#                    практически непрерываемый стрик (rosreestr_dkp_import не
+#                    прерывается другим статусом — импорт либо 'done', либо не бежал).
+#                    НЕ 'processed' по той же причине с другой стороны: это счётчик
+#                    ПОПЫТОК (у newbuilding_enrich processed==attempted==limit даже
+#                    когда succeeded меньше — прод-факт 09.08: processed=25 succeeded=14,
+#                    44% отказов замаскировались бы под measured-25) — сторож нулевого
+#                    результата на нём молчал бы ровно там, где должен сработать, а на
+#                    будущем опустении очереди домов (cian_houses_pending) создал бы
+#                    свой вечный ложный zero-стрик. 'succeeded' у yandex_newbuilding_sweep
+#                    численно совпадает с 'rows_inserted' на всех 42/42 прод-прогонах —
+#                    замена не теряет исходную цель (десять прогонов подряд 26.07-10.08,
+#                    все 'done', succeeded=0 rows_inserted=0 failed_resolve=4-5 — раньше
+#                    ни total_seen/lots_fetched/unique_fetched не было, и
+#                    _run_result_count всегда возвращал None (honest-run-status)).
 # Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
 # прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
 # (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
@@ -145,7 +167,12 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
 # трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
 # (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
 # словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
-_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
+_RESULT_COUNTER_KEYS = (
+    "total_seen",
+    "lots_fetched",
+    "unique_fetched",
+    "succeeded",
+)
 
 
 def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
@@ -277,6 +304,68 @@ def _phase_totally_failed(counters: Mapping[str, Any]) -> str | None:
     return None
 
 
+# honest-run-status (2026-08-15): доля отказов, которая обесценивает формально ненулевой
+# сбор. Прод-факт avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,
+# "blocked":1} — 89% попыток отказали, а mark_backfill_finished всё равно звал mark_done,
+# потому что "produced != 0" (6 обогащено). Ни _sweep_run_did_nothing (нужны
+# anchors_total/errors_count, у backfill'ов их нет), ни _phase_totally_failed (нужна пара
+# "<phase>_attempted"/"<phase>_failed" — здесь голые "attempted"/"failed" без фазового
+# префикса, `"attempted".endswith("_attempted")` не матчит) эту форму counters не ловят —
+# обе проверки написаны под СВОИ формы, а не под backfill'овскую.
+#
+# Порог 'failed' — половина и больше отказов: сбор для практических целей провалился,
+# даже если несколько записей всё же обогатились. Порог 'partial' НЕ заведён отдельным
+# статусом scrape_runs.status — это потребовало бы миграции (DROP+ADD CHECK constraint,
+# 051_scrape_runs_extend.sql) и обучило бы новому значению ещё 4 места (Literal-фильтр
+# admin API, хардкод статусов фронта, оба IN-списка сторожей) — тот же класс "оборванной
+# проводки", из-за которого заведён #2686/ban_kind. Вместо статуса — тот же диагноз, что и
+# у ban_kind: causa в тексте `error`, терминальный статус один ('failed'). 0.15..0.5 —
+# та же 'failed', но с другой формулировкой причины ("деградировал", не "провалился"), чтобы
+# оператор видел разницу читая error, не только status.
+#
+# mark_backfill_finished (единственный писатель "attempted"/"failed" на верхнем уровне
+# counters) живёт только в app.services.scrape_runs — здесь эта проверка сейчас неактивна
+# ни для одного реального вызывающего, но kit-копия держится байт-эквивалентной app-копии
+# (см. docstring модуля), и будущий kit-native job с тем же словарём получит её даром.
+FAILED_RATIO_FAILED_THRESHOLD = 0.5
+FAILED_RATIO_DEGRADED_THRESHOLD = 0.15
+# Минимум попыток, при котором доля вообще что-то значит — иначе 1 отказ из 2 (=0.5)
+# палит статус на шуме единичного случая. То же рассуждение и то же число, что у
+# _PHASE_MIN_ATTEMPTS (см. выше).
+_FAILED_RATIO_MIN_ATTEMPTS = _PHASE_MIN_ATTEMPTS
+
+
+def _failed_ratio_too_high(counters: Mapping[str, Any]) -> str | None:
+    """Прогон, у которого доля отказов слишком велика, даже если что-то собрано.
+
+    Возвращает текст причины (для error) либо None. Читает ГОЛЫЕ ключи "attempted"/
+    "failed" (без фазового префикса) — сейчас это словарь только у четырёх
+    detail-backfill'ов (avito/yandex/domclick/newbuilding_enrich), все идут через
+    mark_backfill_finished → mark_done. `attempted < _FAILED_RATIO_MIN_ATTEMPTS` или
+    отсутствие любого из ключей → None (нечем/не о чём судить — счётчики либо не
+    заполнены, либо принадлежат другому источнику со своим словарём).
+
+    Что признак НЕ доказывает: КТО виноват (площадка, наш прокси, наш парсер) — поэтому
+    'failed' без диагноза, как и у #2625/#2700/#2764.
+    """
+    attempted = _pick_int(counters, "attempted")
+    failed = _pick_int(counters, "failed")
+    if attempted is None or failed is None or attempted < _FAILED_RATIO_MIN_ATTEMPTS:
+        return None
+    ratio = failed / max(attempted, 1)
+    if ratio >= FAILED_RATIO_FAILED_THRESHOLD:
+        verb = "провалился"
+    elif ratio >= FAILED_RATIO_DEGRADED_THRESHOLD:
+        verb = "деградировал"
+    else:
+        return None
+    return (
+        f"failed-ratio-honest-status: сбор {verb} — {failed} из {attempted} попыток "
+        f"отказали (доля {ratio:.0%}); формально ненулевой результат этого не искупает. "
+        f"Причина НЕ установлена — статус 'failed' без диагноза"
+    )
+
+
 def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
 
@@ -287,13 +376,24 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
 
     Приоритет ключей:
-      - total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
-      - new_count  ← 'new_count' (если уже есть) иначе 'lots_inserted'
+      - total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched /
+        succeeded)
+      - new_count  ← 'new_count' / 'lots_inserted' / 'saved_inserted' / 'rows_inserted'
+        (первый присутствующий). 'saved_inserted' — full-load'ы (cian/avito/yandex,
+        CianFullLoadCounters и аналоги в pipeline.py): на проде витрина показывала
+        new_count=0 у трёх подряд cian_full_load при реально сохранённых
+        saved_inserted=482/214/239 (honest-run-status) — ключ 'new_count'/'lots_inserted'
+        у full-load'ов в counters не пишется вовсе. 'rows_inserted' — тот же ключ,
+        которым yandex_newbuilding_sweep и rosreestr_dkp_import сообщают число upsert'ов;
+        здесь (для витринной колонки new_count) это безопасно — в отличие от
+        _RESULT_COUNTER_KEYS этот список не участвует в подсчёте zero-result-стрика.
 
     Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
     тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
     """
-    return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
+    return _run_result_count(counters), _pick_int(
+        counters, "new_count", "lots_inserted", "saved_inserted", "rows_inserted"
+    )
 
 
 def _alert_if_consecutive_failures(db: Session, source: str) -> None:
@@ -632,6 +732,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
     #2700: там же — отказ называть успехом прогон, у которого отказала КАЖДАЯ попытка
     целой фазы (см. _phase_totally_failed). Отличие от #2625: тот случай про «не сделано
     ничего», этот — про «одно направление работы мертво, а суммарный сбор это прячет».
+
+    honest-run-status: там же — отказ называть успехом прогон с высокой долей отказов,
+    даже если собрано > 0 (см. _failed_ratio_too_high). Отличие от #2625/#2700: те два
+    смотрят на «всё или ничего» (все якоря / вся фаза), этот — на ДОЛЮ отказов у
+    detail-backfill'ов, где ни один из первых двух признаков не матчит форму counters.
     """
     did_nothing = _sweep_run_did_nothing(counters)
     if did_nothing is not None:
@@ -643,6 +748,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
         logger.error("%s run_id=%d", phase_dead, run_id)
         mark_failed(db, run_id, phase_dead, counters)
         return
+    ratio_bad = _failed_ratio_too_high(counters)
+    if ratio_bad is not None:
+        logger.error("%s run_id=%d", ratio_bad, run_id)
+        mark_failed(db, run_id, ratio_bad, counters)
+        return
     total_seen, new_count = _column_counts(counters)
     row = db.execute(
         text(