fix(freshness): output-floor on cycle-SUM over fresh_days window (#1945, #1947)

Adversarial review found the single-newest last_success_work basis fragile:
a load cycle = MULTIPLE done-runs (objective per group_name, kn per region +
resume), so the newest run alone misrepresents the cycle. Replace with
SUM(work_col) FILTER(done) over the source's own fresh_days window.

- recent_output = SUM(COALESCE(work_col,0)) FILTER done in fresh_days window
  (make_interval secs => fresh_days*86400; no :: cast; NULL-counter -> 0)
- downgrade when status==ok and recent_output < min_output_rows (strict <)
- registry floors kept: kn_flats=50000 (healthy ~376k, broken sum <=3670),
  objective=1000 (SUM 7d ~946k); comments rewritten to cycle-sum basis +
  kn_flats zombie-resume KNOWN LIMITATION
- window=fresh_days keeps the #1947 aging-FP fix (7-8d run stays in window)
- tests: low-output-cycle->failed, above-floor->ok, boundary==floor->ok,
  weekly-aging->ok, age-stale precedence, registry floors

Verified on prod: domrf_kn_flats latest snapshot 2026-06-22=9 flats vs healthy
2026-05-17=376604 (broken 5+ weeks); objective SUM(rows_lots,7d)=946264.

backend/app/api/v1/admin_scrape.py

@@ -1280,12 +1280,21 @@ class FreshnessSource(BaseModel):
     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) режиму.
+    # НО SUM(work_col) по done-прогонам цикла (окно fresh_days) < min_output_rows,
+    # ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-почти-ничего-не-извлёк"
+    # (напр. WAF-break / тихий частичный extract: status=done, но flats_count суммарно
+    # 9 при здоровых ~376k). По умолчанию OFF — включаем только там, где малый выход =
+    # поломка (не legitimate-low, как nspd/nspd_geo/cadastre). Применимо только к
+    # run-ledger (не data-table) режиму.
     alert_on_zero_output: bool = False
+    # Пол (floor) для recent_output = SUM(work_col) по done-прогонам ЦИКЛА в окне
+    # fresh_days (НЕ один прогон — цикл объединяет несколько прогонов: objective по
+    # group_name, kn по region_codes+resume). Проверяется ТОЛЬКО при
+    # alert_on_zero_output=True: recent_output < min_output_rows → даунгрейд (строгое <).
+    # Окно = fresh_days, поэтому здоровый недельный прогон возрастом 7–8d остаётся
+    # внутри окна и не ложно-срабатывает (#1947 fix). Default 1 → флагует только если
+    # суммарный выход цикла = 0 (безопасный минимальный catch).
+    min_output_rows: int = 1
 
 
 # Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на
@@ -1295,9 +1304,12 @@ class FreshnessSource(BaseModel):
 # забаненных manual-прогона 2026-04-30, никогда не шедулился → ложный "no successful runs").
 # Пороги:
 #   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
