tradein-mvp/backend/app/api/v1/trade_in.py

@@ -180,8 +180,10 @@ async def estimate(
         ) from None
     # #747: атомарно-условный инкремент — источник истины по лимиту. check_and_raise
     # выше остаётся быстрым pre-check (429 до дорогой оценки), но финальное решение
-    # тут: при гонке двух /estimate на used=MONTHLY_LIMIT-1 второй получит False.
-    if not account_quota.increment(db, x_authenticated_user):
+    # тут: при гонке двух /estimate на used=lim-1 второй получит False.
+    # Не списываем квоту за пустой результат (нерезолвящийся адрес и т.п.) — иначе
+    # платный слот сгорает за insufficient_data=True (median=0, n_analogs=0) с HTTP 200.
+    if not result.insufficient_data and not account_quota.increment(db, x_authenticated_user):
         raise HTTPException(status_code=429, detail=account_quota.LIMIT_EXHAUSTED_MESSAGE)
     return result
 

tradein-mvp/backend/app/services/account_quota.py

@@ -1,8 +1,12 @@
 """Сервис квоты оценок trade-in — N успешных оценок в месяц на аккаунт.
 
 Правила:
-- Лимит = settings.estimate_quota_limit успешных оценок за календарный месяц
-  (UTC, период 'YYYY-MM'); конфигурируется через env ESTIMATE_QUOTA_LIMIT, default 15.
+- Лимит по умолчанию = settings.estimate_quota_limit успешных оценок за календарный
+  месяц (UTC, период 'YYYY-MM'); конфигурируется через env ESTIMATE_QUOTA_LIMIT,
+  default 15.
+- Персональный override: таблица account_quota_overrides (username → monthly_limit),
+  см. миграцию 185_account_quota_overrides.sql. Заменяет прежний хак бонусных попыток
+  через negative `used` (ломал /quota — «Осталось 50 из 15»).
 - Без лимита (unlimited): роль admin ИЛИ username == 'kopylov'.
 - Учитываются ТОЛЬКО успешные оценки (инкремент ПОСЛЕ estimate_quality).
 - Если заголовок X-Authenticated-User отсутствует (dev без Caddy) → unlimited,
@@ -53,11 +57,34 @@ def is_unlimited(username: str) -> bool:
         return False
 
 
+def user_limit(db: Session, username: str) -> int:
+    """Персональный месячный лимит для username, иначе глобальный MONTHLY_LIMIT.
+
+    Источник override — таблица account_quota_overrides (см. миграцию
+    185_account_quota_overrides.sql). Заменяет прежний хак бонусных попыток через
+    negative `used`, который ломал /quota (limit=15, used=-35 → remaining=50 —
+    «Осталось 50 из 15»).
+    """
+    row = db.execute(
+        text(
+            """
+            SELECT monthly_limit FROM account_quota_overrides
+            WHERE username = :u
+            """
+        ),
+        {"u": username},
+    ).fetchone()
+    if row is not None and row.monthly_limit is not None:
+        return int(row.monthly_limit)
+    return MONTHLY_LIMIT
+
+
 def get_status(db: Session, username: str | None) -> dict:
     """Возвращает статус квоты для пользователя.
 
-    Если username is None → unlimited True, used 0, remaining 15.
-    Если unlimited → used = фактический или 0, remaining = limit.
+    Если username is None → unlimited True, used 0, remaining = MONTHLY_LIMIT.
+    Если unlimited → used = фактический или 0, remaining = limit (per-user override
+    или глобальный MONTHLY_LIMIT).
     """
     if username is None:
         return {
@@ -69,6 +96,7 @@ def get_status(db: Session, username: str | None) -> dict:
 
     unlimited = is_unlimited(username)
     period = current_period()
+    limit = user_limit(db, username)
 
     row = db.execute(
         text(
@@ -84,15 +112,18 @@ def get_status(db: Session, username: str | None) -> dict:
 
     if unlimited:
         return {
-            "limit": MONTHLY_LIMIT,
+            "limit": limit,
             "used": used,
-            "remaining": MONTHLY_LIMIT,
+            "remaining": limit,
             "unlimited": True,
         }
 
-    remaining = max(0, MONTHLY_LIMIT - used)
+    # Защитный кламп: remaining никогда не превышает limit, даже если used всё же
+    # снова просочится отрицательным (прежний бонус-хак) — max(0, used) обнуляет
+    # отрицательный used перед вычитанием.
+    remaining = max(0, limit - max(0, used))
     return {
-        "limit": MONTHLY_LIMIT,
+        "limit": limit,
         "used": used,
         "remaining": remaining,
         "unlimited": False,
@@ -111,6 +142,7 @@ def check_and_raise(db: Session, username: str | None) -> None:
         return
 
     period = current_period()
+    limit = user_limit(db, username)
     row = db.execute(
         text(
             """
@@ -122,13 +154,13 @@ def check_and_raise(db: Session, username: str | None) -> None:
     ).fetchone()
 
     used = row.used if row is not None else 0
-    if used >= MONTHLY_LIMIT:
+    if used >= limit:
         logger.warning(
             "quota exhausted: username=%r period=%s used=%d limit=%d",
             username,
             period,
             used,
-            MONTHLY_LIMIT,
+            limit,
         )
         raise HTTPException(status_code=429, detail=LIMIT_EXHAUSTED_MESSAGE)
 
@@ -140,14 +172,16 @@ def increment(db: Session, username: str | None) -> bool:
     None / unlimited → True (no-op success).
 
     Защита от TOCTOU: предикат `WHERE used < :lim` применяется к ветке DO UPDATE —
-    два параллельных запроса при used=MONTHLY_LIMIT-1 не могут оба инкрементировать
-    (второй упрётся в WHERE → RETURNING пуст → False). Свежая вставка (used=1) НЕ
-    задевается WHERE (он только для DO UPDATE), поэтому первая оценка месяца проходит.
+    два параллельных запроса при used=lim-1 не могут оба инкрементировать (второй
+    упрётся в WHERE → RETURNING пуст → False). Свежая вставка (used=1) НЕ задевается
+    WHERE (он только для DO UPDATE), поэтому первая оценка месяца проходит. `lim` —
+    персональный лимит (user_limit), НЕ жёстко зашитый глобальный MONTHLY_LIMIT.
     """
     if username is None or is_unlimited(username):
         return True
 
     period = current_period()
+    lim = user_limit(db, username)
     row = db.execute(
         text(
             """
@@ -161,7 +195,7 @@ def increment(db: Session, username: str | None) -> bool:
             RETURNING used
             """
         ),
-        {"u": username, "p": period, "lim": MONTHLY_LIMIT},
+        {"u": username, "p": period, "lim": lim},
     ).fetchone()
     db.commit()
     ok = row is not None
@@ -172,6 +206,6 @@ def increment(db: Session, username: str | None) -> bool:
             "quota increment refused (atomic, #747): username=%r period=%s limit=%d",
             username,
             period,
-            MONTHLY_LIMIT,
+            lim,
         )
     return ok

tradein-mvp/backend/data/sql/185_account_quota_overrides.sql

@@ -0,0 +1,59 @@
+-- Migration 185: account_quota_overrides — персональные месячные лимиты оценок
+--
+-- WHY:
+--   Бонусные попытки для платящих пилотов раздавались хаком через SQL-runbook:
+--   `UPDATE account_estimate_usage SET used = used - N` (negative used). Для user2
+--   (Брусника) это дало used=-35 при глобальном limit=15 → GET /quota отдавал
+--   {limit:15, used:-35, remaining:50} → фронт (EstimateForm.tsx) рендерил абсурд
+--   «Осталось 50 из 15». Правильное решение — персональный лимит, а не отрицательный
+--   счётчик использования.
+--
+-- WHAT:
+--   1. Таблица account_quota_overrides(username PK, monthly_limit, note, updated_at) —
+--      per-user override глобального settings.estimate_quota_limit. Читается через
+--      app.services.account_quota.user_limit().
+--   2. Seed: user2 (Брусника) → monthly_limit=50, пилотный грант 2026-07-13.
+--   3. Сброс хака: account_estimate_usage.used < 0 → 0 (для user2, единственного
+--      затронутого аккаунта). После этой миграции user2: limit=50, used=0,
+--      remaining=50 — корректно вместо «50 из 15».
+--
+-- IDEMPOTENCY:
+--   CREATE TABLE IF NOT EXISTS + INSERT ... ON CONFLICT DO UPDATE (upsert, безопасен
+--   при повторном применении) + UPDATE ... WHERE used < 0 (после первого прогона
+--   used=0, условие больше не матчит — безопасный no-op при ре-apply).
+--
+-- Dependencies: 076_account_estimate_quota.sql (account_estimate_usage).
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS account_quota_overrides (
+    username      text        PRIMARY KEY,
+    monthly_limit integer     NOT NULL,
+    note          text,
+    updated_at    timestamptz NOT NULL DEFAULT now()
+);
+
+COMMENT ON TABLE account_quota_overrides IS
+    'Персональный override месячного лимита оценок trade-in (вместо глобального '
+    'settings.estimate_quota_limit). Читается app.services.account_quota.user_limit().';
+
+COMMENT ON COLUMN account_quota_overrides.monthly_limit IS
+    'Персональный лимит успешных оценок в календарный месяц для username.';
+
+COMMENT ON COLUMN account_quota_overrides.note IS
+    'Причина override (grant / pilot / manual bump) — для аудита ручных изменений.';
+
+INSERT INTO account_quota_overrides (username, monthly_limit, note)
+VALUES ('user2', 50, 'Брусника — пилот, грант 2026-07-13')
+ON CONFLICT (username) DO UPDATE SET
+    monthly_limit = EXCLUDED.monthly_limit,
+    note = EXCLUDED.note,
+    updated_at = now();
+
+-- Сбрасываем прежний хак бонусных попыток (negative used), который ломал /quota:
+-- limit=15, used=-35 → remaining=50 → фронт рендерил «Осталось 50 из 15».
+UPDATE account_estimate_usage
+SET used = 0, updated_at = now()
+WHERE username = 'user2' AND used < 0;
+
+COMMIT;

tradein-mvp/backend/tests/test_account_quota.py

@@ -6,6 +6,9 @@ Coverage:
   (c) increment растит used счётчик
   (d) get_status корректен для different сценариев
   (e) отсутствие заголовка X-Authenticated-User = unlimited (fail-open)
+  (f) #747 — атомарно-условный increment (TOCTOU fix)
+  (g) account_quota_overrides — персональный лимит вместо negative-used хака
+  (h) insufficient_data результат НЕ инкрементит квоту (geocode-fail не сжигает слот)
 
 DB мокируется через MagicMock — реальная БД не требуется.
 """
@@ -14,7 +17,9 @@ from __future__ import annotations
 
 import os
 import sys
-from unittest.mock import MagicMock
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
 
 # psycopg v3 driver required; stub DATABASE_URL before any app import
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
@@ -29,6 +34,7 @@ import pytest  # noqa: E402
 from fastapi import FastAPI  # noqa: E402
 from fastapi.testclient import TestClient  # noqa: E402
 
+from app.schemas.trade_in import AggregatedEstimate  # noqa: E402
 from app.services.account_quota import (  # noqa: E402
     LIMIT_EXHAUSTED_MESSAGE,
     MONTHLY_LIMIT,
@@ -36,6 +42,7 @@ from app.services.account_quota import (  # noqa: E402
     get_status,
     increment,
     is_unlimited,
+    user_limit,
 )
 
 # ---------------------------------------------------------------------------
@@ -43,23 +50,54 @@ from app.services.account_quota import (  # noqa: E402
 # ---------------------------------------------------------------------------
 
 
-def _db_with_used(used: int) -> MagicMock:
-    """DB session mock whose execute().fetchone() returns row with .used = N."""
-    row = MagicMock()
-    row.used = used
-    execute_result = MagicMock()
-    execute_result.fetchone.return_value = row
+class _Row:
+    """Row stand-in поддерживающий и .used/.monthly_limit, и индексный доступ row[0]."""
+
+    def __init__(self, value: int) -> None:
+        self.used = value
+        self.monthly_limit = value
+        self._t = (value,)
+
+    def __getitem__(self, i: int) -> int:
+        return self._t[i]
+
+
+def _result(row: _Row | None) -> MagicMock:
+    result = MagicMock()
+    result.fetchone.return_value = row
+    return result
+
+
+def _override_result(limit: int | None) -> MagicMock:
+    """Mock результата запроса account_quota_overrides.monthly_limit (user_limit())."""
+    return _result(None if limit is None else _Row(limit))
+
+
+def _used_result(used: int | None) -> MagicMock:
+    """Mock результата запроса account_estimate_usage.used (или UPSERT RETURNING used)."""
+    return _result(None if used is None else _Row(used))
+
+
+def _db_with_used(used: int, *, override_limit: int | None = None) -> MagicMock:
+    """DB session mock для get_status/check_and_raise: ровно 2 execute() —
+    (1) user_limit() override lookup, (2) account_estimate_usage.used lookup."""
     db = MagicMock()
-    db.execute.return_value = execute_result
+    db.execute.side_effect = [_override_result(override_limit), _used_result(used)]
     return db
 
 
-def _db_no_row() -> MagicMock:
-    """DB session mock where no row exists yet (first estimate of the month)."""
-    execute_result = MagicMock()
-    execute_result.fetchone.return_value = None
+def _db_no_row(*, override_limit: int | None = None) -> MagicMock:
+    """DB session mock где строки usage ещё нет (первая оценка месяца)."""
     db = MagicMock()
-    db.execute.return_value = execute_result
+    db.execute.side_effect = [_override_result(override_limit), _used_result(None)]
+    return db
+
+
+def _db_for_increment(*, override_limit: int | None, upsert_used: int | None) -> MagicMock:
+    """DB session mock для increment(): (1) user_limit() override lookup,
+    (2) UPSERT ... RETURNING used (None если WHERE used < lim не матчит)."""
+    db = MagicMock()
+    db.execute.side_effect = [_override_result(override_limit), _used_result(upsert_used)]
     return db
 
 
@@ -127,7 +165,7 @@ def test_check_and_raise_pilot_not_blocked_at_14() -> None:
 def test_check_and_raise_pilot_not_blocked_at_15_boundary() -> None:
     """used=15 == MONTHLY_LIMIT → 429 (15-я — последняя допустимая, 16-я блокируется).
 
-    Логика: used >= MONTHLY_LIMIT → block. После 15-й успешной оценки
+    Логика: used >= limit → block. После 15-й успешной оценки
     increment делает used=15, поэтому следующий запрос (16-й) блокируется.
     """
     db = _db_with_used(MONTHLY_LIMIT)
@@ -153,7 +191,7 @@ def test_check_and_raise_pilot_blocked_exact_detail() -> None:
 
 
 def test_check_and_raise_pilot_blocked_over_limit() -> None:
-    """used=20 тоже блокируется."""
+    """used=20 тоже блокируется (нет override → глобальный лимит)."""
     db = _db_with_used(20)
     from fastapi import HTTPException
 
@@ -168,14 +206,15 @@ def test_check_and_raise_pilot_blocked_over_limit() -> None:
 
 
 def test_increment_pilot_calls_upsert() -> None:
-    """increment для pilot → выполняет db.execute (UPSERT) и db.commit."""
-    db = MagicMock()
-    increment(db, "user1")
-    db.execute.assert_called_once()
+    """increment для pilot → выполняет user_limit lookup + UPSERT (2 execute), 1 commit."""
+    db = _db_for_increment(override_limit=None, upsert_used=1)
+    result = increment(db, "user1")
+    assert result is True
+    assert db.execute.call_count == 2
     db.commit.assert_called_once()
 
-    # Проверяем что SQL содержит ON CONFLICT ... DO UPDATE
-    call_args = db.execute.call_args
+    # Второй вызов — UPSERT; проверяем что SQL содержит ON CONFLICT ... DO UPDATE
+    call_args = db.execute.call_args_list[1]
     sql_text = str(call_args[0][0])  # first positional arg — text() object
     assert "ON CONFLICT" in sql_text
     assert "used" in sql_text
@@ -190,9 +229,9 @@ def test_increment_none_username_is_noop() -> None:
 
 def test_increment_pilot_first_estimate_of_month() -> None:
     """Первый инкремент (нет строки в БД) — должен всё равно выполнить UPSERT."""
-    db = MagicMock()
+    db = _db_for_increment(override_limit=None, upsert_used=1)
     increment(db, "user5")
-    db.execute.assert_called_once()
+    assert db.execute.call_count == 2
 
 
 # ---------------------------------------------------------------------------
@@ -212,7 +251,7 @@ def test_get_status_none_username() -> None:
 
 
 def test_get_status_admin() -> None:
-    """admin → unlimited, remaining=15 вне зависимости от used."""
+    """admin → unlimited, remaining=limit вне зависимости от used."""
     db = _db_with_used(7)
     status = get_status(db, "admin")
     assert status["unlimited"] is True
@@ -396,18 +435,7 @@ def test_estimate_no_header_not_blocked(estimate_app_exhausted: FastAPI) -> None
 # ---------------------------------------------------------------------------
 
 
-class _Row:
-    """Row stand-in поддерживающий и .used, и индексный доступ row[0]."""
-
-    def __init__(self, used: int) -> None:
-        self.used = used
-        self._t = (used,)
-
-    def __getitem__(self, i: int) -> int:
-        return self._t[i]
-
-
-class _Result:
+class _AtomicResult:
     def __init__(self, row: _Row | None) -> None:
         self._row = row
 
@@ -416,28 +444,33 @@ class _Result:
 
 
 class _AtomicQuotaFakeDB:
-    """In-memory fake, моделирующий атомарный conditional UPSERT (#747).
-
-    Воспроизводит семантику `INSERT ... ON CONFLICT DO UPDATE SET used=used+1
-    WHERE used < :lim RETURNING used`:
-      - строки ещё нет (used=None) → INSERT used=1, RETURNING row (WHERE не для INSERT);
-      - used < lim → used+=1, RETURNING row;
-      - used >= lim → конфликтная строка НЕ обновлена, RETURNING пуст (None).
+    """In-memory fake, воспроизводящий user_limit() override-lookup + атомарный
+    conditional UPSERT (#747):
+      - execute() без ключа "lim" в params → user_limit() override lookup —
+        возвращает override_limit (или None → increment() падает на глобальный
+        MONTHLY_LIMIT как :lim в следующем вызове).
+      - execute() с ключом "lim" → UPSERT: строки ещё нет (used=None) → INSERT used=1,
+        RETURNING row (WHERE не применяется к INSERT); used < lim → used+=1, RETURNING
+        row; used >= lim → конфликтная строка НЕ обновлена, RETURNING пуст (None).
     """
 
-    def __init__(self, *, used: int | None) -> None:
+    def __init__(self, *, used: int | None, override_limit: int | None = None) -> None:
         self.used = used
+        self.override_limit = override_limit
         self.commits = 0
 
-    def execute(self, _stmt: object, params: dict) -> _Result:
+    def execute(self, _stmt: object, params: dict) -> _AtomicResult:
+        if "lim" not in params:
+            row = _Row(self.override_limit) if self.override_limit is not None else None
+            return _AtomicResult(row)
         lim = params["lim"]
         if self.used is None:
             self.used = 1
         elif self.used < lim:
             self.used += 1
         else:
-            return _Result(None)
-        return _Result(_Row(self.used))
+            return _AtomicResult(None)
+        return _AtomicResult(_Row(self.used))
 
     def commit(self) -> None:
         self.commits += 1
@@ -472,3 +505,207 @@ def test_increment_atomic_unlimited_noop_returns_true() -> None:
     assert increment(db, "admin") is True
     assert db.commits == 0
     assert db.used == MONTHLY_LIMIT  # не тронут
+
+
+def test_increment_atomic_respects_override_limit() -> None:
+    """override=50 (user2): used=49 → increment True (used→50); used=50 → False.
+    Глобальный MONTHLY_LIMIT (15) НЕ применяется — используется персональный override."""
+    db = _AtomicQuotaFakeDB(used=49, override_limit=50)
+    assert increment(db, "user2") is True
+    assert db.used == 50
+    assert increment(db, "user2") is False
+    assert db.used == 50  # не 51
+
+
+def test_increment_atomic_override_below_global_blocks_early() -> None:
+    """override=5 (ниже глобального 15): used=5 уже блокирует, хотя < MONTHLY_LIMIT.
+
+    Демонстрирует что per-user override заменяет глобальный лимит полностью —
+    не является дополнительным потолком поверх него.
+    """
+    db = _AtomicQuotaFakeDB(used=5, override_limit=5)
+    assert increment(db, "user_low_override") is False
+    assert db.used == 5  # не выросло
+
+
+# ---------------------------------------------------------------------------
+# (g) account_quota_overrides — персональный лимит (замена negative-used хака)
+# ---------------------------------------------------------------------------
+
+
+def test_user_limit_no_override_returns_global() -> None:
+    """Нет строки в account_quota_overrides → user_limit() возвращает MONTHLY_LIMIT."""
+    db = MagicMock()
+    db.execute.return_value = _override_result(None)
+    assert user_limit(db, "user1") == MONTHLY_LIMIT
+
+
+def test_user_limit_with_override_returns_override() -> None:
+    """Есть строка override → user_limit() возвращает monthly_limit из неё."""
+    db = MagicMock()
+    db.execute.return_value = _override_result(50)
+    assert user_limit(db, "user2") == 50
+
+
+def test_get_status_override_limit() -> None:
+    """user2 с override=50, used=0 (после сброса хака) → limit=50, remaining=50."""
+    db = _db_with_used(0, override_limit=50)
+    status = get_status(db, "user2")
+    assert status["limit"] == 50
+    assert status["remaining"] == 50
+    assert status["used"] == 0
+
+
+def test_get_status_override_remaining_clamped_even_if_used_negative() -> None:
+    """Кламп: даже если used снова просочится отрицательным (regression прежнего
+    negative-used хака), remaining НЕ превышает limit — не «Осталось 50 из 15»."""
+    db = _db_with_used(-35, override_limit=50)
+    status = get_status(db, "user2")
+    assert status["limit"] == 50
+    assert status["used"] == -35  # raw used не скрываем — диагностическая честность
+    assert status["remaining"] == 50  # clamp: max(0, 50 - max(0, -35)) == 50
+    assert status["remaining"] <= status["limit"]
+
+
+def test_check_and_raise_override_blocks_below_global_limit() -> None:
+    """override=5 (ниже глобального 15) — used=5 блокируется, хотя < MONTHLY_LIMIT."""
+    db = _db_with_used(5, override_limit=5)
+    from fastapi import HTTPException
+
+    with pytest.raises(HTTPException) as exc_info:
+        check_and_raise(db, "user_low_override")
+    assert exc_info.value.status_code == 429
+
+
+def test_check_and_raise_override_allows_above_global_limit() -> None:
+    """override=50 — used=20 (> глобального 15) НЕ блокируется."""
+    db = _db_with_used(20, override_limit=50)
+    check_and_raise(db, "user2")  # не должно поднять исключение
+
+
+def test_increment_override_blocks_at_override_not_global() -> None:
+    """increment уважает per-user override: used>=override → False, даже если
+    used < MONTHLY_LIMIT (15)."""
+    db = _db_for_increment(override_limit=5, upsert_used=None)  # WHERE used<5 не матчит
+    assert increment(db, "user_low_override") is False
+
+
+def test_increment_override_allows_above_global_limit() -> None:
+    """increment с override=50: used=20 (>15 глобального) успешно инкрементит."""
+    db = _db_for_increment(override_limit=50, upsert_used=21)
+    assert increment(db, "user2") is True
+
+
+# ---------------------------------------------------------------------------
+# (h) insufficient_data результат НЕ инкрементит квоту (geocode-fail не сжигает слот)
+# ---------------------------------------------------------------------------
+
+
+def _insufficient_estimate() -> AggregatedEstimate:
+    """Пустой результат — как _empty_estimate() в estimator.py: median=0 → #697
+    computed_field insufficient_data=True."""
+    return AggregatedEstimate(
+        estimate_id=uuid4(),
+        median_price_rub=0,
+        range_low_rub=0,
+        range_high_rub=0,
+        median_price_per_m2=0,
+        confidence="low",
+        confidence_explanation="address_not_geocoded",
+        n_analogs=0,
+        period_months=24,
+        analogs=[],
+        actual_deals=[],
+        expires_at=datetime.now(tz=UTC) + timedelta(hours=24),
+    )
+
+
+def _real_estimate() -> AggregatedEstimate:
+    """Непустой результат — insufficient_data=False (median_price_rub > 0)."""
+    return AggregatedEstimate(
+        estimate_id=uuid4(),
+        median_price_rub=5_000_000,
+        range_low_rub=4_500_000,
+        range_high_rub=5_500_000,
+        median_price_per_m2=100_000,
+        confidence="medium",
+        n_analogs=8,
+        period_months=24,
+        analogs=[],
+        actual_deals=[],
+        expires_at=datetime.now(tz=UTC) + timedelta(hours=24),
+    )
+
+
+@pytest.fixture()
+def estimate_app_ok() -> FastAPI:
+    """FastAPI app где check_and_raise проходит (used=0, без override) —
+    estimate_quality мокается отдельно в каждом тесте."""
+    from app.api.v1 import trade_in as trade_in_module
+    from app.core.db import get_db
+
+    application = FastAPI()
+    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
+
+    def _override_db():
+        yield _db_no_row()
+
+    application.dependency_overrides[get_db] = _override_db
+    return application
+
+
+def test_estimate_insufficient_data_does_not_increment_quota(
+    estimate_app_ok: FastAPI,
+) -> None:
+    """POST /estimate с insufficient_data=True результатом (нерезолвящийся адрес) →
+    account_quota.increment НЕ вызывается (пустой результат не списывает платный слот)."""
+    client = TestClient(estimate_app_ok, raise_server_exceptions=False)
+
+    with (
+        patch(
+            "app.services.estimator.estimate_quality",
+            new=AsyncMock(return_value=_insufficient_estimate()),
+        ),
+        patch("app.services.account_quota.increment") as mock_increment,
+    ):
+        resp = client.post(
+            "/api/v1/trade-in/estimate",
+            json={
+                "address": "г. Екатеринбург, несуществующий адрес xyz",
+                "area_m2": 50.0,
+                "rooms": 2,
+            },
+            headers={"X-Authenticated-User": "user1"},
+        )
+
+    assert resp.status_code == 200
+    assert resp.json()["median_price_rub"] == 0
+    assert resp.json()["insufficient_data"] is True
+    mock_increment.assert_not_called()
+
+
+def test_estimate_real_result_still_increments_quota(estimate_app_ok: FastAPI) -> None:
+    """Контрольный тест: непустой результат (insufficient_data=False) — increment
+    вызывается как обычно (регресс-guard, не ломаем #747 TOCTOU-семантику)."""
+    client = TestClient(estimate_app_ok, raise_server_exceptions=False)
+
+    with (
+        patch(
+            "app.services.estimator.estimate_quality",
+            new=AsyncMock(return_value=_real_estimate()),
+        ),
+        patch("app.services.account_quota.increment", return_value=True) as mock_increment,
+    ):
+        resp = client.post(
+            "/api/v1/trade-in/estimate",
+            json={
+                "address": "г. Екатеринбург, ул. Малышева, 1",
+                "area_m2": 50.0,
+                "rooms": 2,
+            },
+            headers={"X-Authenticated-User": "user1"},
+        )
+
+    assert resp.status_code == 200
+    assert resp.json()["insufficient_data"] is False
+    mock_increment.assert_called_once()