fix(tradein/scraper): сторож «ноль результатов» перестаёт читать «не измерено» как ноль (#2703)

Сторож смотрел колонку scrape_runs.total_seen, у которой DEFAULT 0. Замер на
проде (53 источника): колонку заполняет только lots_fetched — 21 источник, 455
прогонов. Ещё 4 источника (133 прогона, full-load'ы avito/cian/yandex) пишут
unique_fetched, которого сторож не знал, хотя у cian_full_load 6 из 38 успешных
прогонов дали настоящий ноль. Оставшиеся 28 источников (2650 прогонов) не пишут
ни одного результатного ключа вовсе: proxy_healthcheck 1650, три detail-
backfill'а, все deactivate_stale_*, refresh_search_matview с буквально пустыми
{} во всех 55 строках.

Для этих 28 сторож читал 0 ВСЕГДА, в том числе у полностью успешного прогона →
стрик не прерывался никогда → анти-спам «один раз на стрик» превращался в «один
раз навсегда»: сработав однажды на третьем прогоне источника, сторож замолкал
навсегда, продолжая выглядеть настроенным.

Свести счётчики к словарю сторожа невозможно: у 28 источников 28 разных
словарей (deactivated / rows_written / poi_loaded / snapshotted / upserted /
listings_matched …), у трёх мониторов результата нет по смыслу, а ноль у
deactivate_stale_* — здоровый ответ. Поэтому:

- признак берётся из counters, а не из колонки: «не измерено» (None) отличимо
  от «измерено, ноль»;
- прогон без результатной метрики стрик ПРЕРЫВАЕТ — недостижимое условие
  прерывания стало невозможным по построению, а не по корректности счётчиков;
- слепая зона логируется явно (раз на источник за процесс), а не молчит;
- в словарь добавлен unique_fetched — 4 full-load'а входят в зону видимости
  сторожа и перестают показывать total_seen=0 в админке.

Доставка уведомлений этим НЕ чинится: получатель в GlitchTip не настроен (#2673),
за всю историю наружу не ушло ни одного события.

Refs #2703

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

@@ -8,7 +8,8 @@ from __future__ import annotations
 
 import json
 import logging
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
+from functools import cache
 from typing import Any
 
 import sentry_sdk
@@ -46,6 +47,61 @@ BAN_KIND_PLATFORM = "platform"  # площадка показала firewall/403
 BAN_KIND_INFRA = "infra"  # наш сайдкар/прокси не отдал страницу — внутреннее
 
 
+def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
+    """Первое присутствующее из ``keys`` как int; None — ни одного ключа нет."""
+    for key in keys:
+        val = counters.get(key)
+        if val is not None:
+            try:
+                return int(val)
+            except (TypeError, ValueError):
+                return None
+    return None
+
+
+# #2703: ключи, которыми задача сообщает СВОЙ бизнес-результат. Список намеренно
+# короткий и состоит из синонимов ОДНОЙ величины — «сколько объявлений отдала выдача»:
+#   total_seen     — если задача посчитала сама;
+#   lots_fetched   — все city/newbuilding-sweep'ы (21 источник, 455 прогонов на проде);
+#   unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
+#                    сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
+#                    реально дали ноль.
+# Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
+# прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
+# (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
+# …), а у refresh_search_matview counters пусты буквально ({} во всех 55 строках) и у
+# трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
+# (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
+# словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
+_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
+
+
+def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
+    """Бизнес-результат прогона; **None = прогон его не сообщил** (≠ ноль).
+
+    Ровно это различие и было потеряно: сторож читал колонку ``total_seen``, у
+    которой DEFAULT 0, поэтому «не измерено» и «измерено, ноль» выглядели одинаково.
+    """
+    return _pick_int(counters or {}, *_RESULT_COUNTER_KEYS)
+
+
+@cache
+def _warn_source_has_no_result_metric(source: str, keys: tuple[str, ...]) -> None:
+    """Один раз на процесс: у источника нет ключа, по которому сторож судит (#2703).
+
+    Не алерт — алертить не о чем, судить не о чем тоже. Это делает слепую зону
+    ВИДИМОЙ: раньше её признаком был вечно молчащий сторож, выглядящий настроенным.
+    """
+    logger.warning(
+        "zero-result watchdog неприменим к source=%s: counters не содержат ни одного "
+        "результатного ключа %s (есть: %s) — прогоны этого источника больше не считаются "
+        "нулевыми по умолчанию (#2703)",
+        source,
+        _RESULT_COUNTER_KEYS,
+        ", ".join(keys) or "<пусто>",
+    )
+
+
 def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
 
@@ -56,24 +112,13 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
 
     Приоритет ключей:
-      - total_seen ← 'total_seen' (если уже есть в counters) иначе 'lots_fetched'
-      - new_count  ← 'new_count'  (если уже есть)            иначе 'lots_inserted'
+      - total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
+      - new_count  ← 'new_count' (если уже есть) иначе 'lots_inserted'
 
     Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
     тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
     """
-
-    def _pick(*keys: str) -> int | None:
-        for key in keys:
-            val = counters.get(key)
-            if val is not None:
-                try:
-                    return int(val)
-                except (TypeError, ValueError):
-                    return None
-        return None
-
-    return _pick("total_seen", "lots_fetched"), _pick("new_count", "lots_inserted")
+    return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
 
 
 def _alert_if_consecutive_failures(db: Session, source: str) -> None:
@@ -133,7 +178,7 @@ def _alert_if_consecutive_failures(db: Session, source: str) -> None:
 
 def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
     """Отправить Sentry alert если последние CONSECUTIVE_ZERO_RESULT_ALERT_THRESHOLD
-    завершённых 'done' запусков для source имеют total_seen=0 (#2625).
+    завершённых 'done' запусков для source дали ИЗМЕРЕННЫЙ нулевой результат (#2625).
 
     Отличается от _alert_if_consecutive_failures: статус здесь формально 'done'
     (errors_count=0) — деградация невидима существующему failed/banned алерту.
@@ -143,6 +188,14 @@ def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
     Anti-spam: тот же N-й-стрик паттерн, что у _alert_if_consecutive_failures —
     алерт срабатывает ровно когда стрик достигает порога, не на каждом запуске сверх.
 
+    #2703: анти-спам «один раз на стрик» безопасен ТОЛЬКО там, где стрик может
+    прерваться. Сторож читал колонку total_seen (DEFAULT 0), которой у 28 из 53
+    источников не заполняет ничто — значит у них он читал 0 ВСЕГДА, в том числе у
+    полностью успешного прогона, стрик не прерывался никогда, и после первого
+    события сторож замолкал навсегда, продолжая выглядеть настроенным. Теперь
+    признак берётся из counters, а «не измерено» (None) стрик ПРЕРЫВАЕТ — ложный
+    вечный стрик стал невозможен по построению, а слепая зона логируется явно.
+
     Best-effort: весь блок обёрнут в try/except — сбой запроса или неинициализированный
     Sentry НЕ должен нарушать вызывающий mark_done путь.
     """
@@ -150,11 +203,13 @@ def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
     try:
         # Те же non-running статусы, что у _alert_if_consecutive_failures — стрик
         # 'done'-с-нулём прерывается ЛЮБЫМ другим завершением (failed/banned/done-
-        # с-результатом/cancelled), не только успешным сбором.
+        # с-результатом/cancelled/прогон без результатной метрики), не только успешным
+        # сбором. counters, а НЕ колонка total_seen: у колонки DEFAULT 0, по ней
+        # «не измерено» неотличимо от «ноль» (#2703).
         rows = db.execute(
             text(
                 """
-                SELECT status, total_seen FROM scrape_runs
+                SELECT status, counters FROM scrape_runs
                 WHERE source = :source
                   AND status IN ('failed', 'banned', 'done', 'cancelled')
                 ORDER BY finished_at DESC NULLS LAST
@@ -168,7 +223,19 @@ def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
             return
 
         def _is_zero_done(r: Any) -> bool:
-            return r.status == "done" and (r.total_seen or 0) == 0
+            """Только ИЗМЕРЕННЫЙ ноль. Прогон без результатной метрики стрик ПРЕРЫВАЕТ.
+
+            Так недостижимое условие прерывания невозможно по построению: источник,
+            чей словарь счётчиков сторожу неизвестен, не копит ложный стрик и не
+            запирает анти-спам «один раз на стрик» в «один раз навсегда».
+            """
+            return r.status == "done" and _run_result_count(r.counters) == 0
+
+        if _run_result_count(rows[0].counters) is None:
+            # Свежайший завершённый прогон не сообщил результата — судить нечем.
+            # Логируем (один раз на источник за процесс) вместо молчаливого нуля.
+            _warn_source_has_no_result_metric(source, tuple(sorted(rows[0].counters or {})))
+            return
 
         first_n = rows[:n]
         if not all(_is_zero_done(r) for r in first_n):

tradein-mvp/backend/tests/test_2703_zero_result_watchdog.py

@@ -0,0 +1,169 @@
+"""#2703: сторож «ноль результатов» перестаёт читать «не измерено» как ноль.
+
+`_alert_if_consecutive_zero_results` смотрел колонку `scrape_runs.total_seen`, у
+которой DEFAULT 0. Замер на проде (2026-08-06, 53 источника):
+
+  * 21 источник / 455 прогонов — колонку заполняет `lots_fetched` (все city-sweep'ы);
+  * 4 источника / 133 прогона — пишут `unique_fetched` (full-load'ы avito/cian/yandex),
+    сторож их НЕ видел, хотя у `cian_full_load` 6 из 38 успешных прогонов дали ноль;
+  * 28 источников / 2650 прогонов — не пишут НИ ОДНОГО результатного ключа
+    (`proxy_healthcheck` 1650, три detail-backfill'а, `deactivate_stale_*`,
+    `refresh_search_matview` с буквально пустыми `{}` во всех 55 строках …).
+
+Для последних сторож читал 0 ВСЕГДА, в том числе у полностью успешного прогона:
+стрик не прерывался никогда, а анти-спам «один раз на стрик» превращался в «один
+раз навсегда». Ноль записей `total_seen` в counters на всей истории — проверено.
+
+Правка: признак берётся из counters, «не измерено» (None) стрик ПРЕРЫВАЕТ.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+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 _row(status: str, counters: dict[str, Any] | None) -> SimpleNamespace:
+    """Строка scrape_runs.
+
+    `total_seen` кладём НАРОЧНО: старый сторож читал именно её, поэтому на строках
+    без результатного ключа он видел ноль и копил ложный стрик. Тесты ниже
+    фальсифицируются ровно этим — на старом коде они дают другой ответ.
+    """
+    return SimpleNamespace(status=status, counters=counters, total_seen=0)
+
+
+def _db(rows: list[SimpleNamespace]) -> MagicMock:
+    db = MagicMock()
+    db.execute.return_value.fetchall.return_value = rows
+    return db
+
+
+def _run_watchdog(mod: Any, rows: list[SimpleNamespace]) -> MagicMock:
+    """Прогнать сторож на заданных строках, вернуть мок sentry_sdk."""
+    sentry = MagicMock()
+    with patch.object(mod, "sentry_sdk", sentry):
+        mod._alert_if_consecutive_zero_results(_db(rows), "src")
+    return sentry
+
+
+# ── 1. Различие «не измерено» и «измеренный ноль» ────────────────────────────
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+@pytest.mark.parametrize(
+    ("counters", "expected"),
+    [
+        ({"lots_fetched": 0}, 0),
+        ({"lots_fetched": 12}, 12),
+        ({"total_seen": 7}, 7),
+        ({"unique_fetched": 0}, 0),  # full-load'ы — раньше сторож их не видел
+        ({"unique_fetched": 340}, 340),
+        ({"attempted": 1600, "enriched": 0, "failed": 1600}, None),  # backfill
+        ({}, None),  # refresh_search_matview
+        (None, None),
+    ],
+)
+def test_run_result_count(name: str, counters: dict[str, Any] | None, expected: int | None) -> None:
+    """1600 попыток без результатного ключа — это None, а не ноль и не 1600."""
+    assert _MODULES[name]._run_result_count(counters) == expected
+
+
+# ── 2. Слепая зона больше не копит ложный стрик ──────────────────────────────
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_source_without_result_metric_never_alerts(name: str, caplog: Any) -> None:
+    """Backfill: три успешных прогона подряд, результатной метрики нет → тишина + лог.
+
+    Фальсификация: старый сторож читал total_seen=0 у всех трёх → отправлял алерт
+    и запирал анти-спам навсегда. Здесь sentry не вызывается вовсе, а слепая зона
+    названа вслух в логе (раньше её признаком был вечно молчащий сторож).
+    """
+    mod = _MODULES[name]
+    mod._warn_source_has_no_result_metric.cache_clear()
+    rows = [_row("done", {"attempted": 1600, "enriched": 0}) for _ in range(4)]
+
+    with caplog.at_level(logging.WARNING):
+        sentry = _run_watchdog(mod, rows)
+
+    sentry.capture_message.assert_not_called()
+    assert "zero-result watchdog неприменим" in caplog.text
+    assert "attempted, enriched" in caplog.text
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_unmeasured_run_breaks_the_streak(name: str) -> None:
+    """Два измеренных нуля + прогон без метрики → стрик прерван, алерта нет.
+
+    Фальсификация: на старом коде все три строки имели total_seen=0 → стрик
+    считался полным и алерт уходил. Это и есть «условие прерывания недостижимо»:
+    теперь оно достижимо по построению — любой немеряный прогон его выполняет.
+    """
+    mod = _MODULES[name]
+    mod._warn_source_has_no_result_metric.cache_clear()
+    rows = [
+        _row("done", {"lots_fetched": 0}),
+        _row("done", {"lots_fetched": 0}),
+        _row("done", {"deactivated": 3}),  # чужой словарь → не измерено
+        _row("done", {"lots_fetched": 5}),
+    ]
+
+    assert _run_watchdog(mod, rows).capture_message.call_count == 0
+
+
+# ── 3. Настоящая деградация по-прежнему видна ────────────────────────────────
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+@pytest.mark.parametrize("key", ["lots_fetched", "unique_fetched"])
+def test_real_zero_streak_still_alerts(name: str, key: str) -> None:
+    """Три измеренных нуля подряд после нормального прогона → алерт.
+
+    `unique_fetched` (full-load'ы) добавлен этой правкой: раньше эти 4 источника
+    (133 прогона) сторож не видел вовсе. Фальсификация для обоих ключей: старый
+    сторож читал r.total_seen, которого в SELECT больше нет — на новых строках он
+    падал бы в свой best-effort except и молчал.
+    """
+    mod = _MODULES[name]
+    mod._warn_source_has_no_result_metric.cache_clear()
+    rows = [_row("done", {key: 0}) for _ in range(3)] + [_row("done", {key: 42})]
+
+    assert _run_watchdog(mod, rows).capture_message.call_count == 1
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_anti_spam_still_holds_within_a_streak(name: str) -> None:
+    """Четвёртый ноль подряд повторного алерта не даёт — анти-спам не сломан."""
+    mod = _MODULES[name]
+    mod._warn_source_has_no_result_metric.cache_clear()
+    rows = [_row("done", {"lots_fetched": 0}) for _ in range(4)]
+
+    assert _run_watchdog(mod, rows).capture_message.call_count == 0
+
+
+# ── 4. Колонка total_seen тоже начинает видеть full-load'ы ───────────────────
+
+
+@pytest.mark.parametrize("name", list(_MODULES))
+def test_column_counts_picks_unique_fetched(name: str) -> None:
+    """`_column_counts` разделяет словарь со сторожем — одна функция, два потребителя.
+
+    Побочная польза: admin/observability перестаёт показывать total_seen=0 у
+    full-load'ов (avito 70 прогонов, cian 52, yandex 4, exhaustive 7).
+    """
+    assert _MODULES[name]._column_counts({"unique_fetched": 340})[0] == 340
+    assert _MODULES[name]._column_counts({"attempted": 5})[0] is None

tradein-mvp/backend/tests/test_scrape_run_alert.py

@@ -213,21 +213,28 @@ class TestMarkBannedAlertIntegration:
 # ---------------------------------------------------------------------------
 
 
+# #2703: сторож читает counters, а НЕ колонку total_seen — у той DEFAULT 0, по ней
+# «прогон не сообщил результата» неотличимо от «сообщил ноль». total_seen оставлен в
+# фикстурах как исторический контекст: именно его читал прежний сторож.
+
+
 def _zero_row() -> SimpleNamespace:
-    """Успешно завершённый прогон ('done'), но 0 лотов — капча/пустая выдача-под-
-    видом-успеха (#2625)."""
-    return SimpleNamespace(status="done", total_seen=0)
+    """Успешно завершённый прогон ('done') с ИЗМЕРЕННЫМ нулём лотов — капча/пустая
+    выдача-под-видом-успеха (#2625)."""
+    return SimpleNamespace(status="done", counters={"lots_fetched": 0}, total_seen=0)
 
 
 def _nonzero_row(total_seen: int = 50) -> SimpleNamespace:
     """Успешно завершённый прогон с реальным результатом — прерывает "нулевой" стрик."""
-    return SimpleNamespace(status="done", total_seen=total_seen)
+    return SimpleNamespace(
+        status="done", counters={"lots_fetched": total_seen}, total_seen=total_seen
+    )
 
 
 def _other_status_row(status: str) -> SimpleNamespace:
     """failed/banned/cancelled — НЕ 'done', прерывает "нулевой" стрик (уже покрыт
     _alert_if_consecutive_failures отдельно)."""
-    return SimpleNamespace(status=status, total_seen=0)
+    return SimpleNamespace(status=status, counters={"lots_fetched": 0}, total_seen=0)
 
 
 class TestAlertIfConsecutiveZeroResults:
@@ -301,7 +308,7 @@ def _make_db_for_mark_done(source: str, streak_rows: list[SimpleNamespace]) -> M
     Sequence of execute() calls in mark_done:
       1. UPDATE scrape_runs SET status='done' ... RETURNING id  → .first()
       2. (_alert_on_run_id) SELECT source FROM scrape_runs WHERE id=... → .fetchone()
-      3. (_alert_if_consecutive_zero_results) SELECT status, total_seen ... → .fetchall()
+      3. (_alert_if_consecutive_zero_results) SELECT status, counters ... → .fetchall()
     """
     db = MagicMock()
 

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

@@ -15,7 +15,8 @@ from __future__ import annotations
 
 import json
 import logging
-from collections.abc import Callable
+from functools import lru_cache
+from collections.abc import Callable, Mapping
 from typing import Any
 
 from sqlalchemy import text
@@ -57,6 +58,61 @@ BAN_KIND_PLATFORM = "platform"  # площадка показала firewall/403
 BAN_KIND_INFRA = "infra"  # наш сайдкар/прокси не отдал страницу — внутреннее
 
 
+def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
+    """Первое присутствующее из ``keys`` как int; None — ни одного ключа нет."""
+    for key in keys:
+        val = counters.get(key)
+        if val is not None:
+            try:
+                return int(val)
+            except (TypeError, ValueError):
+                return None
+    return None
+
+
+# #2703: ключи, которыми задача сообщает СВОЙ бизнес-результат. Список намеренно
+# короткий и состоит из синонимов ОДНОЙ величины — «сколько объявлений отдала выдача»:
+#   total_seen     — если задача посчитала сама;
+#   lots_fetched   — все city/newbuilding-sweep'ы (21 источник, 455 прогонов на проде);
+#   unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
+#                    сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
+#                    реально дали ноль.
+# Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
+# прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
+# (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
+# …), а у refresh_search_matview counters пусты буквально ({} во всех 55 строках) и у
+# трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
+# (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
+# словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
+_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
+
+
+def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
+    """Бизнес-результат прогона; **None = прогон его не сообщил** (≠ ноль).
+
+    Ровно это различие и было потеряно: сторож читал колонку ``total_seen``, у
+    которой DEFAULT 0, поэтому «не измерено» и «измерено, ноль» выглядели одинаково.
+    """
+    return _pick_int(counters or {}, *_RESULT_COUNTER_KEYS)
+
+
+@lru_cache(maxsize=None)
+def _warn_source_has_no_result_metric(source: str, keys: tuple[str, ...]) -> None:
+    """Один раз на процесс: у источника нет ключа, по которому сторож судит (#2703).
+
+    Не алерт — алертить не о чем, судить не о чем тоже. Это делает слепую зону
+    ВИДИМОЙ: раньше её признаком был вечно молчащий сторож, выглядящий настроенным.
+    """
+    logger.warning(
+        "zero-result watchdog неприменим к source=%s: counters не содержат ни одного "
+        "результатного ключа %s (есть: %s) — прогоны этого источника больше не считаются "
+        "нулевыми по умолчанию (#2703)",
+        source,
+        _RESULT_COUNTER_KEYS,
+        ", ".join(keys) or "<пусто>",
+    )
+
+
 def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
 
@@ -67,24 +123,13 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
     показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
 
     Приоритет ключей:
-      - total_seen ← 'total_seen' (если уже есть в counters) иначе 'lots_fetched'
-      - new_count  ← 'new_count'  (если уже есть)            иначе 'lots_inserted'
+      - total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
+      - new_count  ← 'new_count' (если уже есть) иначе 'lots_inserted'
 
     Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
     тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
     """
-
-    def _pick(*keys: str) -> int | None:
-        for key in keys:
-            val = counters.get(key)
-            if val is not None:
-                try:
-                    return int(val)
-                except (TypeError, ValueError):
-                    return None
-        return None
-
-    return _pick("total_seen", "lots_fetched"), _pick("new_count", "lots_inserted")
+    return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
 
 
 def _alert_if_consecutive_failures(db: Session, source: str) -> None:
@@ -146,7 +191,7 @@ def _alert_if_consecutive_failures(db: Session, source: str) -> None:
 
 def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
     """Отправить Sentry alert если последние CONSECUTIVE_ZERO_RESULT_ALERT_THRESHOLD
-    завершённых 'done' запусков для source имеют total_seen=0 (#2625).
+    завершённых 'done' запусков для source дали ИЗМЕРЕННЫЙ нулевой результат (#2625).
 
     Отличается от _alert_if_consecutive_failures: статус здесь формально 'done'
     (errors_count=0) — деградация невидима существующему failed/banned алерту.
@@ -156,6 +201,14 @@ def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
     Anti-spam: тот же N-й-стрик паттерн, что у _alert_if_consecutive_failures —
     алерт срабатывает ровно когда стрик достигает порога, не на каждом запуске сверх.
 
+    #2703: анти-спам «один раз на стрик» безопасен ТОЛЬКО там, где стрик может
+    прерваться. Сторож читал колонку total_seen (DEFAULT 0), которой у 28 из 53
+    источников не заполняет ничто — значит у них он читал 0 ВСЕГДА, в том числе у
+    полностью успешного прогона, стрик не прерывался никогда, и после первого
+    события сторож замолкал навсегда, продолжая выглядеть настроенным. Теперь
+    признак берётся из counters, а «не измерено» (None) стрик ПРЕРЫВАЕТ — ложный
+    вечный стрик стал невозможен по построению, а слепая зона логируется явно.
+
     Best-effort: весь блок обёрнут в try/except — сбой запроса или неинициализированный
     Sentry НЕ должен нарушать вызывающий mark_done путь.
     """
@@ -165,11 +218,13 @@ def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
     try:
         # Те же non-running статусы, что у _alert_if_consecutive_failures — стрик
         # 'done'-с-нулём прерывается ЛЮБЫМ другим завершением (failed/banned/done-
-        # с-результатом/cancelled), не только успешным сбором.
+        # с-результатом/cancelled/прогон без результатной метрики), не только успешным
+        # сбором. counters, а НЕ колонка total_seen: у колонки DEFAULT 0, по ней
+        # «не измерено» неотличимо от «ноль» (#2703).
         rows = db.execute(
             text(
                 """
-                SELECT status, total_seen FROM scrape_runs
+                SELECT status, counters FROM scrape_runs
                 WHERE source = :source
                   AND status IN ('failed', 'banned', 'done', 'cancelled')
                 ORDER BY finished_at DESC NULLS LAST
@@ -183,7 +238,19 @@ def _alert_if_consecutive_zero_results(db: Session, source: str) -> None:
             return
 
         def _is_zero_done(r: Any) -> bool:
-            return r.status == "done" and (r.total_seen or 0) == 0
+            """Только ИЗМЕРЕННЫЙ ноль. Прогон без результатной метрики стрик ПРЕРЫВАЕТ.
+
+            Так недостижимое условие прерывания невозможно по построению: источник,
+            чей словарь счётчиков сторожу неизвестен, не копит ложный стрик и не
+            запирает анти-спам «один раз на стрик» в «один раз навсегда».
+            """
+            return r.status == "done" and _run_result_count(r.counters) == 0
+
+        if _run_result_count(rows[0].counters) is None:
+            # Свежайший завершённый прогон не сообщил результата — судить нечем.
+            # Логируем (один раз на источник за процесс) вместо молчаливого нуля.
+            _warn_source_has_no_result_metric(source, tuple(sorted(rows[0].counters or {})))
+            return
 
         first_n = rows[:n]
         if not all(_is_zero_done(r) for r in first_n):