feat(parcels): wire §22 forecast into analyze endpoint (3b-ii, #995)

Final forecast wiring — §22 reaches the API.

- horizon query param on POST /{cad}/analyze ({6,12,18}, default 12; 422 before any
  DB/compute work otherwise).
- best-effort fire-and-forget forecast_site_finder_report.delay(cad, horizon,
  created_by) after persist (lazy import, no api↔workers cycle); Celery/Redis down →
  forecast.status='unavailable', analyze still 200 (never blocks/500s). Adds additive
  result_payload['forecast'] = {status: pending|unavailable, horizon}.
- new read-only GET /{cad}/forecast: latest schema_version='1.0' run → 200
  {status:ready, run_id, created_at, report} else 202 {status:pending}; graceful on DB
  error (202, never 500). Mirrors fetch-status RBAC/style.

Additive only — existing analyze response keys unchanged. code-review APPROVE
(analyze uncrashable + additive; 422 pre-flight; GET reads '1.0' not 'analyze-1.0').
96 api tests pass; ruff clean. Refs #995 #992 #961.

backend/app/api/v1/parcels.py

@@ -34,6 +34,7 @@ from app.schemas.parcel import (
 from app.services.analysis_runs.repository import (
     ANALYZE_SCHEMA_VERSION,
     latest_run_dates,
+    latest_run_for,
     persist_analysis_run,
 )
 from app.services.exporters.layout_tz_pdf import render_layout_tz_pdf
@@ -1013,6 +1014,13 @@ def _compute_confidence(
 _INLINE_FETCH_WAIT_S = 15
 _INLINE_FETCH_POLL_INTERVAL_S = 2
 
+# §22-форсайт (3b-ii): schema_version §22-рана в analysis_runs — это "1.0"
+# (SiteFinderReport._SCHEMA_VERSION), отличный от ANALYZE_SCHEMA_VERSION ("analyze-1.0")
+# inline-dict analyze. GET /{cad}/forecast читает именно "1.0".
+_FORECAST_SCHEMA_VERSION = "1.0"
+# Допустимые горизонты прогноза (мес), #995 — иное → 422.
+_ALLOWED_FORECAST_HORIZONS = frozenset({6, 12, 18})
+
 
 @router.get("/by-bbox", response_model=ParcelBboxResponse)
 async def get_parcels_by_bbox(
@@ -1164,6 +1172,57 @@ def get_fetch_status(
     return _fetch_status(db, cad_num)
 
 
+@router.get("/{cad_num}/forecast")
+def get_parcel_forecast(
+    cad_num: str,
+    db: Annotated[Session, Depends(get_db)],
+    response: Response,
+    x_authenticated_user: Annotated[
+        str | None,
+        Header(
+            alias="X-Authenticated-User",
+            description="Идентичность из Caddy basic_auth (#994, nullable, read-only здесь)",
+        ),
+    ] = None,
+) -> dict[str, Any]:
+    """Read-only fetch §22-форсайта участка (3b-ii, #995).
+
+    Клиент поллит после POST /analyze (тот fire-and-forget enqueue'ит фон-таску,
+    считающую §22 SiteFinderReport ~30-180s). Читает ПОСЛЕДНИЙ ран schema_version
+    "1.0" (§22-форсайт, НЕ analyze-1.0 inline-dict) через schema-filtered
+    `latest_run_for` (3b-i seam).
+
+    Returns:
+        • 200 {"status": "ready", "run_id", "created_at", "report"} — §22 готов
+          (report = SiteFinderReport.as_dict() из run.result).
+        • 202 {"status": "pending"} — ещё не посчитан, клиент продолжает поллить.
+
+    Graceful: нет рана → 202 pending (НЕ 404/500 на happy "not yet" path);
+    ошибка БД → 202 pending + warning (клиент ретраит, не видит 500).
+    """
+    try:
+        run = latest_run_for(db, cad_num, schema_version=_FORECAST_SCHEMA_VERSION)
+    except Exception:
+        # БД-сбой на read-only поллинге — не валим клиента 500-кой, отдаём pending.
+        logger.warning(
+            "forecast read failed for cad=%s — returning pending", cad_num, exc_info=True
+        )
+        response.status_code = 202
+        return {"status": "pending"}
+
+    if run is not None:
+        return {
+            "status": "ready",
+            "run_id": run.id,
+            "created_at": run.created_at,
+            "report": run.result,
+        }
+
+    # Форсайт ещё не посчитан (таска в работе или /analyze не вызывали) — клиент поллит.
+    response.status_code = 202
+    return {"status": "pending"}
+
+
 @router.post("/{cad_num}/analyze")
 def analyze_parcel(
     cad_num: str,
@@ -1177,6 +1236,10 @@ def analyze_parcel(
         str | None,
         Query(description="user_id для fallback на default-профиль пользователя"),
     ] = None,
+    horizon: Annotated[
+        int,
+        Query(description="Горизонт прогноза, мес (#995): один из {6, 12, 18}"),
+    ] = 12,
     body: Annotated[
         AnalyzeRequest | None,
         Body(description="Опциональное тело запроса: inline POI-веса (#201)"),
@@ -1205,7 +1268,23 @@ def analyze_parcel(
       через GET /fetch-status.
     - Дедупликация (через `find_active_on_demand_job`): параллельные запросы
       на тот же cad → один Celery job, оба клиента ждут.
+
+    §22-форсайт (3b-ii, #995): после успешного persist рана best-effort enqueue'им
+    `forecast_site_finder_report.delay(cad_num, horizon, ...)` — fire-and-forget,
+    провал Celery/Redis НЕ валит ответ (он уже успешен). `horizon` ∈ {6, 12, 18}
+    (иначе 422). Результат добавляется в ответ как `forecast` stub (additive); сам
+    §22-отчёт клиент забирает через GET /{cad}/forecast (202 pending → 200 ready).
     """
+    # Валидация горизонта прогноза (#995) — до любой работы с БД: иное → 422.
+    if horizon not in _ALLOWED_FORECAST_HORIZONS:
+        raise HTTPException(
+            status_code=422,
+            detail=(
+                f"horizon должен быть одним из {sorted(_ALLOWED_FORECAST_HORIZONS)}, "
+                f"получено: {horizon}"
+            ),
+        )
+
     # 1) Получить геометрию участка — GeoJSON строка через ST_AsGeoJSON
     row = (
         db.execute(
@@ -2691,6 +2770,28 @@ def analyze_parcel(
         created_by=x_authenticated_user,
     )
 
+    # §22-форсайт (3b-ii, #995): best-effort fire-and-forget enqueue после persist.
+    # Таска `forecast_site_finder_report` читает только что сохранённый analyze-1.0
+    # ран и в фоне (~30-180s) считает §22 SiteFinderReport ('1.0'). analyze НЕ ждёт
+    # её — возвращаемся сразу. Celery/Redis down НЕ должен валить ответ (он уже успешен:
+    # frontend зависит от формы). Зеркалит best-effort стиль find_or_enqueue_fetch.
+    # Lazy import — избегаем import-цикла api ↔ workers.tasks на старте.
+    try:
+        from app.workers.tasks.forecast import forecast_site_finder_report
+
+        forecast_site_finder_report.delay(cad_num, horizon, x_authenticated_user)
+        result_payload["forecast"] = {"status": "pending", "horizon": horizon}
+    except Exception:
+        # Enqueue не удался (broker недоступен и т.п.) — §9.x форсайт advisory,
+        # клиент узнаёт по status="unavailable" и не будет зря поллить /forecast.
+        logger.warning(
+            "forecast enqueue failed for cad=%s horizon=%s — analyze response unaffected",
+            cad_num,
+            horizon,
+            exc_info=True,
+        )
+        result_payload["forecast"] = {"status": "unavailable", "horizon": horizon}
+
     return result_payload
 
 

backend/tests/api/v1/test_parcels_forecast.py

@@ -0,0 +1,329 @@
+"""Тесты §22-форсайта в parcels API (3b-ii, #995).
+
+Покрывает:
+  • `horizon` query-param на POST /analyze: default 12; валидные {6,12,18} приняты;
+    невалидные (7, 0) → 422 ДО любой работы с БД.
+  • Best-effort enqueue после успешного analyze: `forecast_site_finder_report.delay`
+    вызван ОДИН раз с (cad_num, horizon, created_by); ответ несёт forecast.status="pending".
+  • Enqueue failure: `.delay` бросает → analyze всё равно 200, forecast.status="unavailable",
+    исключение НЕ всплывает.
+  • GET /{cad}/forecast: "1.0"-ран есть (mock latest_run_for) → 200 + status=ready + report;
+    рана нет → 202 + status=pending.
+
+Стратегия mock: как test_analyze_market_price.py — DB через dependency_overrides,
+тяжёлые сервисы патчим. Celery НЕ дёргаем: патчим `.delay` на реальной таске
+(lazy import внутри analyze_parcel резолвит атрибут модуля в момент вызова).
+RBAC обходится через settings.testing=True (conftest).
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+# ── Константы ─────────────────────────────────────────────────────────────────
+
+_CAD = "66:41:0204016:10"
+_WKT = "POLYGON((60.6 56.838, 60.61 56.838, 60.61 56.845, 60.6 56.845, 60.6 56.838))"
+_GEOJSON = '{"type":"Polygon","coordinates":[[[60.6,56.838],[60.61,56.838]]]}'
+
+# ── Mock factories (зеркалят test_analyze_market_price.py) ──────────────────────
+
+
+def _make_mapping(data: dict[str, Any]) -> MagicMock:
+    """Mock-строка (mapping) с __getitem__ + .get() для dict-like доступа."""
+    m = MagicMock()
+    m.__getitem__ = lambda self, k: data[k]
+    m.get = lambda k, default=None: data.get(k, default)
+    return m
+
+
+def _make_db_for_analyze(geom_found: bool = True) -> MagicMock:
+    """Mock DB Session, достаточный чтобы analyze_parcel дошёл до persist + enqueue."""
+    db = MagicMock()
+
+    geom_row = (
+        _make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"})
+        if geom_found
+        else None
+    )
+    wkt_row = _make_mapping({"wkt": _WKT}) if geom_found else None
+    district_row = _make_mapping(
+        {"district_name": "Октябрьский", "median_price_per_m2": 120000, "dist_to_center": 1500.0}
+    )
+    centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605})
+
+    def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
+        sql = " ".join(str(args[0]).split()) if args else ""
+        first_val: Any = None
+        if "AS geom_geojson" in sql:
+            first_val = geom_row
+        elif "AS wkt" in sql:
+            first_val = wkt_row
+        elif "AS median_price_per_m2" in sql and "district_name" in sql:
+            first_val = district_row
+        elif "AS lon" in sql and "AS lat" in sql:
+            first_val = centroid_row
+
+        r = MagicMock()
+        r.mappings.return_value.first.return_value = first_val
+        r.mappings.return_value.all.return_value = []
+        r.scalar.return_value = 0
+        return r
+
+    db.execute.side_effect = _execute_side_effect
+
+    ctx = MagicMock()
+    ctx.__enter__ = MagicMock(return_value=ctx)
+    ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = ctx
+
+    return db
+
+
+def _override_db(db: MagicMock):
+    def _get_db_override():
+        yield db
+
+    return _get_db_override
+
+
+# Тяжёлые внешние сервисы — заглушки (как в test_analyze_market_price.py).
+_PATCHES = [
+    patch("app.api.v1.parcels._fetch_air_quality_sync", return_value=None),
+    patch("app.api.v1.parcels._fetch_weather_sync", return_value=None),
+    patch("app.api.v1.parcels._fetch_seasonal_weather_sync", return_value=None),
+    patch(
+        "app.api.v1.parcels.get_quarter_dump_data",
+        return_value={
+            "nspd_zoning": None,
+            "nspd_zouit_overlaps": [],
+            "nspd_engineering_nearby": [],
+            "nspd_dump": {"available": False, "stale": False, "harvest_triggered": False},
+        },
+    ),
+    patch("app.api.v1.parcels.compute_velocity", return_value=None),
+    patch("app.api.v1.parcels.compute_gate_verdict", return_value={"verdict": "unknown"}),
+]
+
+
+def _start_patches() -> None:
+    for p in _PATCHES:
+        p.start()
+
+
+def _stop_patches() -> None:
+    for p in _PATCHES:
+        p.stop()
+
+
+# ── horizon query-param ─────────────────────────────────────────────────────────
+
+
+def test_horizon_default_12_and_enqueued() -> None:
+    """Без horizon → default 12 прокидывается в enqueue, forecast.status=pending."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    _start_patches()
+    delay_mock = MagicMock()
+    try:
+        with patch(
+            "app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
+        ):
+            client = TestClient(app)
+            resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
+            assert resp.status_code == 200, resp.text
+            body = resp.json()
+            assert body["forecast"] == {"status": "pending", "horizon": 12}
+            delay_mock.assert_called_once_with(_CAD, 12, None)
+    finally:
+        app.dependency_overrides.clear()
+        _stop_patches()
+
+
+def test_horizon_valid_values_accepted() -> None:
+    """horizon ∈ {6,12,18} принимается → 200 + enqueue с этим горизонтом."""
+    from app.core.db import get_db
+
+    for h in (6, 12, 18):
+        db = _make_db_for_analyze()
+        app.dependency_overrides[get_db] = _override_db(db)
+        _start_patches()
+        delay_mock = MagicMock()
+        try:
+            with patch(
+                "app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
+            ):
+                client = TestClient(app)
+                resp = client.post(f"/api/v1/parcels/{_CAD}/analyze?horizon={h}")
+                assert resp.status_code == 200, f"horizon={h}: {resp.text}"
+                assert resp.json()["forecast"] == {"status": "pending", "horizon": h}
+                delay_mock.assert_called_once_with(_CAD, h, None)
+        finally:
+            app.dependency_overrides.clear()
+            _stop_patches()
+
+
+def test_horizon_invalid_returns_422() -> None:
+    """horizon вне {6,12,18} (7, 0) → 422 ДО enqueue (delay не вызывается)."""
+    from app.core.db import get_db
+
+    for h in (7, 0):
+        db = _make_db_for_analyze()
+        app.dependency_overrides[get_db] = _override_db(db)
+        _start_patches()
+        delay_mock = MagicMock()
+        try:
+            with patch(
+                "app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
+            ):
+                client = TestClient(app)
+                resp = client.post(f"/api/v1/parcels/{_CAD}/analyze?horizon={h}")
+                assert resp.status_code == 422, f"horizon={h}: {resp.text}"
+                delay_mock.assert_not_called()
+        finally:
+            app.dependency_overrides.clear()
+            _stop_patches()
+
+
+def test_enqueue_passes_created_by_header() -> None:
+    """X-Authenticated-User прокидывается в .delay как created_by (3-й позиц. арг)."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    _start_patches()
+    delay_mock = MagicMock()
+    try:
+        with patch(
+            "app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
+        ):
+            client = TestClient(app)
+            resp = client.post(
+                f"/api/v1/parcels/{_CAD}/analyze",
+                headers={"X-Authenticated-User": "alice"},
+            )
+            assert resp.status_code == 200, resp.text
+            delay_mock.assert_called_once_with(_CAD, 12, "alice")
+    finally:
+        app.dependency_overrides.clear()
+        _stop_patches()
+
+
+# ── Enqueue failure ──────────────────────────────────────────────────────────
+
+
+def test_enqueue_failure_returns_200_unavailable() -> None:
+    """.delay бросает (broker down) → analyze всё равно 200, forecast.status=unavailable."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    _start_patches()
+    delay_mock = MagicMock(side_effect=RuntimeError("redis down"))
+    try:
+        with patch(
+            "app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
+        ):
+            client = TestClient(app)
+            resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
+            assert resp.status_code == 200, resp.text
+            body = resp.json()
+            assert body["forecast"] == {"status": "unavailable", "horizon": 12}
+            delay_mock.assert_called_once()
+    finally:
+        app.dependency_overrides.clear()
+        _stop_patches()
+
+
+def test_analyze_response_keys_unchanged_additive() -> None:
+    """forecast — additive: ключевые существующие блоки analyze остаются на месте."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    _start_patches()
+    try:
+        with patch("app.workers.tasks.forecast.forecast_site_finder_report.delay", MagicMock()):
+            client = TestClient(app)
+            resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
+            assert resp.status_code == 200, resp.text
+            body = resp.json()
+            # Существующие ключи (до 3b-ii) присутствуют + новый forecast добавлен.
+            for key in ("cad_num", "district", "score", "score_breakdown", "risks", "forecast"):
+                assert key in body, f"ключ {key} отсутствует в ответе analyze"
+    finally:
+        app.dependency_overrides.clear()
+        _stop_patches()
+
+
+# ── GET /{cad}/forecast ──────────────────────────────────────────────────────
+
+
+def test_get_forecast_ready_when_run_present() -> None:
+    """Есть "1.0"-ран → 200 + status=ready + run_id + report=run.result."""
+    from app.core.db import get_db
+
+    report_dict = {"schema_version": "1.0", "deficit": -1.0, "moi": 117}
+    fake_run = MagicMock()
+    fake_run.id = 66
+    fake_run.created_at = dt.datetime(2026, 6, 5, 7, 32, 0)
+    fake_run.result = report_dict
+
+    db = MagicMock()
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        with patch("app.api.v1.parcels.latest_run_for", return_value=fake_run) as lrf:
+            client = TestClient(app)
+            resp = client.get(f"/api/v1/parcels/{_CAD}/forecast")
+            assert resp.status_code == 200, resp.text
+            body = resp.json()
+            assert body["status"] == "ready"
+            assert body["run_id"] == 66
+            assert body["report"] == report_dict
+            # Читаем именно §22-схему "1.0" (3b-i schema-filtered seam).
+            _, kwargs = lrf.call_args
+            assert kwargs.get("schema_version") == "1.0"
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_get_forecast_pending_when_no_run() -> None:
+    """Рана нет (latest_run_for → None) → 202 + status=pending."""
+    from app.core.db import get_db
+
+    db = MagicMock()
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        with patch("app.api.v1.parcels.latest_run_for", return_value=None):
+            client = TestClient(app)
+            resp = client.get(f"/api/v1/parcels/{_CAD}/forecast")
+            assert resp.status_code == 202, resp.text
+            assert resp.json() == {"status": "pending"}
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_get_forecast_db_error_returns_pending_not_500() -> None:
+    """Ошибка БД на read-only поллинге → 202 pending (graceful, не 500)."""
+    from app.core.db import get_db
+
+    db = MagicMock()
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        with patch(
+            "app.api.v1.parcels.latest_run_for", side_effect=RuntimeError("db gone")
+        ):
+            client = TestClient(app)
+            resp = client.get(f"/api/v1/parcels/{_CAD}/forecast")
+            assert resp.status_code == 202, resp.text
+            assert resp.json() == {"status": "pending"}
+    finally:
+        app.dependency_overrides.clear()