feat(freshness): zero-output check + kn_flats source (#1945)

The data-freshness monitor classified by run RECENCY only, so the domrf_kn
FLATS loader running status=done but extracting 0 flats for ~5 weeks went
undetected — and the kn source watched objects_count (healthy ~1548), not
flats_count (the broken =0 metric).

Add an opt-in zero-output check: an otherwise-fresh run-ledger source (recent
success, would-be fresh by age) that produced 0 work-rows in the 7d window is
downgraded to status="failed" (so scrape_freshness_check alerts), with an
additive "reason". Guards: alert_on_zero_output flag, run-ledger only
(timestamp_col is None), status=="ok" (age-stale/failed already covered), and
upd_7d==0 (SUM of the source's own work_col over done-runs).

Registry: new kn_flats source (kn_scrape_runs, work_col=flats_count, critical,
flag on) — watches the column that was broken; existing kn (objects_count)
unchanged. Flag also enabled on objective (rows_lots, critical). nspd/nspd_geo/
cadastre left unflagged (legitimate-0 / data-table).

JSON additive only (new nullable "reason" key; endpoint is dict[str,Any], no
frontend consumer / no codegen needed). 4 new tests (downgrade, no-false-
positive, age-precedence, registry). code-reviewer APPROVE.

Would have caught #1945 within ~8-14d instead of 5 weeks.

backend/app/api/v1/admin_scrape.py

@@ -1279,6 +1279,13 @@ class FreshnessSource(BaseModel):
     fresh_days: float = 7.0  # < fresh_days → fresh
     stale_days: float = 30.0  # < stale_days → stale, иначе critical
     critical: bool = False  # критичный источник (для Sentry-алёртов в beat-таске)
+    # True → источник, который прогнался успешно И недавно (был бы fresh по возрасту),
+    # НО произвёл 0 work-строк в окне (SUM(work_col) FILTER(status='done') == 0),
+    # ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-ничего-не-извлёк"
+    # (напр. WAF-break / тихий пустой extract: status=done, но flats_count=0). По
+    # умолчанию OFF — включаем только там, где 0 строк = поломка (не legitimate-0,
+    # как nspd/nspd_geo/cadastre). Применимо только к run-ledger (не data-table) режиму.
+    alert_on_zero_output: bool = False
 
 
 # Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на
@@ -1287,7 +1294,10 @@ class FreshnessSource(BaseModel):
 # write-back /analyze-резолвера #1891), а НЕ defunct ledger nspd_scrape_runs (3 WAF-
 # забаненных manual-прогона 2026-04-30, никогда не шедулился → ложный "no successful runs").
 # Пороги:
-#   kn          — beat ~ежедневно (scrape_kn cron) → fresh<2d, stale<5d
+#   kn          — beat ~еженедельно (scrape_kn cron, Mon) → fresh<8d, stale<14d
+#   kn_flats    — та же run-таблица, work_col=flats_count + alert_on_zero_output (issue
+#                 #1945: kn-run status=done но flats_count=0 ~5 недель — монитор молчал,
+#                 т.к. kn смотрит objects_count ~1548, не flats_count) → fresh<8d, stale<14d
 #   objective   — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d
 #   nspd        — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d
 #   nspd_geo    — ручной bulk geo-job → fresh<7d, stale<30d
@@ -1306,6 +1316,22 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [
         stale_days=14.0,
         critical=True,
     ),
+    FreshnessSource(
+        # kn-run пишет ДВА счётчика: objects_count (ЖК, ~1548 — выше) и flats_count
+        # (квартиры). Источник истины для "лоадер прогнался, но ничего не извлёк"
+        # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count=0 ~5 недель,
+        # монитор молчал — objects_count оставался здоровым). Та же run-таблица и пороги,
+        # что у kn, но work_col=flats_count + alert_on_zero_output → 0 квартир за неделю
+        # done-прогонов даунгрейдит otherwise-fresh источник до failed.
+        source="kn_flats",
+        label="DOM.РФ kn-API (квартиры)",
+        table="kn_scrape_runs",
+        work_col="flats_count",
+        fresh_days=8.0,
+        stale_days=14.0,
+        critical=True,
+        alert_on_zero_output=True,
+    ),
     FreshnessSource(
         source="objective",
         label="Объектив (лоты/цены)",
@@ -1314,6 +1340,9 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [
         fresh_days=7.0,
         stale_days=14.0,
         critical=True,
+        # Объектив-sync, прогнавшийся с 0 лотов за неделю — тоже поломка (тихий
+        # пустой extract), а не легитимный 0. Даунгрейд otherwise-fresh до failed.
+        alert_on_zero_output=True,
     ),
     FreshnessSource(
         source="nspd",
@@ -1377,6 +1406,13 @@ def compute_freshness(db: Session) -> dict[str, Any]:
       - age_days         — возраст last_success в днях (NULL если успехов не было)
       - freshness        — fresh / stale / critical / unknown (по порогам источника)
       - status           — ok / stale / failed (для UI-светофора)
+      - reason           — пояснение к zero-output-даунгрейду (иначе None)
+
+    Zero-output (issue #1945): для run-ledger источников с alert_on_zero_output=True
+    источник, который иначе был бы fresh→ok (недавний успешный прогон), но произвёл
+    0 work-строк за 7d окно (objects_updated_7d == 0), ДАУНГРЕЙДИТСЯ до failed — ловит
+    "лоадер-прогнался-но-ничего-не-извлёк" (kn flats_count=0 ~5 недель при status=done).
+    Только когда возраст НЕ покрывает поломку (status=='ok'), чтобы не двойнить.
 
     Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps)
     нет run-ledger семантики (status/started/finished отсутствуют), поэтому:
@@ -1481,6 +1517,9 @@ def compute_freshness(db: Session) -> dict[str, Any]:
 
         freshness = _classify_freshness(age_days, src.fresh_days, src.stale_days)
 
+        upd_24h = int(row["upd_24h"] or 0) if row else 0
+        upd_7d = int(row["upd_7d"] or 0) if row else 0
+
         # status (UI-светофор): нет успехов вообще → failed; иначе мапим freshness.
         if last_success is None:
             status = "failed"
@@ -1491,6 +1530,27 @@ def compute_freshness(db: Session) -> dict[str, Any]:
         else:  # critical
             status = "failed"
 
+        # Zero-output downgrade (issue #1945): источник прогнался успешно И недавно
+        # (был бы fresh→ok по возрасту), НО за окно done-прогонов произвёл 0 work-строк
+        # (objects_updated_7d == SUM(work_col) FILTER(status='done') в 7d == 0) →
+        # ДАУНГРЕЙД до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила.
+        # Только для run-ledger (timestamp_col is None) с alert_on_zero_output=True и
+        # ТОЛЬКО когда источник иначе был бы fresh/ok — по возрасту stale/failed уже
+        # покрыт, не двойним. last_success is None уже отсеян (status=='ok' гарантирует
+        # недавний успех), поэтому это ДОПОЛНИТЕЛЬНЫЙ способ завалить otherwise-fresh.
+        reason: str | None = None
+        if (
+            src.alert_on_zero_output
+            and src.timestamp_col is None
+            and status == "ok"
+            and upd_7d == 0
+        ):
+            status = "failed"
+            reason = (
+                f"zero-output: прогон успешен (age={age_days}d), но {src.work_col}=0 "
+                f"за 7d done-прогонов — лоадер прогнался, но ничего не извлёк (#1945)"
+            )
+
         if src.critical and severity[status] > severity[worst_critical]:
             worst_critical = status
 
@@ -1502,12 +1562,15 @@ def compute_freshness(db: Session) -> dict[str, Any]:
                 "last_success_at": last_success.isoformat() if last_success else None,
                 "last_attempt_at": last_attempt.isoformat() if last_attempt else None,
                 "last_status": last_status,
-                "objects_updated_24h": int(row["upd_24h"] or 0) if row else 0,
+                "objects_updated_24h": upd_24h,
-                "objects_updated_7d": int(row["upd_7d"] or 0) if row else 0,
+                "objects_updated_7d": upd_7d,
                 "age_days": age_days,
                 "thresholds": {"fresh_days": src.fresh_days, "stale_days": src.stale_days},
                 "freshness": freshness,
                 "status": status,
+                # Доп. контекст при zero-output-даунгрейде (None в обычном случае).
+                # Additive — не ломает существующий UI/JSON-контракт.
+                "reason": reason,
             }
         )
 

backend/tests/api/v1/test_admin_scrape_freshness.py

@@ -1,10 +1,11 @@
 """Тесты для GET /api/v1/admin/scrape/freshness (#73 B5-3).
 
-Эндпоинт читает 5 run-таблиц (kn / objective / nspd / nspd_geo / cadastre) и
+Эндпоинт читает 6 источников по 5 run-/data-таблицам (kn / kn_flats / objective /
-для каждой считает свежесть. Здесь DB замокан: на каждый per-source SELECT
+nspd / nspd_geo / cadastre — kn и kn_flats делят kn_scrape_runs, разные work_col)
+и для каждого считает свежесть. Здесь DB замокан: на каждый per-source SELECT
 возвращаем заранее заданную строку, чтобы проверить классификацию
-fresh/stale/critical, маппинг status (ok/stale/failed) и агрегат overall_status
+fresh/stale/critical, маппинг status (ok/stale/failed), zero-output downgrade
-по критичным источникам.
+(#1945) и агрегат overall_status по критичным источникам.
 """
 
 from __future__ import annotations
@@ -108,7 +109,8 @@ def test_freshness_mixed_statuses() -> None:
     assert resp.status_code == 200, resp.text
     body = resp.json()
     assert "generated_at" in body
-    assert len(body["sources"]) == 5
+    # kn / kn_flats / objective / nspd / nspd_geo / cadastre.
+    assert len(body["sources"]) == len(admin_scrape._FRESHNESS_SOURCES)
     by_src = {s["source"]: s for s in body["sources"]}
 
     assert by_src["kn"]["status"] == "ok"
@@ -279,6 +281,162 @@ def test_classify_freshness_boundaries() -> None:
     assert f(100.0, 2.0, 5.0) == "critical"
 
 
+def test_zero_output_downgrades_otherwise_fresh_source() -> None:
+    """issue #1945: kn_flats прогнался успешно И недавно (был бы fresh→ok по возрасту),
+    но flats_count=0 за 7d done-прогонов → DOWNGRADE до failed (alertable).
+
+    Та же run-таблица kn_scrape_runs у kn (alert_on_zero_output=False) и kn_flats
+    (True). Возвращаем строку с upd_7d=0 → kn остаётся ok (нет флага = нет ложного
+    срабатывания), kn_flats падает в failed (zero-output) с пояснением в reason.
+    """
+    fresh_zero = {
+        "last_success_at": _ago(1.0),  # < fresh_days=8 → был бы fresh
+        "last_attempt_at": _ago(1.0),
+        "upd_24h": 0,
+        "upd_7d": 0,  # 0 work-строк за окно done-прогонов → zero-output
+        "last_status": "done",
+    }
+    # Остальные источники свежие+ненулевые, чтобы изолировать поведение kn/kn_flats.
+    nonzero_fresh = {
+        "last_success_at": _ago(0.5),
+        "last_attempt_at": _ago(0.5),
+        "upd_24h": 10,
+        "upd_7d": 100,
+        "last_status": "done",
+    }
+    rows: dict[str, dict[str, Any]] = {}
+    for s in admin_scrape._FRESHNESS_SOURCES:
+        if s.timestamp_col is not None:
+            rows[s.table] = {**nonzero_fresh, "last_status": None}
+        elif s.table == "kn_scrape_runs":
+            rows[s.table] = dict(fresh_zero)
+        else:
+            rows[s.table] = dict(nonzero_fresh)
+    _override_db(_make_db(rows))
+    try:
+        resp = TestClient(app).get(ENDPOINT)
+    finally:
+        app.dependency_overrides.clear()
+
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    by_src = {s["source"]: s for s in body["sources"]}
+
+    # kn_flats: otherwise-fresh, но flats_count=0 → DOWNGRADE до failed (alertable).
+    assert by_src["kn_flats"]["freshness"] == "fresh"  # по возрасту был бы fresh
+    assert by_src["kn_flats"]["status"] == "failed"  # zero-output downgrade
+    assert by_src["kn_flats"]["objects_updated_7d"] == 0
+    assert by_src["kn_flats"]["reason"] is not None
+    assert "flats_count" in by_src["kn_flats"]["reason"]
+
+    # kn — та же таблица, upd_7d=0, но БЕЗ флага → остаётся ok (no false positive).
+    assert by_src["kn"]["status"] == "ok"
+    assert by_src["kn"]["reason"] is None
+
+    # critical kn_flats упал → overall_status=failed.
+    assert by_src["kn_flats"]["critical"] is True
+    assert body["overall_status"] == "failed"
+
+
+def test_zero_output_no_false_positive_when_nonzero() -> None:
+    """alert_on_zero_output=True источник с НЕнулевым выходом → остаётся ok (нет даунгрейда).
+
+    Регрессия: zero-output даунгрейд НЕ должен бить по здоровому источнику, который
+    прогнался и реально извлёк строки.
+    """
+    nonzero_fresh = {
+        "last_success_at": _ago(1.0),
+        "last_attempt_at": _ago(1.0),
+        "upd_24h": 12,
+        "upd_7d": 1234,  # ненулевой выход
+        "last_status": "done",
+    }
+    rows: dict[str, dict[str, Any]] = {}
+    for s in admin_scrape._FRESHNESS_SOURCES:
+        rows[s.table] = (
+            {**nonzero_fresh, "last_status": None}
+            if s.timestamp_col is not None
+            else dict(nonzero_fresh)
+        )
+    _override_db(_make_db(rows))
+    try:
+        resp = TestClient(app).get(ENDPOINT)
+    finally:
+        app.dependency_overrides.clear()
+
+    assert resp.status_code == 200, resp.text
+    by_src = {s["source"]: s for s in resp.json()["sources"]}
+    # Все alert_on_zero_output-источники с ненулевым выходом → ok, reason=None.
+    for s in admin_scrape._FRESHNESS_SOURCES:
+        if s.alert_on_zero_output:
+            assert by_src[s.source]["status"] == "ok", s.source
+            assert by_src[s.source]["reason"] is None, s.source
+
+
+def test_zero_output_not_applied_when_already_stale() -> None:
+    """Если источник уже stale/failed по возрасту — zero-output НЕ передоунгрейдит
+    (age уже покрывает), и не двойнит. objective: старый успех (10d → stale) + upd_7d=0
+    → остаётся stale (НЕ перепрыгивает в failed через zero-output)."""
+    base_fresh = {
+        "last_success_at": _ago(0.5),
+        "last_attempt_at": _ago(0.5),
+        "upd_24h": 5,
+        "upd_7d": 50,
+        "last_status": "done",
+    }
+    rows: dict[str, dict[str, Any]] = {}
+    for s in admin_scrape._FRESHNESS_SOURCES:
+        rows[s.table] = (
+            {**base_fresh, "last_status": None}
+            if s.timestamp_col is not None
+            else dict(base_fresh)
+        )
+    # objective: stale по возрасту (10d: fresh<7, stale<14) + 0 выход.
+    rows["objective_scrape_runs"] = {
+        "last_success_at": _ago(10.0),
+        "last_attempt_at": _ago(10.0),
+        "upd_24h": 0,
+        "upd_7d": 0,
+        "last_status": "done",
+    }
+    _override_db(_make_db(rows))
+    try:
+        resp = TestClient(app).get(ENDPOINT)
+    finally:
+        app.dependency_overrides.clear()
+
+    assert resp.status_code == 200, resp.text
+    by_src = {s["source"]: s for s in resp.json()["sources"]}
+    # zero-output применяется ТОЛЬКО к otherwise-ok; stale остаётся stale, reason=None.
+    assert by_src["objective"]["status"] == "stale"
+    assert by_src["objective"]["reason"] is None
+
+
+def test_kn_flats_registry_config() -> None:
+    """kn_flats: та же run-таблица что у kn, но work_col=flats_count + alert_on_zero_output,
+    критичный, недельные пороги. kn (objects_count) остаётся без флага (unchanged)."""
+    kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats")
+    assert kn_flats.table == "kn_scrape_runs"
+    assert kn_flats.work_col == "flats_count"
+    assert kn_flats.timestamp_col is None  # run-ledger режим
+    assert kn_flats.alert_on_zero_output is True
+    assert kn_flats.critical is True
+    assert kn_flats.fresh_days == 8.0
+    assert kn_flats.stale_days == 14.0
+
+    kn = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn")
+    assert kn.work_col == "objects_count"
+    assert kn.alert_on_zero_output is False  # существующий источник unchanged
+
+    objective = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "objective")
+    assert objective.alert_on_zero_output is True
+
+    # Legitimate-0 источники НЕ должны иметь флаг (избегаем ложных срабатываний).
+    for name in ("nspd", "nspd_geo", "cadastre"):
+        src = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == name)
+        assert src.alert_on_zero_output is False, name
+
+
 def test_kn_thresholds_match_weekly_cadence() -> None:
     """kn шедулится еженедельно (Mon cron) → пороги должны покрывать полную неделю,
     иначе ложный 'stale' каждую среду-воскресенье. Регрессия на калибровку."""