feat(parcels): persist analysis runs + real last_analysis_date (#994)

Persist each successful POST /analyze into analysis_runs (migration 127):
best-effort SAVEPOINT-wrapped INSERT in a thin analysis_runs repository, then
explicit db.commit() (get_db has no commit-on-success; SAVEPOINT RELEASE alone
does not persist — without the commit the row rolls back on request teardown,
a silent no-op caught in code review). A persist failure never breaks the
response or poisons the session. Result serialized via jsonable_encoder→
json.dumps (handles embedded Pydantic models/dates); confidence/status
normalized to satisfy the 127 CHECK constraints (unknown→NULL/'complete').

Populate by-bbox last_analysis_date from v_analysis_runs_latest via a single
batch query (no N+1), replacing the #307 placeholder. The read is best-effort
wrapped too — a view-missing deploy window or future drift must not 500 the
map (falls back to last_analysis_date=None).

Additive only — analyze response shape unchanged. Tests: by-bbox suite patches
latest_run_dates (mock-db returns same rows for any query) + new test asserts
ISO last_analysis_date from a latest run. Analyze mock-suite unaffected
(graceful side_effect overflow + best-effort persist absorb the extra INSERT).
Incidental: ruff-format fixed one pre-existing f-string spacing (line ~2391)
the format hook flags on touch.

Closes #994. Refs #961.

backend/app/api/v1/parcels.py

@@ -31,6 +31,11 @@ from app.schemas.parcel import (
     RedLine,
     RiskZone,
 )
+from app.services.analysis_runs.repository import (
+    ANALYZE_SCHEMA_VERSION,
+    latest_run_dates,
+    persist_analysis_run,
+)
 from app.services.exporters.layout_tz_pdf import render_layout_tz_pdf
 from app.services.exporters.snapshot_pdf import generate_snapshot_pdf
 from app.services.site_finder.best_layouts import get_best_layouts
@@ -1023,7 +1028,8 @@ async def get_parcels_by_bbox(
 
     Использует GIST-индекс на cad_parcels.geom (ST_Intersects).
     Если передан user_id — добавляет статус из parcel_user_status (overlay).
-    last_analysis_date — placeholder до реализации B2 auth (#307).
+    last_analysis_date — реальная дата последнего анализа из v_analysis_runs_latest
+    (#994, миграция 127): один batch-запрос на весь bbox, участки без рана → None.
     """
     if min_lat >= max_lat or min_lon >= max_lon:
         raise HTTPException(status_code=400, detail="Некорректный bbox: min >= max")
@@ -1081,6 +1087,16 @@ async def get_parcels_by_bbox(
         .all()
     )
 
+    # #994: дата последнего анализа на участок — один batch-запрос на весь bbox
+    # к v_analysis_runs_latest (НЕ N+1). Участки без рана отсутствуют в dict → None.
+    # Best-effort: провал чтения (напр. view 127 ещё не применён в deploy-окне, или
+    # будущий drift) НЕ должен ронять карту → fallback {} (last_analysis_date=None).
+    try:
+        last_analysis = latest_run_dates(db, [row["cad_num"] for row in rows])
+    except Exception:
+        logger.exception("by-bbox: latest_run_dates failed → last_analysis_date=None для всех")
+        last_analysis = {}
+
     markers: list[ParcelMapMarker] = [
         ParcelMapMarker(
             cad_num=row["cad_num"],
@@ -1089,7 +1105,11 @@ async def get_parcels_by_bbox(
             area_m2=float(row["area_m2"]) if row["area_m2"] is not None else None,
             land_category=row["land_category"],
             status=row["user_status"] if user_id else None,
-            last_analysis_date=None,  # placeholder, real after B2 auth (#307)
+            last_analysis_date=(
+                last_analysis[row["cad_num"]].date().isoformat()
+                if row["cad_num"] in last_analysis
+                else None
+            ),
         )
         for row in rows
     ]
@@ -1165,6 +1185,13 @@ def analyze_parcel(
         str | None,
         Header(description="Session ID пользователя для custom POI scoring (#254)"),
     ] = None,
+    x_authenticated_user: Annotated[
+        str | None,
+        Header(
+            alias="X-Authenticated-User",
+            description="Идентичность из Caddy basic_auth → created_by рана (#994, nullable)",
+        ),
+    ] = None,
 ) -> dict[str, Any]:
     """Анализ участка: близость к социалке + district context + конкуренты.
 
@@ -2364,7 +2391,7 @@ def analyze_parcel(
                             "contribution": round(_contribution, 2),
                             "verbal": (
                                 f"Custom POI '{cp['name']}' ({round(_distance_m)}м) — "
-                                f"{'+'if _contribution >= 0 else ''}{round(_contribution, 2)} б."
+                                f"{'+' if _contribution >= 0 else ''}{round(_contribution, 2)} б."
                             ),
                             "lat": cp["lat"],
                             "lon": cp["lon"],
@@ -2502,7 +2529,7 @@ def analyze_parcel(
         "top_sellers": _top_sellers_list,
     }
 
-    return {
+    result_payload: dict[str, Any] = {
         "cad_num": cad_num,
         "source": source,
         "geom_geojson": json.loads(geom_geojson) if geom_geojson else None,
@@ -2619,6 +2646,39 @@ def analyze_parcel(
         "risks": risks_block,
     }
 
+    # #994 (961-C3, ТЗ §22): persist завершённого рана в analysis_runs.
+    # Best-effort — repository обёрнут в SAVEPOINT + try/except, провал НЕ меняет
+    # форму/успех ответа (frontend зависит от него) и не отравляет outer-сессию.
+    # district денормализуем из result["district"]["district_name"] (для фильтрации
+    # без JSON-разбора); confidence — отчётный уровень high/medium/low из
+    # confidence_label (нормализуется под CHECK в repository). schema_version —
+    # ANALYZE_SCHEMA_VERSION: результат здесь — inline-dict analyze, НЕ
+    # SiteFinderReport.as_dict() (у того свой _SCHEMA_VERSION "1.0").
+    _district_name = (
+        result_payload["district"]["district_name"]
+        if isinstance(result_payload.get("district"), dict)
+        else None
+    )
+    persist_analysis_run(
+        db,
+        cad_num=cad_num,
+        result=result_payload,
+        params={
+            "profile_id": profile_id,
+            "profile_user_id": profile_user_id,
+            "inline_weights": _inline_weights,
+            "weights_source": _weights_source,
+            "x_session_id": _session_id,
+        },
+        district=_district_name,
+        confidence=confidence_info["label"],
+        status="complete",
+        schema_version=ANALYZE_SCHEMA_VERSION,
+        created_by=x_authenticated_user,
+    )
+
+    return result_payload
+
 
 @router.get(
     "/{cad_num}/connection-points",

backend/app/services/analysis_runs/__init__.py

@@ -0,0 +1,5 @@
+"""analysis_runs — durable persist/read слой анализ-ранов Site Finder (#994, ТЗ §22).
+
+Тонкий data-access поверх таблицы `analysis_runs` + view `v_analysis_runs_latest`
+(миграция 127, #993). Endpoint'ы (parcels.py) держим тонкими — вся SQL-логика здесь.
+"""

backend/app/services/analysis_runs/repository.py

@@ -0,0 +1,195 @@
+"""Data-access для analysis_runs (#994, 961-C3, ТЗ §22, EPIC 17 «GG-форсайт»).
+
+Тонкий repository поверх таблицы `analysis_runs` + view `v_analysis_runs_latest`
+(миграция 127 / #993). Держит endpoint `analyze_parcel` тонким: persist одного
+завершённого рана + чтение «даты последнего анализа» (powers by-bbox карту).
+
+Контракт миграции 127 (см. data/sql/127_analysis_runs.sql):
+  - result/params/segment — JSONB; пишем через CAST(:x AS jsonb) + json.dumps
+    (psycopg v3: НИКОГДА :x::type — backend.md). result может содержать Pydantic-
+    модели (RiskZone/OpportunityParcel/RedLine) → прогоняем через jsonable_encoder
+    перед json.dumps, иначе TypeError на сериализации.
+  - CHECK'и: status ∈ {complete,partial,error}; confidence ∈ {high,medium,low} ИЛИ
+    NULL. Чтобы НИКОГДА не словить IntegrityError на write — валидируем обе колонки
+    локально (`_normalize_confidence` / `_normalize_status`): неизвестное → NULL/дефолт,
+    не сырое значение. Плохой ран не должен превращать успешный analyze в 500.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime
+from typing import Any
+
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy import Row, text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Версия схемы рана, когда result — это dict эндпоинта analyze (НЕ SiteFinderReport).
+# SiteFinderReport.as_dict() несёт собственный _SCHEMA_VERSION ("1.0"); для inline-
+# результата analyze, у которого своей версии формы нет, фиксируем ОТДЕЛЬНУЮ строку
+# с префиксом, чтобы re-open мог различить два источника (#992 формализует контракт).
+ANALYZE_SCHEMA_VERSION = "analyze-1.0"
+
+# CHECK-зеркала миграции 127 — единственный источник допустимых значений на write.
+_ALLOWED_CONFIDENCE: frozenset[str] = frozenset({"high", "medium", "low"})
+_ALLOWED_STATUS: frozenset[str] = frozenset({"complete", "partial", "error"})
+_DEFAULT_STATUS = "complete"
+
+
+def _normalize_confidence(confidence: str | None) -> str | None:
+    """Привести confidence к CHECK-допустимому high|medium|low или None.
+
+    Пустая строка / неизвестное значение → None (а НЕ '' — это нарушило бы CHECK).
+    """
+    if confidence is None:
+        return None
+    value = confidence.strip().lower()
+    return value if value in _ALLOWED_CONFIDENCE else None
+
+
+def _normalize_status(status: str | None) -> str:
+    """Привести status к CHECK-допустимому complete|partial|error (дефолт complete)."""
+    if status is None:
+        return _DEFAULT_STATUS
+    value = status.strip().lower()
+    return value if value in _ALLOWED_STATUS else _DEFAULT_STATUS
+
+
+def _jsonb_param(value: Any) -> str:
+    """Сериализовать произвольный (возможно Pydantic-содержащий) объект в JSON-строку.
+
+    jsonable_encoder разворачивает Pydantic-модели / даты / Enum в JSON-native типы;
+    json.dumps(..., ensure_ascii=False) — кириллица как есть (зеркало pzz_loader).
+    """
+    return json.dumps(jsonable_encoder(value), ensure_ascii=False)
+
+
+def persist_analysis_run(
+    db: Session,
+    *,
+    cad_num: str,
+    result: dict[str, Any],
+    params: dict[str, Any],
+    district: str | None,
+    confidence: str | None,
+    status: str,
+    schema_version: str,
+    created_by: str | None,
+    segment: dict[str, Any] | None = None,
+) -> int | None:
+    """Сохранить ОДИН завершённый ран анализа в analysis_runs (#994).
+
+    Best-effort: обёрнут в SAVEPOINT (db.begin_nested, backend.md) + try/except —
+    провал write НЕ ломает ответ analyze и НЕ отравляет outer-транзакцию (откатывает
+    только свой savepoint, прошлые reads сессии остаются валидны). На успехе —
+    возвращает id новой строки, на провале — None (+ logger.warning/exception).
+
+    confidence/status нормализуются под CHECK'и миграции 127 ДО write — IntegrityError
+    на enum'ах структурно невозможен. advisory всегда TRUE (форсайт-стек советующий).
+    """
+    norm_confidence = _normalize_confidence(confidence)
+    norm_status = _normalize_status(status)
+    try:
+        with db.begin_nested():
+            row = (
+                db.execute(
+                    text("""
+                        INSERT INTO analysis_runs (
+                            cad_num, district, segment, params, result,
+                            schema_version, advisory, confidence, status, created_by
+                        )
+                        VALUES (
+                            CAST(:cad_num AS text),
+                            CAST(:district AS text),
+                            CAST(:segment AS jsonb),
+                            CAST(:params AS jsonb),
+                            CAST(:result AS jsonb),
+                            CAST(:schema_version AS text),
+                            TRUE,
+                            CAST(:confidence AS text),
+                            CAST(:status AS text),
+                            CAST(:created_by AS text)
+                        )
+                        RETURNING id
+                    """),
+                    {
+                        "cad_num": cad_num,
+                        "district": district,
+                        "segment": _jsonb_param(segment) if segment is not None else None,
+                        "params": _jsonb_param(params),
+                        "result": _jsonb_param(result),
+                        "schema_version": schema_version,
+                        "confidence": norm_confidence,
+                        "status": norm_status,
+                        "created_by": created_by,
+                    },
+                )
+                .mappings()
+                .first()
+            )
+        new_id = int(row["id"]) if row else None
+        # SAVEPOINT RELEASE (выход из begin_nested) НЕ персистит — лишь сливает вложенную
+        # tx в outer. get_db() (core/db.py) коммита-на-успехе НЕ делает (yield→finally:close
+        # = ROLLBACK), а analyze — единственный write на success-path. Без явного commit
+        # строка откатится на teardown (silent no-op). Коммитим ВНУТРИ try → провал commit
+        # тоже best-effort. Паттерн как в trade_in.py / photos.py (explicit commit на write).
+        db.commit()
+        logger.info(
+            "analysis_run persisted: cad=%s id=%s status=%s confidence=%s",
+            cad_num,
+            new_id,
+            norm_status,
+            norm_confidence,
+        )
+        return new_id
+    except Exception:
+        # Persist — best-effort: ран уже посчитан, ответ отдаём в любом случае.
+        # exception (не warning) — нужен traceback, write-провалы должны быть видны.
+        logger.exception("analysis_run persist failed for cad=%s (response unaffected)", cad_num)
+        return None
+
+
+def latest_run_dates(db: Session, cad_nums: list[str]) -> dict[str, datetime]:
+    """Дата последнего анализа на участок для набора cad_num (batch, НЕ N+1).
+
+    Один запрос к v_analysis_runs_latest (canonical latest-per-parcel из 127):
+    `WHERE cad_num = ANY(CAST(:cads AS text[]))`. Powers last_analysis_date в by-bbox.
+    Участки без ранов в результат не попадают (caller трактует отсутствие как None).
+    Пустой вход → пустой dict (без обращения к БД).
+    """
+    if not cad_nums:
+        return {}
+    rows = (
+        db.execute(
+            text("""
+                SELECT cad_num, created_at
+                  FROM v_analysis_runs_latest
+                 WHERE cad_num = ANY(CAST(:cads AS text[]))
+            """),
+            {"cads": cad_nums},
+        )
+        .mappings()
+        .all()
+    )
+    return {row["cad_num"]: row["created_at"] for row in rows}
+
+
+def latest_run_for(db: Session, cad_num: str) -> Row[Any] | None:
+    """Последний ран на участок целиком (для re-open / «текущий анализ участка»).
+
+    Читает из v_analysis_runs_latest (DISTINCT ON cad_num, max created_at). None если
+    участок ещё не анализировался. Возвращает Row (id/result/params/.../created_at).
+    """
+    return db.execute(
+        text("""
+                SELECT id, cad_num, district, segment, params, result,
+                       schema_version, advisory, confidence, status, created_by, created_at
+                  FROM v_analysis_runs_latest
+                 WHERE cad_num = CAST(:cad_num AS text)
+            """),
+        {"cad_num": cad_num},
+    ).first()

backend/tests/api/v1/test_parcel_by_bbox.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any
 from unittest.mock import MagicMock, patch
 
@@ -58,7 +59,13 @@ def test_by_bbox_valid_returns_structure(client: TestClient) -> None:
     rows = [_make_db_row()]
     mock_db = _build_mock_db(rows)
 
-    with patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])):
+    # latest_run_dates патчим явно (#994): mock_db.execute() возвращает те же rows на
+    # ЛЮБОЙ запрос, поэтому без патча latest_run_dates получил бы parcel-rows (без
+    # created_at) → KeyError. Пустой dict = «участок не анализировался» → date None.
+    with (
+        patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])),
+        patch("app.api.v1.parcels.latest_run_dates", return_value={}),
+    ):
         resp = client.get(
             "/api/v1/parcels/by-bbox",
             params={
@@ -113,7 +120,10 @@ def test_by_bbox_status_overlay_with_user_id(client: TestClient) -> None:
     rows = [_make_db_row(user_status="in_work")]
     mock_db = _build_mock_db(rows)
 
-    with patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])):
+    with (
+        patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])),
+        patch("app.api.v1.parcels.latest_run_dates", return_value={}),
+    ):
         resp = client.get(
             "/api/v1/parcels/by-bbox",
             params={
@@ -134,7 +144,10 @@ def test_by_bbox_no_user_id_status_is_null(client: TestClient) -> None:
     rows = [_make_db_row(user_status="favorite")]  # DB вернёт статус
     mock_db = _build_mock_db(rows)
 
-    with patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])):
+    with (
+        patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])),
+        patch("app.api.v1.parcels.latest_run_dates", return_value={}),
+    ):
         resp = client.get(
             "/api/v1/parcels/by-bbox",
             params={
@@ -150,3 +163,33 @@ def test_by_bbox_no_user_id_status_is_null(client: TestClient) -> None:
     parcel = resp.json()["parcels"][0]
     # Без user_id статус принудительно null (endpoint не раскрывает чужие статусы)
     assert parcel["status"] is None
+
+
+# ── Test 5: last_analysis_date заполняется из v_analysis_runs_latest (#994) ──
+
+
+def test_by_bbox_last_analysis_date_from_latest_run() -> None:
+    """#994: участок с раном в v_analysis_runs_latest → last_analysis_date = ISO-дата."""
+    client = TestClient(app)
+    rows = [_make_db_row()]
+    mock_db = _build_mock_db(rows)
+    latest = {"66:41:0204016:10": datetime(2026, 6, 1, 14, 30, 0)}
+
+    with (
+        patch("app.api.v1.parcels.get_db", return_value=iter([mock_db])),
+        patch("app.api.v1.parcels.latest_run_dates", return_value=latest),
+    ):
+        resp = client.get(
+            "/api/v1/parcels/by-bbox",
+            params={
+                "min_lat": 56.80,
+                "min_lon": 60.60,
+                "max_lat": 56.90,
+                "max_lon": 60.70,
+            },
+        )
+
+    assert resp.status_code == 200
+    parcel = resp.json()["parcels"][0]
+    # created_at → .date().isoformat() (только дата, без времени), str|None per parcel.py:387
+    assert parcel["last_analysis_date"] == "2026-06-01"