+#   kn_flats    — та же run-таблица, work_col=flats_count + alert_on_zero_output +
+#                 min_output_rows=50000 (issue #1945: kn-run status=done но flats_count
+#                 мал/нестабилен — монитор смотрел objects_count ~1548, не flats_count;
+#                 #1947 fix: floor сравнивается с SUM(flats_count) по done-прогонам цикла
+#                 в окне fresh_days=8d — ловит low-output 9/2296/3670 и не ложит на 7–8d
+#                 aging) → 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
@@ -1318,11 +1330,23 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [
     ),
     FreshnessSource(
         # kn-run пишет ДВА счётчика: objects_count (ЖК, ~1548 — выше) и flats_count
-        # (квартиры). Источник истины для "лоадер прогнался, но ничего не извлёк"
-        # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count=0 ~5 недель,
+        # (квартиры). Источник истины для "лоадер прогнался, но почти ничего не извлёк"
+        # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count мал/нестабилен,
         # монитор молчал — objects_count оставался здоровым). Та же run-таблица и пороги,
-        # что у kn, но work_col=flats_count + alert_on_zero_output → 0 квартир за неделю
-        # done-прогонов даунгрейдит otherwise-fresh источник до failed.
+        # что у kn, но work_col=flats_count + alert_on_zero_output + min_output_rows.
+        #
+        # Floor сравнивается с recent_output = SUM(flats_count) по ВСЕМ done-прогонам в
+        # окне fresh_days=8d (kn пишет строку на region_codes + resume — это НЕ один
+        # прогон). Магнитуды (прод gendesign-postgres-1, #1947): healthy цикл ~376k
+        # квартир (snapshot 05-17=376604); сломанное состояние — суммарно ≤3670, сейчас
+        # SUM(flats_count,8d)=9 (latest done = resume, считает только хвост; ground truth
+        # snapshot 2026-06-22=9, сломан 5+ недель). Порог 50000 чисто разделяет:
+        # healthy ≫ 50k ≫ broken ≤3670.
+        #
+        # KNOWN LIMITATION: zombie-resume, загрузивший лишь малый хвост БЕЗ полного
+        # прогона в окне, может транзиентно недосчитать → максимум одно-цикловый ложный
+        # алёрт, самоисцеляется следующим полным прогоном. Приемлемо (а в текущем реальном
+        # resume-кейсе snapshot реально сломан=9 → флаг корректен).
         source="kn_flats",
         label="DOM.РФ kn-API (квартиры)",
         table="kn_scrape_runs",
@@ -1331,6 +1355,7 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [
         stale_days=14.0,
         critical=True,
         alert_on_zero_output=True,
+        min_output_rows=50000,
     ),
     FreshnessSource(
         source="objective",
@@ -1340,9 +1365,16 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [
         fresh_days=7.0,
         stale_days=14.0,
         critical=True,
-        # Объектив-sync, прогнавшийся с 0 лотов за неделю — тоже поломка (тихий
-        # пустой extract), а не легитимный 0. Даунгрейд otherwise-fresh до failed.
+        # Объектив-sync, прогнавшийся с почти нулём лотов — тоже поломка (тихий
+        # частичный extract), а не легитимный low. objective пишет строку на group_name
+        # (4 группы: Свердловская/Тюмень/Пермь/Челябинск), поэтому floor сравнивается с
+        # recent_output = SUM(rows_lots) по группам в окне fresh_days=7d (прод: SUM 7d=
+        # 946264; даже наименьшая отдельная группа Челябинск=68387). Floor 1000 ловит
+        # вырожденный цикл с огромным запасом, не задевая нормальный разброс — и
+        # робастен к тому, КАКАЯ группа дописалась последней (single-newest был бы
+        # хрупок: 324570 или 68387 в зависимости от порядка финиша).
         alert_on_zero_output=True,
+        min_output_rows=1000,
     ),
     FreshnessSource(
         source="nspd",
@@ -1403,15 +1435,21 @@ def compute_freshness(db: Session) -> dict[str, Any]:
       - last_attempt_at  — max(started_at) (или created_at fallback) по всем прогонам
       - last_status      — status последнего прогона
       - objects_updated_24h / _7d — сумма run-счётчика по done-прогонам в окне
+      - recent_output    — SUM(work_col) по done-прогонам цикла в окне fresh_days (NULL
+                           если успехов в окне нет)
       - age_days         — возраст last_success в днях (NULL если успехов не было)
       - freshness        — fresh / stale / critical / unknown (по порогам источника)
       - status           — ok / stale / failed (для UI-светофора)
-      - reason           — пояснение к zero-output-даунгрейду (иначе None)
+      - reason           — пояснение к output-floor-даунгрейду (иначе 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).
+    Output-floor (issue #1945, redesigned #1947): для run-ledger источников с
+    alert_on_zero_output=True источник, который иначе был бы fresh→ok (недавний
+    успешный прогон), но SUM(work_col) по ВСЕМ done-прогонам цикла в окне fresh_days
+    (recent_output) < min_output_rows, ДАУНГРЕЙДИТСЯ до failed — ловит
+    "лоадер-прогнался-но-почти-ничего-не-извлёк" (kn flats_count суммарно 9 при
+    здоровых ~376k). Cycle-SUM (не один прогон) робастен к multi-run циклам: objective
+    пишет строку на group_name, kn — на region_codes + resume. Окно fresh_days (не
+    хардкод 7d) → здоровый недельный прогон 7–8d остаётся в окне (нет aging-FP).
     Только когда возраст НЕ покрывает поломку (status=='ok'), чтобы не двойнить.
 
     Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps)
@@ -1496,7 +1534,25 @@ def compute_freshness(db: Session) -> dict[str, Any]:
                                 SELECT status FROM {src.table}
                                  ORDER BY {order_col} DESC NULLS LAST
                                  LIMIT 1
-                            )                                       AS last_status
+                            )                                       AS last_status,
+                            -- Cycle-sum: SUM(work_col) по ВСЕМ done-прогонам в окне
+                            -- fresh_days. Логический цикл загрузки = НЕСКОЛЬКО прогонов
+                            -- (objective — по строке на group_name, kn — по region_codes
+                            -- + resume), поэтому суммируем, а не берём один прогон.
+                            -- Окно = fresh_days (per-source), а НЕ хардкод 7d: здоровый
+                            -- недельный прогон возрастом 7–8d остаётся ВНУТРИ окна →
+                            -- нет aging-FP (#1947). Inner COALESCE(work_col,0) — NULL-
+                            -- счётчик на done-прогоне вносит 0, не NULL. fresh_days —
+                            -- float из доверенного реестра → f-string безопасен. Окно
+                            -- через make_interval(secs => fresh_days*86400): secs —
+                            -- double precision, поэтому дробный fresh_days валиден (у
+                            -- days int-only → 7.5 упал бы). NB без двоеточийного cast.
+                            SUM(COALESCE({src.work_col}, 0)) FILTER (
+                                WHERE status = 'done'
+                                  AND COALESCE(finished_at, {started})
+                                      > now()
+                                        - make_interval(secs => {src.fresh_days} * 86400)
+                            )                                       AS recent_output
                           FROM {src.table}
                         """
                     ),
@@ -1509,6 +1565,13 @@ def compute_freshness(db: Session) -> dict[str, Any]:
         last_success = row["last_success_at"] if row else None
         last_attempt = row["last_attempt_at"] if row else None
         last_status = row["last_status"] if row else None
+        # SUM(work_col) по done-прогонам в окне fresh_days (run-ledger режим = весь
+        # цикл загрузки). В data-table режиме колонки нет → None; .get() терпит
+        # отсутствие ключа в mapping-строке. None также когда done-прогонов в окне нет.
+        recent_output_raw = row.get("recent_output") if row else None
+        recent_output: int | None = (
+            int(recent_output_raw) if recent_output_raw is not None else None
+        )
 
         age_days: float | None = None
         if last_success is not None:
@@ -1530,25 +1593,32 @@ 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.
+        # Output-floor downgrade (issue #1945, redesigned #1947): источник прогнался
+        # успешно И недавно (был бы fresh→ok по возрасту), НО SUM(work_col) по всем
+        # done-прогонам ЦИКЛА (окно fresh_days) ниже порога min_output_rows → ДАУНГРЕЙД
+        # до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила.
+        #   - cycle-SUM (а не один прогон): цикл = несколько done-прогонов (objective —
+        #     по group_name, kn — по region_codes + resume), сумма даёт магнитуду цикла
+        #     (objective ~946k, kn_flats ~376k healthy) → робастно к multi-run.
+        #   - окно fresh_days (а не хардкод 7d): здоровый недельный прогон 7–8d остаётся
+        #     внутри окна → нет aging-FP первой версии #1947.
+        # Только run-ledger (timestamp_col is None) c alert_on_zero_output=True и ТОЛЬКО
+        # когда источник иначе был бы fresh/ok (по возрасту stale/failed уже покрыт — не
+        # двойним). status=='ok' гарантирует недавний успех (last_success is not None).
+        # Строгое '<': SUM ровно на пороге → ok (порог = минимально-приемлемый объём).
         reason: str | None = None
         if (
             src.alert_on_zero_output
             and src.timestamp_col is None
             and status == "ok"
-            and upd_7d == 0
+            and recent_output is not None
+            and recent_output < src.min_output_rows
         ):
             status = "failed"
             reason = (
-                f"zero-output: прогон успешен (age={age_days}d), но {src.work_col}=0 "
-                f"за 7d done-прогонов — лоадер прогнался, но ничего не извлёк (#1945)"
+                f"low-output: done-прогоны за {src.fresh_days:g}d дали "
+                f"SUM({src.work_col})={recent_output} < порога {src.min_output_rows} "
+                f"(#1945, #1947)"
             )
 
         if src.critical and severity[status] > severity[worst_critical]:
@@ -1564,11 +1634,15 @@ def compute_freshness(db: Session) -> dict[str, Any]:
                 "last_status": last_status,
                 "objects_updated_24h": upd_24h,
                 "objects_updated_7d": upd_7d,
+                # SUM(work_col) по done-прогонам цикла в окне fresh_days (None для
+                # data-table режима / без done-прогонов в окне). Additive — база для
+                # output-floor-проверки, полезно и для UI-диагностики.
+                "recent_output": recent_output,
                 "age_days": age_days,
                 "thresholds": {"fresh_days": src.fresh_days, "stale_days": src.stale_days},
                 "freshness": freshness,
                 "status": status,
-                # Доп. контекст при zero-output-даунгрейде (None в обычном случае).
+                # Доп. контекст при output-floor-даунгрейде (None в обычном случае).
                 # Additive — не ломает существующий UI/JSON-контракт.
                 "reason": reason,
             }

backend/tests/api/v1/test_admin_scrape_freshness.py

@@ -4,8 +4,9 @@
 nspd / nspd_geo / cadastre — kn и kn_flats делят kn_scrape_runs, разные work_col)
 и для каждого считает свежесть. Здесь DB замокан: на каждый per-source SELECT
 возвращаем заранее заданную строку, чтобы проверить классификацию
-fresh/stale/critical, маппинг status (ok/stale/failed), zero-output downgrade
-(#1945) и агрегат overall_status по критичным источникам.
+fresh/stale/critical, маппинг status (ok/stale/failed), output-floor downgrade
+(#1945 / redesigned #1947 — recent_output = SUM(work_col) по done-прогонам цикла в
+окне fresh_days < min_output_rows) и агрегат overall_status по критичным источникам.
 """
 
 from __future__ import annotations
@@ -68,6 +69,7 @@ def test_freshness_mixed_statuses() -> None:
             "upd_24h": 1500,
             "upd_7d": 1500,
             "last_status": "done",
+            "recent_output": 376000,  # здоровый цикл flats_count → kn_flats ok
         },
         "objective_scrape_runs": {
             "last_success_at": _ago(10.0),  # fresh<7, stale<14 → stale
@@ -75,6 +77,7 @@ def test_freshness_mixed_statuses() -> None:
             "upd_24h": 0,
             "upd_7d": 325000,
             "last_status": "done",
+            "recent_output": 325000,
         },
         # nspd теперь data-table режим (nspd_quarter_dumps.fetched_at_utc):
         # last_success_at = MAX(timestamp_col), last_status = NULL (нет run-ledger).
@@ -150,6 +153,7 @@ def _healthy_run_rows_except_nspd(nspd_row: dict[str, Any]) -> dict[str, dict[st
         "upd_24h": 10,
         "upd_7d": 100,
         "last_status": "done",
+        "recent_output": 100000,  # выше любого floor → flagged-источники ok
     }
     rows: dict[str, dict[str, Any]] = {}
     for s in admin_scrape._FRESHNESS_SOURCES:
@@ -224,6 +228,7 @@ def test_freshness_all_healthy() -> None:
         "upd_24h": 10,
         "upd_7d": 100,
         "last_status": "done",
+        "recent_output": 100000,  # выше любого floor → flagged-источники ok
     }
     rows = {
         s.table: dict(fresh_row) for s in admin_scrape._FRESHNESS_SOURCES
@@ -249,15 +254,18 @@ def test_freshness_critical_source_failed_drives_overall() -> None:
         "upd_24h": 1,
         "upd_7d": 1,
         "last_status": "done",
+        "recent_output": 100000,
     }
     rows = {s.table: dict(base) for s in admin_scrape._FRESHNESS_SOURCES}
-    # objective критичный — обнулим успех.
+    # objective критичный — обнулим успех (last_success=None → нет done-прогона,
+    # recent_output=None → floor не применяется, status=failed по отсутствию успеха).
     rows["objective_scrape_runs"] = {
         "last_success_at": None,
         "last_attempt_at": _ago(5.0),
         "upd_24h": 0,
         "upd_7d": 0,
         "last_status": "running",
+        "recent_output": None,
     }
     _override_db(_make_db(rows))
     try:
@@ -281,37 +289,48 @@ 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).
+# NB по мокам output-floor: _make_db ключуется по ИМЕНИ таблицы → kn и kn_flats
+# делят одну kn_scrape_runs-строку, поэтому оба видят ОДИН recent_output (per-work_col
+# дивергенция objects_count vs flats_count есть только в живом SQL, не в моке). Здесь
+# мок проверяет Python-сравнение (recent_output < min_output_rows + все guard'ы);
+# семантику оконной SQL-SUM подтвердили на проде (objective SUM(rows_lots,7d)=946264,
+# kn_flats SUM(flats_count,8d)=9 — gendesign-postgres-1, #1947).
 
