fix(freshness): repoint nspd source to live nspd_quarter_dumps (data-table mode)

The freshness monitor watched defunct nspd_scrape_runs (3 WAF-banned manual
runs 2026-04-30, triggered_by=manual, never scheduled) → false "no successful
runs" alert. Investigation (surfaced by the monitor's first prod run) showed
NSPD is healthy: the live harvest (tasks.nspd_sync.harvest_stale_quarters +
the #1891 /analyze resolver write-back) writes nspd_quarter_dumps directly —
370 rows, 56 in the last 7 days, last fetched 2026-06-22 — but never writes
nspd_scrape_runs.

Add a data-table freshness mode (FreshnessSource.timestamp_col): when set, the
source uses MAX(timestamp_col) instead of run-ledger status semantics. Repoint
nspd → nspd_quarter_dumps.fetched_at_utc (fresh<14d, stale<30d). Run-ledger
sources (kn/objective/nspd_geo/cadastre) untouched. No schema/codegen change.

Refs #73

backend/app/api/v1/admin_scrape.py

@@ -1264,24 +1264,32 @@ class FreshnessSource(BaseModel):
 
     source: str  # стабильный ключ (kn / nspd / objective / nspd_geo / cadastre)
     label: str  # человекочитаемое имя для UI
-    table: str  # run-таблица
+    table: str  # run-таблица ЛИБО data-таблица (если задан timestamp_col)
     # Колонка-счётчик "сколько объектов обновил done-прогон". Агрегируется как
     # COALESCE(SUM(<work_col>) FILTER (...), 0) — FILTER применяется к SUM, не к COALESCE.
+    # В timestamp-режиме (timestamp_col задан) work_col игнорируется.
     work_col: str
     started_col: str = "started_at"
     # Колонка-fallback для last_attempt когда started_at NULL (queued-but-not-run).
     attempt_fallback_col: str | None = None
+    # Data-table режим: если задан, источник — это ПЛОСКАЯ data-таблица (не run-ledger,
+    # без status/started/finished). Свежесть = MAX(timestamp_col); upd_24h/_7d = COUNT(*)
+    # по строкам в окне. work_col/started_col/attempt_fallback_col при этом игнорируются.
+    timestamp_col: str | None = None
     fresh_days: float = 7.0  # < fresh_days → fresh
     stale_days: float = 30.0  # < stale_days → stale, иначе critical
     critical: bool = False  # критичный источник (для Sentry-алёртов в beat-таске)
 
 
-# Реестр источников. Все 5 таблиц проверены на проде (information_schema) —
-# статусы done присутствуют у kn/nspd_geo/objective/cadastre; nspd только failed.
+# Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на
+# проде (information_schema) — статусы done присутствуют. nspd — data-table режим:
+# смотрит nspd_quarter_dumps.fetched_at_utc (живой harvest harvest_stale_quarters +
+# 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
 #   objective   — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d
-#   nspd        — harvest_stale_quarters (пн) → fresh<8d, stale<30d
+#   nspd        — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d
 #   nspd_geo    — ручной bulk geo-job → fresh<7d, stale<30d
 #   cadastre    — ручной cadastre-job → fresh<14d, stale<45d
 _FRESHNESS_SOURCES: list[FreshnessSource] = [
@@ -1309,10 +1317,17 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [
     ),
     FreshnessSource(
         source="nspd",
-        label="NSPD кварталы/здания (harvest)",
-        table="nspd_scrape_runs",
-        work_col="quarters_ok",
-        fresh_days=8.0,
+        label="NSPD quarter dumps (harvest+resolver)",
+        # Data-table режим: nspd_quarter_dumps — это контент-таблица квартальных
+        # дампов (NO status/started/finished). Живой harvest пишет fetched_at_utc;
+        # defunct nspd_scrape_runs (manual WAF-ban 2026-04-30) больше НЕ источник истины.
+        table="nspd_quarter_dumps",
+        timestamp_col="fetched_at_utc",
+        # В timestamp-режиме не используется — оставляем валидное имя колонки.
+        work_col="total_features",
+        # Медленный кадастровый + lazy-refresh источник: дампы освежаются по мере
+        # устаревания кварталов / on-demand из /analyze. Текущий возраст ~4d → fresh.
+        fresh_days=14.0,
         stale_days=30.0,
     ),
     FreshnessSource(
@@ -1354,7 +1369,7 @@ def compute_freshness(db: Session) -> dict[str, Any]:
     scrape_freshness_check). Принимает открытую сессию, НЕ коммитит/закрывает её —
     жизненным циклом сессии управляет вызывающий (FastAPI-зависимость или таска).
 
-    Для каждого источника один проход по его run-таблице агрегирует:
+    Для run-ledger источников один проход по run-таблице агрегирует:
       - last_success_at  — max(COALESCE(finished_at, started_at)) WHERE status='done'
       - last_attempt_at  — max(started_at) (или created_at fallback) по всем прогонам
       - last_status      — status последнего прогона
@@ -1363,6 +1378,13 @@ def compute_freshness(db: Session) -> dict[str, Any]:
       - freshness        — fresh / stale / critical / unknown (по порогам источника)
       - status           — ok / stale / failed (для UI-светофора)
 
+    Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps)
+    нет run-ledger семантики (status/started/finished отсутствуют), поэтому:
+      - last_success_at = last_attempt_at = MAX(timestamp_col)
+      - objects_updated_24h / _7d = COUNT(*) строк, обновлённых в окне
+      - last_status = NULL (косметика только для run-ledger'ов)
+    Остальной downstream (age_days / _classify_freshness / status-маппинг) — общий.
+
     `overall_status` агрегирует худший статус по критичным источникам.
     """
     sources_out: list[dict[str, Any]] = []
@@ -1370,54 +1392,83 @@ def compute_freshness(db: Session) -> dict[str, Any]:
     severity = {"ok": 0, "stale": 1, "failed": 2}
 
     for src in _FRESHNESS_SOURCES:
-        started = src.started_col
-        attempt_expr = (
-            f"MAX(COALESCE({started}, {src.attempt_fallback_col}))"
-            if src.attempt_fallback_col
-            else f"MAX({started})"
-        )
-        # last_status — отдельным скаляром: status последнего по времени прогона.
-        order_col = (
-            f"COALESCE({started}, {src.attempt_fallback_col})"
-            if src.attempt_fallback_col
-            else started
-        )
-        # work_col / table / column-имена — из доверенного серверного реестра
-        # (НЕ из пользовательского ввода), поэтому f-string в SQL здесь безопасен;
+        # table / column-имена — из доверенного серверного реестра (НЕ из
+        # пользовательского ввода), поэтому f-string в SQL здесь безопасен;
         # все временные границы передаются параметрами (:d1/:d7).
-        # FILTER применяется к SUM(...), а COALESCE(...,0) оборачивает результат —
-        # COALESCE(...) FILTER(...) — синтаксическая ошибка в PostgreSQL.
-        row = (
-            db.execute(
-                text(
-                    f"""
-                    SELECT
-                        MAX(COALESCE(finished_at, {started}))
-                            FILTER (WHERE status = 'done')          AS last_success_at,
-                        {attempt_expr}                              AS last_attempt_at,
-                        COALESCE(SUM({src.work_col}) FILTER (
-                            WHERE status = 'done'
-                              AND COALESCE(finished_at, {started})
-                                  > NOW() - CAST(:d1 AS interval)
-                        ), 0)                                       AS upd_24h,
-                        COALESCE(SUM({src.work_col}) FILTER (
-                            WHERE status = 'done'
-                              AND COALESCE(finished_at, {started})
-                                  > NOW() - CAST(:d7 AS interval)
-                        ), 0)                                       AS upd_7d,
-                        (
-                            SELECT status FROM {src.table}
-                             ORDER BY {order_col} DESC NULLS LAST
-                             LIMIT 1
-                        )                                           AS last_status
-                      FROM {src.table}
-                    """
-                ),
-                {"d1": "24 hours", "d7": "7 days"},
+        if src.timestamp_col is not None:
+            # Data-table режим: плоская контент-таблица без run-ledger семантики
+            # (нет status/started/finished). Свежесть = MAX(timestamp_col),
+            # upd_24h/_7d = COUNT(*) строк, обновлённых в окне. last_status=NULL
+            # (косметика только для run-ledger'ов).
+            ts = src.timestamp_col
+            row = (
+                db.execute(
+                    text(
+                        f"""
+                        SELECT
+                            MAX({ts})                               AS last_success_at,
+                            MAX({ts})                               AS last_attempt_at,
+                            COALESCE(COUNT(*) FILTER (
+                                WHERE {ts} > NOW() - CAST(:d1 AS interval)
+                            ), 0)                                   AS upd_24h,
+                            COALESCE(COUNT(*) FILTER (
+                                WHERE {ts} > NOW() - CAST(:d7 AS interval)
+                            ), 0)                                   AS upd_7d,
+                            NULL                                    AS last_status
+                          FROM {src.table}
+                        """
+                    ),
+                    {"d1": "24 hours", "d7": "7 days"},
+                )
+                .mappings()
+                .first()
+            )
+        else:
+            started = src.started_col
+            attempt_expr = (
+                f"MAX(COALESCE({started}, {src.attempt_fallback_col}))"
+                if src.attempt_fallback_col
+                else f"MAX({started})"
+            )
+            # last_status — отдельным скаляром: status последнего по времени прогона.
+            order_col = (
+                f"COALESCE({started}, {src.attempt_fallback_col})"
+                if src.attempt_fallback_col
+                else started
+            )
+            # FILTER применяется к SUM(...), а COALESCE(...,0) оборачивает результат —
+            # COALESCE(...) FILTER(...) — синтаксическая ошибка в PostgreSQL.
+            row = (
+                db.execute(
+                    text(
+                        f"""
+                        SELECT
+                            MAX(COALESCE(finished_at, {started}))
+                                FILTER (WHERE status = 'done')      AS last_success_at,
+                            {attempt_expr}                          AS last_attempt_at,
+                            COALESCE(SUM({src.work_col}) FILTER (
+                                WHERE status = 'done'
+                                  AND COALESCE(finished_at, {started})
+                                      > NOW() - CAST(:d1 AS interval)
+                            ), 0)                                   AS upd_24h,
+                            COALESCE(SUM({src.work_col}) FILTER (
+                                WHERE status = 'done'
+                                  AND COALESCE(finished_at, {started})
+                                      > NOW() - CAST(:d7 AS interval)
+                            ), 0)                                   AS upd_7d,
+                            (
+                                SELECT status FROM {src.table}
+                                 ORDER BY {order_col} DESC NULLS LAST
+                                 LIMIT 1
+                            )                                       AS last_status
+                          FROM {src.table}
+                        """
+                    ),
+                    {"d1": "24 hours", "d7": "7 days"},
+                )
+                .mappings()
+                .first()
             )
-            .mappings()
-            .first()
-        )
 
         last_success = row["last_success_at"] if row else None
         last_attempt = row["last_attempt_at"] if row else None

backend/tests/api/v1/test_admin_scrape_freshness.py

@@ -58,8 +58,8 @@ def _override_db(db: MagicMock) -> None:
 
 
 def test_freshness_mixed_statuses() -> None:
-    """kn свежий (ok), objective stale, nspd без успехов (failed), nspd_geo critical,
-    cadastre fresh. overall_status = failed (по критичному nspd_geo→failed)."""
+    """kn свежий (ok), objective stale, nspd (data-table) свежий (ok), nspd_geo critical,
+    cadastre fresh. overall_status = stale (по критичному objective→stale)."""
     rows = {
         "kn_scrape_runs": {
             "last_success_at": _ago(1.0),  # < fresh_days=2 → fresh → ok
@@ -75,12 +75,14 @@ def test_freshness_mixed_statuses() -> None:
             "upd_7d": 325000,
             "last_status": "done",
         },
-        "nspd_scrape_runs": {
-            "last_success_at": None,  # нет успехов → failed
-            "last_attempt_at": _ago(20.0),
-            "upd_24h": 0,
-            "upd_7d": 0,
-            "last_status": "failed",
+        # nspd теперь data-table режим (nspd_quarter_dumps.fetched_at_utc):
+        # last_success_at = MAX(timestamp_col), last_status = NULL (нет run-ledger).
+        "nspd_quarter_dumps": {
+            "last_success_at": _ago(4.0),  # < fresh_days=14 → fresh → ok
+            "last_attempt_at": _ago(4.0),
+            "upd_24h": 5,
+            "upd_7d": 56,
+            "last_status": None,
         },
         "nspd_geo_jobs": {
             "last_success_at": _ago(40.0),  # >= stale_days=30 → critical → failed
@@ -117,9 +119,11 @@ def test_freshness_mixed_statuses() -> None:
     assert by_src["objective"]["status"] == "stale"
     assert by_src["objective"]["freshness"] == "stale"
 
-    assert by_src["nspd"]["status"] == "failed"
-    assert by_src["nspd"]["last_success_at"] is None
-    assert by_src["nspd"]["age_days"] is None
+    # nspd — data-table режим: свежий dump → ok, last_status=NULL не ломает маппинг.
+    assert by_src["nspd"]["status"] == "ok"
+    assert by_src["nspd"]["freshness"] == "fresh"
+    assert by_src["nspd"]["last_status"] is None
+    assert by_src["nspd"]["objects_updated_7d"] == 56
 
     assert by_src["nspd_geo"]["status"] == "failed"
     assert by_src["nspd_geo"]["freshness"] == "critical"
@@ -132,6 +136,84 @@ def test_freshness_mixed_statuses() -> None:
     assert body["overall_status"] == "stale"
 
 
+def _healthy_run_rows_except_nspd(nspd_row: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Все run-ledger источники свежие, nspd — отдельной data-table строкой.
+
+    Используется для изоляции поведения nspd-источника в data-table режиме, не
+    затрагивая остальные источники.
+    """
+    fresh_run = {
+        "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:
+        rows[s.table] = nspd_row if s.source == "nspd" else dict(fresh_run)
+    return rows
+
+
+def test_nspd_uses_timestamp_col_mode_fresh() -> None:
+    """nspd теперь читает data-таблицу nspd_quarter_dumps по timestamp_col=fetched_at_utc.
+
+    Свежий dump (~4d < fresh_days=14) + last_status=NULL → freshness=fresh, status=ok
+    (НЕ failed). Регрессия на repoint defunct nspd_scrape_runs → живой harvest-dump.
+    """
+    nspd = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "nspd")
+    # Подтверждаем что источник в data-table режиме и указывает на живую таблицу.
+    assert nspd.timestamp_col == "fetched_at_utc"
+    assert nspd.table == "nspd_quarter_dumps"
+    assert nspd.critical is False  # остаётся некритичным
+
+    rows = _healthy_run_rows_except_nspd(
+        {
+            "last_success_at": _ago(4.0),  # < fresh_days=14 → fresh
+            "last_attempt_at": _ago(4.0),
+            "upd_24h": 5,
+            "upd_7d": 56,
+            "last_status": None,  # data-table → NULL, маппинг не должен падать
+        }
+    )
+    _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"]}
+    assert by_src["nspd"]["freshness"] == "fresh"
+    assert by_src["nspd"]["status"] == "ok"
+    assert by_src["nspd"]["last_status"] is None
+    assert by_src["nspd"]["objects_updated_7d"] == 56
+    assert by_src["nspd"]["age_days"] is not None and by_src["nspd"]["age_days"] < 14.0
+
+
+def test_nspd_timestamp_col_stale_dump_is_critical() -> None:
+    """Старый dump (40d >= stale_days=30) в data-table режиме → critical → status=failed."""
+    rows = _healthy_run_rows_except_nspd(
+        {
+            "last_success_at": _ago(40.0),  # >= stale_days=30 → critical
+            "last_attempt_at": _ago(40.0),
+            "upd_24h": 0,
+            "upd_7d": 0,
+            "last_status": None,
+        }
+    )
+    _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"]}
+    assert by_src["nspd"]["freshness"] == "critical"
+    assert by_src["nspd"]["status"] == "failed"
+
+
 def test_freshness_all_healthy() -> None:
     """Все источники свежие → overall_status=ok."""
     fresh_row = {