backend/app/services/forecasting/macro_series.py

@@ -371,12 +371,20 @@ def _month_grid(start: date, end: date) -> list[date]:
 
 
 def _query_key_rate_monthly(db: Session, *, since: date) -> dict[date, float]:
-    """Ресэмпл дневного key_rate (region 'rf') → {month1st: value}. Graceful → {}."""
+    """Ресэмпл дневного key_rate (region 'rf') → {month1st: value}. Graceful → {}.
+
+    SAVEPOINT (#2464 cluster A finding #2): `db` — общая §22-сессия отчёта; при сбое
+    этого запроса БЕЗ SAVEPOINT транзакция Postgres остаётся aborted и следующие
+    запросы get_monthly_macro (inflation, mortgage) + все ПОСЛЕДУЮЩИЕ §9.x-слои на
+    той же сессии тоже падают. `with db.begin_nested():` откатывает ТОЛЬКО этот
+    SAVEPOINT (ROLLBACK TO SAVEPOINT), внешняя транзакция остаётся рабочей.
+    """
     try:
-        rows = db.execute(
-            _KEY_RATE_MONTHLY_SQL,
-            {"itype": "key_rate", "region": "rf", "since": since},
-        ).all()
+        with db.begin_nested():
+            rows = db.execute(
+                _KEY_RATE_MONTHLY_SQL,
+                {"itype": "key_rate", "region": "rf", "since": since},
+            ).all()
     except Exception:
         logger.exception("get_monthly_macro: key_rate query failed")
         return {}
@@ -389,9 +397,14 @@ def _query_inflation_monthly(db: Session, *, since: date) -> dict[date, float]:
     Ряд УЖЕ месячный (obs_date = 1-е число, залит cbr_macro_sync) — берём как есть
     через reuse get_macro_series (свой SQL не пишем). _month_start — страховка.
     Сбой/пустой ряд → {} (НЕ crash), inflation_yoy тогда None по всей сетке.
+
+    SAVEPOINT (#2464 cluster A finding #2): см. `_query_key_rate_monthly` — та же
+    общая §22-сессия, тот же риск отравления транзакции для последующих запросов
+    (mortgage-поля + §9.x-слои). `with db.begin_nested():` изолирует сбой в SAVEPOINT.
     """
     try:
-        series = get_macro_series(db, "inflation_yoy", region="rf", since=since)
+        with db.begin_nested():
+            series = get_macro_series(db, "inflation_yoy", region="rf", since=since)
     except Exception:
         logger.exception("get_monthly_macro: inflation_yoy query failed")
         return {}
@@ -404,11 +417,20 @@ def _query_mortgage_monthly(db: Session, *, since: date) -> dict[str, dict[date,
     Возвращает {field: {month1st: value}}. obs_date уже нормализован к 1-му числу
     в backfill, но _month_start применяем повторно (страховка). Сбой одного ряда
     не валит остальные (graceful: пустой подсловарь).
+
+    SAVEPOINT (#2464 cluster A finding #2): все 5 полей читаются на ОДНОЙ `db`-Session
+    (get_monthly_macro вызывается внутри общей §22-сессии отчёта). Без SAVEPOINT сбой
+    ОДНОГО поля оставляет транзакцию Postgres aborted — каждое СЛЕДУЮЩЕЕ поле в этом
+    же цикле тоже падает (хотя его данные были бы доступны), а `except` здесь молча
+    отдаёт [] по каждому, маскируя каскад под «нормальную» построчную деградацию.
+    `with db.begin_nested():` — SAVEPOINT на КАЖДОЕ поле: сбой откатывает только его
+    SAVEPOINT (ROLLBACK TO SAVEPOINT), сессия остаётся рабочей для следующего поля.
     """
     out: dict[str, dict[date, float]] = {}
     for indicator_type, field in _MORTGAGE_FIELDS:
         try:
-            series = get_macro_series(db, indicator_type, region="sverdl", since=since)
+            with db.begin_nested():
+                series = get_macro_series(db, indicator_type, region="sverdl", since=since)
         except Exception:
             logger.exception("get_monthly_macro: mortgage series %s failed", indicator_type)
             series = []

backend/app/services/forecasting/orchestrator.py

@@ -124,7 +124,7 @@ def _primary_horizon(horizons: Sequence[int]) -> int:
     return horizons[0] if horizons else _PREFERRED_PRIMARY_HORIZON
 
 
-def _safe_call(label: str, fn: Any) -> Any:
+def _safe_call(label: str, db: Session, fn: Any) -> Any:
     """Вызвать §9.x-сервис graceful: сбой → None + logger.exception (не crash отчёта).
 
     Зеркало product_scoring._safe_call: любой §9.x-слой может бросить (тонкие данные / нет
@@ -133,15 +133,29 @@ def _safe_call(label: str, fn: Any) -> Any:
     широкий Exception (изоляция одного слоя от отчёта) с ОБЯЗАТЕЛЬНЫМ logger.exception —
     НЕ молчаливое глотание. §9.x уже graceful внутри; это belt-and-suspenders на шве.
 
+    SAVEPOINT (#2464 cluster A finding #1): все §9.x-слои шарят ОДИН `db`-Session на
+    отчёт (module docstring `forecast_request_cache.py`). Без обёртки сбойный
+    `db.execute` внутри слоя оставляет транзакцию Postgres в состоянии aborted
+    («current transaction is aborted, commands ignored until end of transaction
+    block») — КАЖДЫЙ последующий слой на той же сессии тоже падает, хотя его данные
+    были бы доступны. `with db.begin_nested():` заводит SAVEPOINT НА ВЕСЬ вызов слоя
+    (слой может делать несколько `db.execute` внутри себя — например §9.6 внутри
+    §9.8/§11); при исключении SAVEPOINT откатывается автоматически (ROLLBACK TO
+    SAVEPOINT), внешняя транзакция остаётся рабочей для следующего слоя. НЕ
+    `db.rollback()` — тот откатил бы ВСЮ внешнюю транзакцию (см. `backend.md` §
+    SAVEPOINT pattern, established anti-pattern).
+
     Args:
         label: имя слоя для лога (диагностика какой §9.x-вызов деградировал).
+        db: общая §22-сессия (для SAVEPOINT вокруг вызова слоя).
         fn: нулевой-аргумент thunk вокруг §9.x-вызова.
 
     Returns:
         Результат `fn()` или None при исключении.
     """
     try:
-        return fn()
+        with db.begin_nested():
+            return fn()
     except Exception:
         logger.exception("orchestrator: §9.x layer %s failed → section degraded", label)
         return None
@@ -334,21 +348,25 @@ def _build_site_finder_report_impl(
     # ── 2. §9.x-слои, каждый graceful через _safe_call (ГЕТЕРОГЕННЫЕ сигнатуры) ──
     market_metrics = _safe_call(
         "market_metrics",
+        db,
         lambda: compute_market_metrics(db, district=district, premise_kind=_PREMISE_KIND),
     )
     supply_rows = _safe_call(
         "supply_layers",
+        db,
         lambda: compute_all_layers(db, district=district, premise_kind=_PREMISE_KIND),
     )
     supply_layers = _summarize_supply_layers(supply_rows)
     future_supply = _safe_call(
         "future_supply",
+        db,
         lambda: compute_future_supply_pressure(
             db, district=district, horizon_months=primary, premise_kind=_PREMISE_KIND
         ),
     )
     forecasts = _safe_call(
         "demand_supply_forecast",
+        db,
         lambda: compute_demand_supply_forecast(
             db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
         ),
@@ -359,6 +377,7 @@ def _build_site_finder_report_impl(
     # _safe_call оборачивает любой сбой → None → штатно деградируем (collapse=False).
     scenarios_result = _safe_call(
         "scenarios",
+        db,
         lambda: compute_scenarios(
             db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
         ),
@@ -371,18 +390,21 @@ def _build_site_finder_report_impl(
         scenarios, scenarios_collapsed, scenarios_collapse_reason = scenarios_result
     product_scores = _safe_call(
         "score_card",
+        db,
         lambda: compute_score_card(
             db, spec=spec, district=district, cad_num=cad_num, horizon_months=primary
         ),
     )
     special_indices = _safe_call(
         "special_indices",
+        db,
         lambda: compute_special_indices(
             db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
         ),
     )
     recommendation_overlay = _safe_call(
         "forecast_overlay",
+        db,
         lambda: build_forecast_overlay(
             db,
             district=district,
@@ -393,7 +415,7 @@ def _build_site_finder_report_impl(
     )
 
     # ── Макро-свежесть (audit MEDIUM): только лог; проводка в отчёт — 3b ─────────
-    macro = _safe_call("monthly_macro", lambda: get_monthly_macro(db))
+    macro = _safe_call("monthly_macro", db, lambda: get_monthly_macro(db))
     macro_as_of = _macro_as_of(macro)
     if macro_as_of is not None:
         logger.info(

backend/app/services/forecasting/sales_series.py

@@ -549,6 +549,12 @@ def _query_source_a(
 
     Graceful → {} при сбое/пустых данных. price_bucket в spec для Source A
     игнорируется (агрегат не несёт per-lot цены) — фиксируется логом.
+
+    SAVEPOINT (#2464 cluster A finding #4): `db` — общая §22-сессия отчёта (может
+    переиспользоваться другими §9.x-слоями после этого вызова). Без SAVEPOINT сбойный
+    `db.execute` оставляет транзакцию Postgres aborted — все ПОСЛЕДУЮЩИЕ запросы на
+    той же сессии тоже падают. `with db.begin_nested():` откатывает ТОЛЬКО этот
+    SAVEPOINT при сбое, оставляя внешнюю транзакцию рабочей.
     """
     if spec.price_bucket is not None:
         logger.info(
@@ -567,7 +573,8 @@ def _query_source_a(
         "room_bucket": spec.room_bucket,
     }
     try:
-        rows = db.execute(_SOURCE_A_SQL, params).mappings().all()
+        with db.begin_nested():
+            rows = db.execute(_SOURCE_A_SQL, params).mappings().all()
     except Exception:
         logger.exception("build_sales_series: source A query failed")
         return {}
@@ -582,6 +589,9 @@ def _query_source_b(
     Graceful → {} при сбое/пустых данных. Передаёт bucket-пороги/-метки в SQL
     (зеркало pure-helpers), чтобы room×area / price сегментация считалась тем же
     правилом и в БД, и в Python.
+
+    SAVEPOINT (#2464 cluster A finding #4): см. `_query_source_a` — та же общая
+    §22-сессия, тот же риск отравления транзакции для последующих слоёв/запросов.
     """
     # district (админ-имя ЕКБ) → набор informal микро (objective_lots хранит микро).
     # None → EKB-wide (без district-фильтра).
@@ -613,7 +623,8 @@ def _query_source_b(
         "p_unknown": PRICE_BUCKET_UNKNOWN,
     }
     try:
-        rows = db.execute(_SOURCE_B_SQL, params).mappings().all()
+        with db.begin_nested():
+            rows = db.execute(_SOURCE_B_SQL, params).mappings().all()
     except Exception:
         logger.exception("build_sales_series: source B query failed")
         return {}

backend/app/services/forecasting/special_indices.py

@@ -994,16 +994,29 @@ def _query_parcel_centroid(db: Session, *, cad_num: str) -> tuple[float, float]
 
     Нет геометрии / сбой → None (гео-веса упадут на floor; overlap считается по остальным
     осям, индекс НЕ деградирует целиком). Параметризовано (psycopg v3). Детерминированно.
+
+    SAVEPOINT НА ВНУТРЕННЕМ swallow-сайте (#2464 cluster A, RELEASE-trap): этот helper
+    зовётся из `_build_cannibalization` — builder, обёрнутый внешним SAVEPOINT в
+    `compute_special_indices._run`. БЕЗ собственного `db.begin_nested():` сбойный
+    `db.execute` тут проглатывается локально (→ None), оставляя транзакцию Postgres
+    aborted; тогда ВНЕШНИЙ `_run`-savepoint выходит «нормально» и делает RELEASE
+    SAVEPOINT, а RELEASE в aborted-tx САМ падает (Postgres в aborted допускает только
+    ROLLBACK / ROLLBACK TO SAVEPOINT) — внешний except ловит уже RELEASE-ошибку, но
+    транзакция так и не откачена → отравление каскадит в следующий §25-индекс. Savepoint
+    ИМЕННО ЗДЕСЬ (в точке перехвата) откатывает сбой (ROLLBACK TO SAVEPOINT), поэтому
+    внешний RELEASE проходит. Тот же принцип, что saturation.py: savepoint у db.execute
+    в точке, где ошибка ловится.
     """
     try:
-        row = (
-            db.execute(
-                _PARCEL_CENTROID_SQL,
-                {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)},
+        with db.begin_nested():
+            row = (
+                db.execute(
+                    _PARCEL_CENTROID_SQL,
+                    {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)},
+                )
+                .mappings()
+                .first()
             )
-            .mappings()
-            .first()
-        )
     except Exception:
         logger.exception("cannibalization: centroid query failed for cad_num=%s", cad_num)
         return None
@@ -1657,9 +1670,20 @@ def compute_special_indices(
     segment = spec.as_dict()
 
     def _run(key: str, builder: Any) -> SpecialIndex:
-        """Выполнить builder одного индекса в собственном try/except (graceful)."""
+        """Выполнить builder одного индекса в собственном try/except (graceful).
+
+        SAVEPOINT (#2464 cluster A finding #3): все шесть builder'ов делят ОДНУ `db`-
+        Session (общая сессия §22-отчёта). Без SAVEPOINT сбойный `db.execute` внутри
+        builder'а оставляет транзакцию Postgres aborted — КАЖДЫЙ следующий индекс на
+        той же сессии тоже падает (хотя его данные были бы доступны), что маскируется
+        под шесть независимых деградаций. `with db.begin_nested():` заводит SAVEPOINT
+        на ВЕСЬ builder() (некоторые builder'ы делают несколько db.execute внутри
+        себя) — сбой откатывает только его SAVEPOINT, сессия остаётся рабочей для
+        следующего индекса.
+        """
         try:
-            return builder()  # type: ignore[no-any-return]
+            with db.begin_nested():
+                return builder()  # type: ignore[no-any-return]
         except Exception:
             # Сбой одного индекса НЕ валит карточку: деградация-None, остальные считаются.
             logger.exception(

backend/app/services/site_finder/district_resolver.py

@@ -68,12 +68,22 @@ def _admin_names(db: Session) -> frozenset[str]:
 
     На ошибке БД возвращает пустой set (logged) — резолвер тогда трактует ЛЮБОЙ
     вход как raw-микро (legacy pass-through), что безопаснее, чем уронить /analyze.
+
+    SAVEPOINT НА ВНУТРЕННЕМ swallow-сайте (#2464 cluster A, RELEASE-trap): резолвер
+    зовётся §22-форсайт-билдерами (sales_series `_query_source_a/_b`,
+    `_query_artificial_demand`) на ОБЩЕЙ сессии отчёта, которые сами обёрнуты во
+    внешние SAVEPOINT'ы. БЕЗ `db.begin_nested():` здесь сбойный `db.execute`
+    проглатывается (→ frozenset()), оставляя транзакцию aborted; тогда внешний
+    SAVEPOINT при RELEASE падает (RELEASE запрещён в aborted-tx) и отравление
+    каскадит. Savepoint в точке перехвата откатывает сбой (ROLLBACK TO SAVEPOINT),
+    сессия остаётся рабочей.
     """
     global _ADMIN_NAMES_CACHE
     if _ADMIN_NAMES_CACHE is not None:
         return _ADMIN_NAMES_CACHE
     try:
-        rows = db.execute(_ADMIN_NAMES_SQL).scalars().all()
+        with db.begin_nested():
+            rows = db.execute(_ADMIN_NAMES_SQL).scalars().all()
         _ADMIN_NAMES_CACHE = frozenset(str(r) for r in rows if r)
     except Exception:
         # Не кешируем пустой результат-от-сбоя: следующий вызов попробует снова.
@@ -122,8 +132,13 @@ def resolve_objective_districts(db: Session, district: str | None) -> list[str]
         return [district]
 
     # district — официальное админ-имя: разворачиваем в чистые ЕКБ-микро.
+    # SAVEPOINT на внутреннем swallow-сайте (#2464 cluster A, RELEASE-trap): см.
+    # `_admin_names` — тот же общий-сессионный вызов из §22-билдеров под внешним
+    # SAVEPOINT. `db.begin_nested():` откатывает сбой в точке перехвата, чтобы
+    # внешний RELEASE не упал в aborted-tx (RELEASE запрещён в aborted).
     try:
-        rows = db.execute(_MICROS_FOR_ADMIN_SQL, {"d": district}).scalars().all()
+        with db.begin_nested():
+            rows = db.execute(_MICROS_FOR_ADMIN_SQL, {"d": district}).scalars().all()
         micros = [str(m) for m in rows]
     except Exception:
         logger.exception(

backend/tests/services/forecasting/test_macro_series.py

@@ -13,18 +13,25 @@ psycopg v3 правило проверяется явно: bind-параметр
 
 from __future__ import annotations
 
+import contextlib
 import datetime as dt
 import os
+from collections.abc import Iterator
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
 from app.services.forecasting.macro_series import (
     _CONFOUNDER_DATES,
+    _MORTGAGE_FIELDS,
     _REGIME_EPS_PP,
     MonthlyMacro,
     _carry_forward,
     _month_grid,
+    _query_inflation_monthly,
+    _query_key_rate_monthly,
+    _query_mortgage_monthly,
     _shift_months,
     classify_regime,
     get_monthly_macro,
@@ -476,3 +483,110 @@ class TestGetMonthlyMacroGraceful:
             out = get_monthly_macro(db, months_back=1)
         assert len(out) == 2
         assert all(m.mortgage_rate_weighted is None for m in out)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# SAVEPOINT (#2464 cluster A finding #2, `_query_mortgage_monthly`): сбой ОДНОГО
+# mortgage-поля НЕ должен «отравить» общую §22-сессию для остальных 4 полей / для
+# последующих §9.x-слоёв, переиспользующих ту же `db`-Session. Без `db.begin_nested()`
+# сбойный `db.execute` (внутри `get_macro_series`) оставляет транзакцию Postgres
+# aborted — каждый ПОСЛЕДУЮЩИЙ execute на той же сессии тоже падал бы.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class _FakeMortgageResult:
+    """Мини-результат db.execute() — покрывает `.all()` (см. `get_macro_series`)."""
+
+    def all(self) -> list[Any]:
+        return []
+
+
+class _PoisonableSession:
+    """Fake Session, симулирующая Postgres 'aborted transaction' семантику.
+
+    `execute()` бросает РОВНО один раз (на первое поле) и, как реальный Postgres,
+    «отравляет» сессию: БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` (даже
+    несвязанный) тоже бросает — `current transaction is aborted`. `begin_nested()`
+    зеркалит `Session.begin_nested()`: при исключении откатывает ТОЛЬКО SAVEPOINT
+    (сбрасывает poison-флаг) и пробрасывает исключение дальше — как настоящий
+    `ROLLBACK TO SAVEPOINT`, НЕ общий `db.rollback()` (откатил бы всю внешнюю
+    транзакцию).
+    """
+
+    def __init__(self) -> None:
+        self.poisoned = False
+        self._raised = False
+        self.calls = 0
+
+    def execute(self, *args: Any, **kwargs: Any) -> _FakeMortgageResult:
+        self.calls += 1
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        if not self._raised:
+            self._raised = True
+            self.poisoned = True
+            raise RuntimeError("simulated mortgage field query failure")
+        return _FakeMortgageResult()
+
+    @contextlib.contextmanager
+    def begin_nested(self) -> Iterator[None]:
+        try:
+            yield
+        except Exception:
+            # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT,
+            # сессия остаётся рабочей для следующего поля.
+            self.poisoned = False
+            raise
+
+
+class TestQueryMortgageMonthlySavepoint:
+    """`_query_mortgage_monthly` оборачивает КАЖДОЕ поле в `db.begin_nested()`."""
+
+    def test_first_field_failure_leaves_session_usable_for_later_fields(self) -> None:
+        db = _PoisonableSession()
+        out = _query_mortgage_monthly(db, since=dt.date(2024, 1, 1))
+        # все 5 полей присутствуют в результате (контракт функции не изменён).
+        assert set(out.keys()) == {field for _itype, field in _MORTGAGE_FIELDS}
+        # сбойное (первое) поле → пустой fallback-словарь, НЕ crash.
+        first_field = _MORTGAGE_FIELDS[0][1]
+        assert out[first_field] == {}
+        # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ
+        # сбойного поля проходит без 'current transaction aborted' (не poisoned).
+        db.execute("SELECT 1")
+
+
+class TestQueryKeyRateMonthlySavepoint:
+    """`_query_key_rate_monthly` оборачивает db.execute в `db.begin_nested()`.
+
+    Тот же poisoning-класс, что и mortgage-цикл: key_rate — ПЕРВЫЙ запрос
+    get_monthly_macro; без SAVEPOINT его сбой отравил бы транзакцию для inflation +
+    mortgage + всех последующих §9.x-слоёв на общей §22-сессии.
+    """
+
+    def test_db_failure_leaves_session_usable(self) -> None:
+        db = _PoisonableSession()
+        out = _query_key_rate_monthly(db, since=dt.date(2024, 1, 1))
+        # сбой запроса → пустой ресэмпл (fallback, НЕ crash).
+        assert out == {}
+        # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ
+        # сбоя проходит без 'current transaction aborted' (не poisoned).
+        db.execute("SELECT 1")
+
+
+class TestQueryInflationMonthlySavepoint:
+    """`_query_inflation_monthly` оборачивает db.execute (через get_macro_series) в SAVEPOINT.
+
+    Тот же poisoning-класс: inflation — ВТОРОЙ запрос get_monthly_macro; без SAVEPOINT
+    его сбой отравил бы транзакцию для mortgage-полей + последующих §9.x-слоёв.
+    """
+
+    def test_db_failure_leaves_session_usable(self) -> None:
+        db = _PoisonableSession()
+        out = _query_inflation_monthly(db, since=dt.date(2024, 1, 1))
+        # сбой запроса → пустой ряд (fallback, НЕ crash).
+        assert out == {}
+        # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ
+        # сбоя проходит без 'current transaction aborted' (не poisoned).
+        db.execute("SELECT 1")

backend/tests/services/forecasting/test_orchestrator.py

@@ -371,3 +371,101 @@ class TestNormalizeDistrict:
         # list / int / прочие неподдерживаемые типы → None + warning.
         assert _normalize_district([1, 2, 3]) is None
         assert _normalize_district(42) is None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# SAVEPOINT (#2464 cluster A finding #1, orchestrator.py `_safe_call`): один сбойный
+# §9.x-слой НЕ должен «отравить» общую §22-сессию для ОСТАЛЬНЫХ слоёв. Без
+# `db.begin_nested()` сбойный `db.execute` внутри слоя оставляет транзакцию Postgres
+# aborted («current transaction is aborted, commands ignored until end of
+# transaction block») — каждый ПОСЛЕДУЮЩИЙ слой на той же сессии тоже падал бы.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class _FakeResult:
+    """Мини-результат db.execute() — покрывает `.mappings().first()`/`.all()`."""
+
+    def mappings(self) -> _FakeResult:
+        return self
+
+    def first(self) -> None:
+        return None
+
+    def all(self) -> list[Any]:
+        return []
+
+
+class _PoisonableSession:
+    """Fake Session, симулирующая Postgres 'aborted transaction' семантику.
+
+    `execute()` бросает РОВНО один раз (на первый вызов) и, как реальный Postgres,
+    «отравляет» сессию: БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` (даже
+    совершенно несвязанный) тоже бросает — `current transaction is aborted`.
+    `begin_nested()` зеркалит `Session.begin_nested()`: при исключении откатывает
+    ТОЛЬКО SAVEPOINT (сбрасывает poison-флаг) и пробрасывает исключение дальше — как
+    настоящий `ROLLBACK TO SAVEPOINT`, НЕ общий `db.rollback()` (который откатил бы
+    всю внешнюю транзакцию — established anti-pattern, `backend.md`).
+    """
+
+    def __init__(self) -> None:
+        self.poisoned = False
+        self._raised = False
+        self.calls = 0
+
+    def execute(self, *args: Any, **kwargs: Any) -> _FakeResult:
+        self.calls += 1
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        if not self._raised:
+            self._raised = True
+            self.poisoned = True
+            raise RuntimeError("simulated §9.x layer query failure")
+        return _FakeResult()
+
+    @contextlib.contextmanager
+    def begin_nested(self) -> Iterator[None]:
+        try:
+            yield
+        except Exception:
+            # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT,
+            # сессия остаётся рабочей для следующего слоя.
+            self.poisoned = False
+            raise
+
+
+class TestSafeCallSavepointDoesNotPoisonSession:
+    """`_safe_call` оборачивает вызов слоя в `db.begin_nested()` (SAVEPOINT)."""
+
+    def test_layer_db_failure_leaves_session_usable_for_later_layers(self) -> None:
+        db = _PoisonableSession()
+        mocks = _patched_layers()
+        # market_metrics — единственный слой в этом тесте, реально трогающий `db`:
+        # его мок сам вызывает db.execute (через _PoisonableSession бросает один раз).
+        # Остальные слои остаются sentinel-моками из _patched_layers() (db не трогают).
+        mocks["compute_market_metrics"] = MagicMock(
+            side_effect=lambda *a, **kw: db.execute("SELECT 1")
+        )
+        captured: dict[str, Any] = {}
+
+        def _fake_assemble(analyze: dict[str, Any], **kwargs: Any) -> SiteFinderReport:
+            captured.update(kwargs)
+            return SiteFinderReport()
+
+        with _patch_layers(mocks, _fake_assemble):
+            report = build_site_finder_report(
+                db,
+                analyze=_analyze([{"obj_class": "Комфорт"}]),
+                cad_num="66:41:0702048:27",
+                district="Ленинский",
+                horizons=[6, 12, 18],
+            )
+        assert isinstance(report, SiteFinderReport)
+        # упавший слой → None в слоте (fallback), остальные слои — sentinel'ы, целы.
+        assert captured["market_metrics"] is None
+        assert captured["scenarios"] == ["SC"]
+        assert captured["forecasts"] == ["F6", "F12"]
+        # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ
+        # сбойного слоя проходит без 'current transaction aborted' (не poisoned).
+        db.execute("SELECT 1")

backend/tests/services/forecasting/test_sales_series.py

@@ -15,10 +15,12 @@ psycopg v3 правило проверяется явно: bind-параметр
 
 from __future__ import annotations
 
+import contextlib
 import datetime as dt
 import math
 import os
 from collections.abc import Iterator
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -667,6 +669,85 @@ class TestBuildSalesSeriesGraceful:
         assert out.units == [0]
 
 
+# ──────────────────────────────────────────────────────────────────────────────
+# SAVEPOINT (#2464 cluster A finding #4, `_query_source_a`/`_query_source_b`): сбой
+# db.execute НЕ должен «отравить» общую §22-сессию для ПОСЛЕДУЮЩИХ слоёв/запросов,
+# переиспользующих ту же `db`-Session. Без `db.begin_nested()` сбойный `db.execute`
+# оставляет транзакцию Postgres aborted — каждый ПОСЛЕДУЮЩИЙ execute на той же сессии
+# тоже падал бы.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class _FakeSalesResult:
+    """Мини-результат db.execute() — покрывает `.mappings().all()`."""
+
+    def mappings(self) -> _FakeSalesResult:
+        return self
+
+    def all(self) -> list[Any]:
+        return []
+
+
+class _PoisonableSession:
+    """Fake Session, симулирующая Postgres 'aborted transaction' семантику.
+
+    `execute()` бросает РОВНО один раз и, как реальный Postgres, «отравляет» сессию:
+    БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()` (даже несвязанный) тоже
+    бросает — `current transaction is aborted`. `begin_nested()` зеркалит
+    `Session.begin_nested()`: при исключении откатывает ТОЛЬКО SAVEPOINT (сбрасывает
+    poison-флаг) и пробрасывает исключение дальше — как настоящий `ROLLBACK TO
+    SAVEPOINT`, НЕ общий `db.rollback()` (откатил бы всю внешнюю транзакцию).
+    """
+
+    def __init__(self) -> None:
+        self.poisoned = False
+        self._raised = False
+        self.calls = 0
+
+    def execute(self, *args: Any, **kwargs: Any) -> _FakeSalesResult:
+        self.calls += 1
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        if not self._raised:
+            self._raised = True
+            self.poisoned = True
+            raise RuntimeError("simulated sales query failure")
+        return _FakeSalesResult()
+
+    @contextlib.contextmanager
+    def begin_nested(self) -> Iterator[None]:
+        try:
+            yield
+        except Exception:
+            # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT,
+            # сессия остаётся рабочей для последующих запросов.
+            self.poisoned = False
+            raise
+
+
+class TestSourceQuerySavepointDoesNotPoisonSession:
+    """`_query_source_a`/`_query_source_b` оборачивают db.execute в SAVEPOINT."""
+
+    def test_source_a_failure_leaves_session_usable(self) -> None:
+        db = _PoisonableSession()
+        out = build_sales_series(db, spec=SegmentSpec(), source="corpus_room_month", months_back=1)
+        # сбой запроса → zero-filled сетка, low confidence (fallback, НЕ crash).
+        assert out.confidence == "low"
+        assert out.units == [0, 0]
+        # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ
+        # сбоя проходит без 'current transaction aborted' (не poisoned).
+        db.execute("SELECT 1")
+
+    def test_source_b_failure_leaves_session_usable(self) -> None:
+        db = _PoisonableSession()
+        out = build_sales_series(db, spec=SegmentSpec(), source="objective_lots", months_back=1)
+        assert out.confidence == "low"
+        assert out.units == [0, 0]
+        db.execute("SELECT 1")
+
+
 class TestDistrictResolution:
     """Step 2: spec.district (админ-имя ЕКБ) резолвится в МИКРО-набор в SQL-фильтре.
 

backend/tests/services/forecasting/test_special_indices.py

@@ -16,6 +16,7 @@ test_market_metrics.
 
 from __future__ import annotations
 
+import contextlib
 from datetime import date
 from typing import Any
 from unittest.mock import MagicMock, patch
@@ -65,6 +66,7 @@ from app.services.forecasting.special_indices import (
     _price_bucket_to_band,
     _price_overlap,
     _query_artificial_demand,
+    _query_parcel_centroid,
     _timing_overlap,
     _unit_mix_similarity,
     _void_index,
@@ -1570,6 +1572,178 @@ class TestComputeSpecialIndicesGraceful:
         assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.25)
 
 
+# ──────────────────────────────────────────────────────────────────────────────
+# SAVEPOINT (#2464 cluster A finding #3, `_run` внутри `compute_special_indices`):
+# сбой ОДНОГО §25-индекса НЕ должен «отравить» общую §22-сессию для ОСТАЛЬНЫХ
+# индексов/слоёв, переиспользующих ту же `db`-Session. Без `db.begin_nested()`
+# сбойный `db.execute` (внутри `_query_artificial_demand`) оставляет транзакцию
+# Postgres aborted — каждый ПОСЛЕДУЮЩИЙ execute на той же сессии тоже падал бы.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class _FakeIndicesResult:
+    """Мини-результат db.execute() — покрывает `.mappings().first()`."""
+
+    def mappings(self) -> _FakeIndicesResult:
+        return self
+
+    def first(self) -> None:
+        return None
+
+
+class _PoisonableSession:
+    """Fake Session, симулирующая Postgres 'aborted transaction' семантику.
+
+    `execute()` бросает РОВНО один раз (Artificial Demand SQL) и, как реальный
+    Postgres, «отравляет» сессию: БЕЗ SAVEPOINT-отката любой ПОСЛЕДУЮЩИЙ `execute()`
+    (даже несвязанный) тоже бросает — `current transaction is aborted`.
+    `begin_nested()` зеркалит `Session.begin_nested()`: при исключении откатывает
+    ТОЛЬКО SAVEPOINT (сбрасывает poison-флаг) и пробрасывает исключение дальше — как
+    настоящий `ROLLBACK TO SAVEPOINT`, НЕ общий `db.rollback()`.
+    """
+
+    def __init__(self) -> None:
+        self.poisoned = False
+        self._raised = False
+        self.calls = 0
+
+    def execute(self, *args: Any, **kwargs: Any) -> _FakeIndicesResult:
+        self.calls += 1
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        if not self._raised:
+            self._raised = True
+            self.poisoned = True
+            raise RuntimeError("simulated artificial_demand query failure")
+        return _FakeIndicesResult()
+
+    @contextlib.contextmanager
+    def begin_nested(self):  # type: ignore[no-untyped-def]
+        try:
+            yield
+        except Exception:
+            # ROLLBACK TO SAVEPOINT — откатывает только вложенный SAVEPOINT,
+            # сессия остаётся рабочей для следующего индекса.
+            self.poisoned = False
+            raise
+
+
+class TestRunSavepointDoesNotPoisonSession:
+    """`_run` (внутренний gate compute_special_indices) оборачивает builder в SAVEPOINT."""
+
+    def test_artificial_demand_db_failure_leaves_session_usable(self) -> None:
+        db = _PoisonableSession()
+        # district=None/cad_num=None → Cannibalization/Competitor Strength деградируют
+        # БЕЗ обращения к db (см. test_no_cad_num_degrades_competitor_indices); own
+        # portfolio патчим в [] (own_developer_ids-гейт не задействуем) — единственный
+        # builder, реально трогающий db в этом тесте, это Artificial Demand.
+        with _full_stack_patch(), patch(f"{_MOD}.get_own_portfolio", return_value=[]):
+            card = compute_special_indices(db, spec=_SPEC, district=None, cad_num=None)
+        # упавший индекс деградировал в fallback (НЕ крашнул карточку).
+        assert card.indices[KEY_ARTIFICIAL_DEMAND].method == _METHOD_UNAVAILABLE
+        assert card.indices[KEY_ARTIFICIAL_DEMAND].value is None
+        assert len(card.indices) == 6
+        # КЛЮЧЕВОЙ ассерт savepoint-фикса: сессия НЕ отравлена — db.execute ПОСЛЕ
+        # сбойного индекса проходит без 'current transaction aborted' (не poisoned).
+        db.execute("SELECT 1")
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# RELEASE-SAVEPOINT trap (#2464 cluster A, second-order): `_query_parcel_centroid` —
+# ВНУТРЕННИЙ helper, зовётся из `_build_cannibalization` ПОД внешним SAVEPOINT
+# `_run`. Если он глотает db-сбой БЕЗ своего savepoint, tx остаётся aborted; внешний
+# `_run`-savepoint выходит «нормально» → RELEASE SAVEPOINT → RELEASE в aborted-tx САМ
+# падает (Postgres допускает лишь ROLLBACK / ROLLBACK TO SAVEPOINT) → внешний except
+# ловит RELEASE-ошибку, но tx не откачена → отравление каскадит в следующий индекс.
+# Тест воспроизводит ИМЕННО RELEASE-trap: `_ReleaseTrapSession` моделирует, что
+# RELEASE в aborted-tx падает (без внутреннего savepoint тест краснеет).
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class _FakeReleaseResult:
+    """Мини-результат db.execute() — покрывает `.mappings().first()`."""
+
+    def mappings(self) -> _FakeReleaseResult:
+        return self
+
+    def first(self) -> None:
+        return None
+
+
+class _ReleaseTrapSession:
+    """Fake Session, моделирующая Postgres SAVEPOINT/RELEASE/ROLLBACK в aborted-tx.
+
+    `execute()` бросает РОВНО раз (помечает tx aborted). `begin_nested()`:
+      • ВХОД в poisoned-tx → SAVEPOINT запрещён → бросает;
+      • ВЫХОД с исключением → ROLLBACK TO SAVEPOINT (снимает poison), пробрасывает;
+      • ВЫХОД нормально → RELEASE SAVEPOINT, и если poisoned — RELEASE ПАДАЕТ (trap).
+    """
+
+    def __init__(self) -> None:
+        self.poisoned = False
+        self._raised = False
+        self.calls = 0
+
+    def execute(self, *args: Any, **kwargs: Any) -> _FakeReleaseResult:
+        self.calls += 1
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        if not self._raised:
+            self._raised = True
+            self.poisoned = True
+            raise RuntimeError("simulated centroid query failure")
+        return _FakeReleaseResult()
+
+    @contextlib.contextmanager
+    def begin_nested(self):  # type: ignore[no-untyped-def]
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        try:
+            yield
+        except Exception:
+            self.poisoned = False  # ROLLBACK TO SAVEPOINT
+            raise
+        else:
+            if self.poisoned:
+                # RELEASE SAVEPOINT в aborted-tx ПАДАЕТ — тот самый trap.
+                raise RuntimeError(
+                    "current transaction is aborted, "
+                    "commands ignored until end of transaction block"
+                )
+
+
+def _outer_savepoint_call(db: Any, fn: Any, *, fallback: Any) -> Any:
+    """Зеркало `_run`/`_safe_call`: fn() под ВНЕШНИМ savepoint (RELEASE на норм-выходе)."""
+    try:
+        with db.begin_nested():
+            return fn()
+    except Exception:
+        return fallback
+
+
+class TestReleaseSavepointTrapParcelCentroid:
+    """`_query_parcel_centroid` НЕ отравляет внешний SAVEPOINT (RELEASE-trap)."""
+
+    def test_centroid_failure_survives_outer_release(self) -> None:
+        db = _ReleaseTrapSession()  # первый execute (centroid) падает
+        result = _outer_savepoint_call(
+            db,
+            lambda: _query_parcel_centroid(db, cad_num="66:41:0303161:123"),
+            fallback="OUTER_FALLBACK",
+        )
+        # centroid проглочен → его fallback (None), НЕ внешний OUTER_FALLBACK: значит
+        # внутренний savepoint откатил сбой и внешний RELEASE прошёл чисто.
+        assert result is None
+        # КЛЮЧЕВОЙ ассерт: сессия НЕ отравлена — follow-up execute без 'aborted'.
+        db.execute("SELECT 1")
+
+
 # ──────────────────────────────────────────────────────────────────────────────
 # §25.3 Cannibalization dispatch — TRUE own-portfolio vs PROXY fallback
 # ──────────────────────────────────────────────────────────────────────────────

backend/tests/services/site_finder/test_district_resolver.py

@@ -17,7 +17,10 @@ admin-names (.scalars().all()) + micro-lookup (.scalars().all()). КРИТИЧН
 
 from __future__ import annotations
 
+import contextlib
 import os
+from collections.abc import Callable, Iterator
+from typing import Any
 from unittest.mock import MagicMock
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
@@ -26,6 +29,7 @@ import pytest
 
 from app.services.site_finder import district_resolver
 from app.services.site_finder.district_resolver import (
+    _admin_names,
     _reset_admin_cache,
     resolve_objective_districts,
 )
@@ -220,3 +224,129 @@ class TestResolverSqlShape:
         # 8 админов = ekb_districts с непустым geom ('не определён' без geom исключён).
         sql = str(district_resolver._ADMIN_NAMES_SQL)
         assert "geom IS NOT NULL" in sql
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# RELEASE-SAVEPOINT trap (#2464 cluster A, second-order): резолвер зовётся §22-
+# билдерами (sales_series `_query_source_a/_b`, `_query_artificial_demand`) на общей
+# сессии отчёта ПОД внешним SAVEPOINT. Если внутренний swallow (_admin_names /
+# micro-lookup) съедает db-сбой БЕЗ своего savepoint, транзакция остаётся aborted;
+# внешний savepoint выходит «нормально» и делает RELEASE SAVEPOINT, который САМ падает
+# в aborted-tx (Postgres допускает лишь ROLLBACK / ROLLBACK TO SAVEPOINT) — внешний
+# except ловит RELEASE-ошибку, но tx так и не откачена → отравление каскадит. Fix =
+# savepoint В ТОЧКЕ ПЕРЕХВАТА. Тест воспроизводит ИМЕННО RELEASE-trap (не плоский
+# swallow): `_ReleaseTrapSession` моделирует, что RELEASE в aborted-tx падает.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class _FakeResult:
+    """Мини-результат db.execute() — покрывает `.scalars().all()` и `.mappings().first()`."""
+
+    def __init__(self, rows: list[Any]) -> None:
+        self._rows = list(rows)
+
+    def scalars(self) -> _FakeResult:
+        return self
+
+    def mappings(self) -> _FakeResult:
+        return self
+
+    def all(self) -> list[Any]:
+        return list(self._rows)
+
+    def first(self) -> Any:
+        return self._rows[0] if self._rows else None
+
+
+class _ReleaseTrapSession:
+    """Fake Session, моделирующая Postgres SAVEPOINT/RELEASE/ROLLBACK в aborted-tx.
+
+    - `execute()` на `fail_on_call`-м вызове бросает РОВНО раз и помечает tx aborted
+      (poisoned). Пока poisoned — любой execute бросает 'current transaction is aborted'.
+    - `begin_nested()` моделирует SAVEPOINT:
+        • ВХОД в poisoned-tx → SAVEPOINT запрещён → бросает (как реальный Postgres);
+        • ВЫХОД с исключением → ROLLBACK TO SAVEPOINT (снимает poison), пробрасывает;
+        • ВЫХОД нормально → RELEASE SAVEPOINT, и если tx poisoned — RELEASE ПАДАЕТ
+          (это и есть RELEASE-trap; без внутреннего savepoint тест краснеет).
+    """
+
+    def __init__(
+        self, *, fail_on_call: int = 1, results: dict[int, list[Any]] | None = None
+    ) -> None:
+        self.poisoned = False
+        self.calls = 0
+        self.fail_on_call = fail_on_call
+        self._results = results or {}
+
+    def execute(self, *args: Any, **kwargs: Any) -> _FakeResult:
+        self.calls += 1
+        if self.poisoned:
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        if self.calls == self.fail_on_call:
+            self.poisoned = True
+            raise RuntimeError("simulated inner helper query failure")
+        return _FakeResult(self._results.get(self.calls, []))
+
+    @contextlib.contextmanager
+    def begin_nested(self) -> Iterator[None]:
+        if self.poisoned:
+            # SAVEPOINT запрещён в aborted-tx.
+            raise RuntimeError(
+                "current transaction is aborted, commands ignored until end of transaction block"
+            )
+        try:
+            yield
+        except Exception:
+            # ROLLBACK TO SAVEPOINT — разрешён в aborted, снимает poison.
+            self.poisoned = False
+            raise
+        else:
+            if self.poisoned:
+                # RELEASE SAVEPOINT в aborted-tx ПАДАЕТ — тот самый trap.
+                raise RuntimeError(
+                    "current transaction is aborted, "
+                    "commands ignored until end of transaction block"
+                )
+
+
+def _outer_savepoint_call(db: Any, fn: Callable[[], Any], *, fallback: Any) -> Any:
+    """Зеркало `_run`/`_safe_call`: обернуть fn() во ВНЕШНИЙ savepoint (RELEASE на норм-выходе).
+
+    Именно этот RELEASE падает в aborted-tx, если внутренний helper отравил сессию без
+    своего savepoint → возвращается `fallback` (сигнал, что trap сработал).
+    """
+    try:
+        with db.begin_nested():
+            return fn()
+    except Exception:
+        return fallback
+
+
+class TestReleaseSavepointTrap:
+    """Внутренние swallow-сайты резолвера НЕ отравляют внешний SAVEPOINT (RELEASE-trap)."""
+
+    def test_admin_names_failure_survives_outer_release(self) -> None:
+        _reset_admin_cache()
+        db = _ReleaseTrapSession(fail_on_call=1)  # admin-names lookup падает
+        result = _outer_savepoint_call(db, lambda: _admin_names(db), fallback="OUTER_FALLBACK")
+        # _admin_names вернул СВОЙ fallback (пустой frozenset) — внешний RELEASE прошёл,
+        # значит внутренний savepoint откатил сбой (иначе result был бы OUTER_FALLBACK).
+        assert result == frozenset()
+        # КЛЮЧЕВОЙ ассерт: сессия НЕ отравлена — follow-up execute без 'aborted'.
+        db.execute("SELECT 1")
+
+    def test_micro_lookup_failure_survives_outer_release(self) -> None:
+        _reset_admin_cache()
+        # call 1 = admin-names (успех, "Кировский" распознан); call 2 = micro-lookup (падает).
+        db = _ReleaseTrapSession(fail_on_call=2, results={1: ["Кировский"]})
+        result = _outer_savepoint_call(
+            db,
+            lambda: resolve_objective_districts(db, "Кировский"),
+            fallback="OUTER_FALLBACK",
+        )
+        # micro-lookup проглочен → EKB-wide fallback (None), НЕ внешний OUTER_FALLBACK.
+        assert result is None
+        # КЛЮЧЕВОЙ ассерт: сессия НЕ отравлена — follow-up execute без 'aborted'.
+        db.execute("SELECT 1")