-    Та же run-таблица kn_scrape_runs у kn (alert_on_zero_output=False) и kn_flats
-    (True). Возвращаем строку с upd_7d=0 → kn остаётся ok (нет флага = нет ложного
-    срабатывания), kn_flats падает в failed (zero-output) с пояснением в reason.
+
+def test_output_floor_downgrades_low_output_cycle() -> None:
+    """issue #1945 / #1947: kn_flats прогнался успешно И недавно (был бы fresh→ok), но
+    recent_output=SUM(flats_count) по циклу=9 < min_output_rows=50000 → failed (alertable).
+
+    Прод-сценарий: цикл дал суммарно flats_count=9 (latest done = resume, считает хвост)
+    при здоровых ~376k. kn БЕЗ флага (alert_on_zero_output=False) → branch не входит,
+    остаётся ok даже читая ту же низкую recent_output (no false positive).
     """
-    fresh_zero = {
+    # Сломанный цикл: свежий (1d), но recent_output (SUM flats_count за 8d)=9.
+    broken_kn = {
         "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
+        "upd_24h": 9,
+        "upd_7d": 9,
         "last_status": "done",
+        "recent_output": 9,  # SUM(flats_count) по done-прогонам цикла ≪ 50000
     }
-    # Остальные источники свежие+ненулевые, чтобы изолировать поведение kn/kn_flats.
-    nonzero_fresh = {
+    # Остальные источники свежие+здоровые, чтобы изолировать поведение kn/kn_flats.
+    healthy = {
         "last_success_at": _ago(0.5),
         "last_attempt_at": _ago(0.5),
-        "upd_24h": 10,
-        "upd_7d": 100,
+        "upd_24h": 100,
+        "upd_7d": 1000,
         "last_status": "done",
+        "recent_output": 100000,  # выше любого floor
     }
     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}
+            rows[s.table] = {**healthy, "last_status": None, "recent_output": None}
         elif s.table == "kn_scrape_runs":
-            rows[s.table] = dict(fresh_zero)
+            rows[s.table] = dict(broken_kn)
         else:
-            rows[s.table] = dict(nonzero_fresh)
+            rows[s.table] = dict(healthy)
     _override_db(_make_db(rows))
     try:
         resp = TestClient(app).get(ENDPOINT)
@@ -322,14 +341,16 @@ def test_zero_output_downgrades_otherwise_fresh_source() -> None:
     body = resp.json()
     by_src = {s["source"]: s for s in body["sources"]}
 
-    # kn_flats: otherwise-fresh, но flats_count=0 → DOWNGRADE до failed (alertable).
+    # kn_flats: otherwise-fresh, но recent_output=9 < 50000 → DOWNGRADE до failed.
     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"]["status"] == "failed"  # output-floor downgrade
+    assert by_src["kn_flats"]["recent_output"] == 9
     assert by_src["kn_flats"]["reason"] is not None
     assert "flats_count" in by_src["kn_flats"]["reason"]
+    assert "9" in by_src["kn_flats"]["reason"]
 
-    # kn — та же таблица, upd_7d=0, но БЕЗ флага → остаётся ok (no false positive).
+    # kn — alert_on_zero_output=False гейтит ВЕСЬ branch → ok, даже читая recent_output=9
+    # из той же mocked-строки (per-work_col дивергенция есть только в живом SQL).
     assert by_src["kn"]["status"] == "ok"
     assert by_src["kn"]["reason"] is None
 
@@ -338,25 +359,26 @@ def test_zero_output_downgrades_otherwise_fresh_source() -> None:
     assert body["overall_status"] == "failed"
 
 
-def test_zero_output_no_false_positive_when_nonzero() -> None:
-    """alert_on_zero_output=True источник с НЕнулевым выходом → остаётся ok (нет даунгрейда).
+def test_output_floor_no_false_positive_when_above_floor() -> None:
+    """alert_on_zero_output источник с recent_output ВЫШЕ floor → остаётся ok.
 
-    Регрессия: zero-output даунгрейд НЕ должен бить по здоровому источнику, который
-    прогнался и реально извлёк строки.
+    Регрессия: output-floor НЕ должен бить по здоровому источнику, чей цикл извлёк
+    нормальный объём (kn_flats ~376k > 50000; objective ~946k > 1000).
     """
-    nonzero_fresh = {
+    healthy = {
         "last_success_at": _ago(1.0),
         "last_attempt_at": _ago(1.0),
         "upd_24h": 12,
-        "upd_7d": 1234,  # ненулевой выход
+        "upd_7d": 1234,
         "last_status": "done",
+        "recent_output": 376000,  # здоровый цикл ≫ всех floor'ов
     }
     rows: dict[str, dict[str, Any]] = {}
     for s in admin_scrape._FRESHNESS_SOURCES:
         rows[s.table] = (
-            {**nonzero_fresh, "last_status": None}
+            {**healthy, "last_status": None, "recent_output": None}
             if s.timestamp_col is not None
-            else dict(nonzero_fresh)
+            else dict(healthy)
         )
     _override_db(_make_db(rows))
     try:
@@ -366,38 +388,134 @@ def test_zero_output_no_false_positive_when_nonzero() -> None:
 
     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.
+    # Все alert_on_zero_output-источники с выходом выше floor → 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)."""
+def test_output_floor_boundary_equal_floor_stays_ok() -> None:
+    """Граница: recent_output == min_output_rows → остаётся ok (проверка строгое '<').
+
+    kn_flats recent_output ровно 50000 (= floor) → НЕ даунгрейдится (порог = минимально-
+    приемлемый объём, на нём ещё ok). Регрессия на off-by-one в сравнении.
+    """
+    kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats")
+    at_floor = {
+        "last_success_at": _ago(1.0),
+        "last_attempt_at": _ago(1.0),
+        "upd_24h": 0,
+        "upd_7d": kn_flats.min_output_rows,
+        "last_status": "done",
+        "recent_output": kn_flats.min_output_rows,  # ровно на пороге → ok
+    }
+    healthy = {
+        "last_success_at": _ago(0.5),
+        "last_attempt_at": _ago(0.5),
+        "upd_24h": 100,
+        "upd_7d": 1000,
+        "last_status": "done",
+        "recent_output": 1000000,
+    }
+    rows: dict[str, dict[str, Any]] = {}
+    for s in admin_scrape._FRESHNESS_SOURCES:
+        if s.timestamp_col is not None:
+            rows[s.table] = {**healthy, "last_status": None, "recent_output": None}
+        elif s.table == "kn_scrape_runs":
+            rows[s.table] = dict(at_floor)
+        else:
+            rows[s.table] = dict(healthy)
+    _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["kn_flats"]["recent_output"] == kn_flats.min_output_rows
+    assert by_src["kn_flats"]["status"] == "ok"  # ровно на пороге → НЕ failed
+    assert by_src["kn_flats"]["reason"] is None
+
+
+def test_output_floor_no_false_positive_on_weekly_window_aging() -> None:
+    """#1947 fix: здоровый недельный прогон возрастом 7–8d остаётся в окне fresh_days=8d
+    → recent_output здоров (376k) → НЕ ложно-флагуется.
+
+    Регрессия на самый первый #1947-баг: оконная SUM была хардкод 7d, поэтому прогон
+    7–8d (>7d назад) давал SUM=0 при status=ok → ложный downgrade ~раз/неделю. Теперь
+    окно = fresh_days (8d) → прогон 7.5d ВНУТРИ окна, recent_output=376k > floor → ok.
+    upd_7d=0 (мок старого 7d-счётчика) НЕ влияет — floor смотрит recent_output.
+    """
+    weekly_aged = {
+        "last_success_at": _ago(7.5),  # < fresh_days=8 → всё ещё fresh, и внутри окна
+        "last_attempt_at": _ago(7.5),
+        "upd_24h": 0,
+        "upd_7d": 0,  # старый 7d-счётчик: прогон >7d назад → 0 (раньше ложно флагал)
+        "last_status": "done",
+        "recent_output": 376000,  # окно fresh_days=8d ловит прогон 7.5d → цикл здоров
+    }
+    healthy = {
+        "last_success_at": _ago(0.5),
+        "last_attempt_at": _ago(0.5),
+        "upd_24h": 50,
+        "upd_7d": 500,
+        "last_status": "done",
+        "recent_output": 100000,
+    }
+    rows: dict[str, dict[str, Any]] = {}
+    for s in admin_scrape._FRESHNESS_SOURCES:
+        if s.timestamp_col is not None:
+            rows[s.table] = {**healthy, "last_status": None, "recent_output": None}
+        elif s.table == "kn_scrape_runs":
+            rows[s.table] = dict(weekly_aged)
+        else:
+            rows[s.table] = dict(healthy)
+    _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"]}
+    # kn_flats: fresh по возрасту (7.5<8), upd_7d=0 НЕ важен — floor смотрит
+    # recent_output (376k > 50000, окно fresh_days покрыло прогон) → ok, без downgrade.
+    assert by_src["kn_flats"]["freshness"] == "fresh"
+    assert by_src["kn_flats"]["status"] == "ok"
+    assert by_src["kn_flats"]["objects_updated_7d"] == 0
+    assert by_src["kn_flats"]["recent_output"] == 376000
+    assert by_src["kn_flats"]["reason"] is None
+
+
+def test_output_floor_not_applied_when_already_stale() -> None:
+    """Если источник уже stale/failed по возрасту — output-floor НЕ передоунгрейдит
+    (age уже покрывает), и не двойнит. objective: старый успех (10d → stale) +
+    recent_output=0 → остаётся stale (НЕ перепрыгивает в failed через floor)."""
     base_fresh = {
         "last_success_at": _ago(0.5),
         "last_attempt_at": _ago(0.5),
         "upd_24h": 5,
         "upd_7d": 50,
         "last_status": "done",
+        "recent_output": 500000,
     }
     rows: dict[str, dict[str, Any]] = {}
     for s in admin_scrape._FRESHNESS_SOURCES:
         rows[s.table] = (
-            {**base_fresh, "last_status": None}
+            {**base_fresh, "last_status": None, "recent_output": None}
             if s.timestamp_col is not None
             else dict(base_fresh)
         )
-    # objective: stale по возрасту (10d: fresh<7, stale<14) + 0 выход.
+    # objective: stale по возрасту (10d: fresh<7, stale<14) + нулевой выход.
     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",
+        "recent_output": 0,
     }
     _override_db(_make_db(rows))
     try:
@@ -407,19 +525,21 @@ def test_zero_output_not_applied_when_already_stale() -> None:
 
     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.
+    # output-floor применяется ТОЛЬКО к 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: та же run-таблица что у kn, но work_col=flats_count + alert_on_zero_output
+    + min_output_rows=50000, критичный, недельные пороги. kn (objects_count) — без флага."""
     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
+    # healthy цикл ~376k ≫ floor ≫ broken-sum ≤3670 (прод #1947).
+    assert kn_flats.min_output_rows == 50000
     assert kn_flats.critical is True
     assert kn_flats.fresh_days == 8.0
     assert kn_flats.stale_days == 14.0
@@ -427,14 +547,19 @@ def test_kn_flats_registry_config() -> None:
     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
+    assert kn.min_output_rows == 1  # default — не проверяется без флага
 
     objective = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "objective")
     assert objective.alert_on_zero_output is True
+    # healthy SUM(rows_lots,7d)~946k по 4 группам; floor 1000 с огромным запасом.
+    assert objective.min_output_rows == 1000
 
-    # Legitimate-0 источники НЕ должны иметь флаг (избегаем ложных срабатываний).
+    # Legitimate-low источники НЕ должны иметь флаг (избегаем ложных срабатываний);
+    # min_output_rows остаётся default=1 и не проверяется.
     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
+        assert src.min_output_rows == 1, name
 
 
 def test_kn_thresholds_match_weekly_cadence() -> None: