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

@@ -52,6 +52,27 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
+# PR-D1: единственное определение «оценка читаема» — раньше SQL-фильтр (404,
+# ниже в get_estimate) и Python-проверка (410, в estimate_pdf) уже разошлись
+# по коду ответа; третий потребитель (`/r/<token>`, PR-9) разошёлся бы
+# неизбежно без унификации. `retain_until > NOW()` при NULL даёт NULL → false
+# в SQL — для всех существующих строк (retain_until IS NULL) поведение не
+# меняется вообще. Не копировать это выражение по месту — только через
+# константу/хелпер ниже. Payments retention, PR #2754.
+ESTIMATE_READABLE_SQL = "(expires_at > NOW() OR retain_until > NOW())"
+
+
+def estimate_readable(expires_at: datetime, retain_until: datetime | None) -> bool:
+    """Python-зеркало ESTIMATE_READABLE_SQL — та же дизъюнкция, без похода в БД.
+
+    tzinfo-нормализация повторяет прежнюю Python-проверку (estimate_pdf) —
+    `.replace(tzinfo=UTC)`, не переизобретается.
+    """
+    now = datetime.now(tz=UTC)
+    if expires_at.replace(tzinfo=UTC) > now:
+        return True
+    return retain_until is not None and retain_until.replace(tzinfo=UTC) > now
+
 
 def _assert_estimate_access(created_by: str | None, x_authenticated_user: str | None) -> None:
     """IDOR guard (#690): только владелец оценки или admin могут её читать.
@@ -249,11 +270,11 @@ def get_estimate(
     """
     row = db.execute(
         text(
-            """
+            f"""
             SELECT id, median_price, range_low, range_high, median_price_per_m2,
                    confidence, confidence_explanation, n_analogs,
                    analogs, actual_deals, sources_used, data_freshness_minutes,
-                   expires_at, address, lat, lon,
+                   expires_at, retain_until, address, lat, lon,
                    area_m2, rooms, floor, total_floors,
                    year_built, house_type, repair_state, has_balcony,
                    canonical_address, house_cadnum, house_fias_id,
@@ -263,7 +284,7 @@ def get_estimate(
                    asking_to_sold_ratio, ratio_basis, created_by, created_at
             FROM trade_in_estimates
             WHERE id = CAST(:id AS uuid)
-              AND expires_at > NOW()
+              AND {ESTIMATE_READABLE_SQL}
             """
         ),
         {"id": str(estimate_id)},
@@ -372,6 +393,7 @@ def get_estimate(
         analogs=analogs,
         actual_deals=actual_deals,
         expires_at=row.expires_at,
+        retain_until=row.retain_until,
         target_address=row.address,
         target_lat=row.lat,
         target_lon=row.lon,
@@ -433,7 +455,7 @@ def estimate_pdf(
             SELECT id, median_price, range_low, range_high, median_price_per_m2,
                    confidence, confidence_explanation, n_analogs,
                    analogs, actual_deals, sources_used, data_freshness_minutes,
-                   expires_at,
+                   expires_at, retain_until,
                    address, lat, lon, area_m2, rooms, floor, total_floors,
                    year_built, house_type, repair_state, has_balcony,
                    canonical_address, house_cadnum, house_fias_id,
@@ -453,8 +475,12 @@ def estimate_pdf(
 
     _assert_estimate_access(row.created_by, x_authenticated_user)
 
-    if row.expires_at.replace(tzinfo=UTC) < datetime.now(tz=UTC):
-        raise HTTPException(status_code=410, detail="estimate expired (24h TTL)")
+    # PR-D1: тот же гейт, что в get_estimate (см. ESTIMATE_READABLE_SQL) — раньше
+    # здесь была независимая Python-проверка expires_at, разошедшаяся с SQL-
+    # фильтром GET-ручки. "estimate expired (24h TTL)" убрано из текста: при
+    # годовом retain_until упоминание 24ч в ответе API стало бы ложью.
+    if not estimate_readable(row.expires_at, row.retain_until):
+        raise HTTPException(status_code=410, detail="estimate expired")
 
     from app.services.estimator import _qc_geo_to_precision
 
@@ -477,6 +503,7 @@ def estimate_pdf(
         analogs=analogs,
         actual_deals=actual_deals,
         expires_at=row.expires_at,
+        retain_until=row.retain_until,
         target_address=row.address,
         target_lat=row.lat,
         target_lon=row.lon,

tradein-mvp/backend/app/core/config.py

@@ -836,6 +836,22 @@ class Settings(BaseSettings):
     # срок — решение DPO/юриста, не инженера). ENV: TRADE_IN_LEAD_RETENTION_DAYS.
     trade_in_lead_retention_days: int = 180
 
+    # ── Платный отчёт живёт год (retain_until, migration 240, PR #2754) ─────
+    # trade_in_estimates.retain_until TTL (дни ОТ ОПЛАТЫ) — срок жизни ССЫЛКИ/
+    # СТРОКИ для оплаченной оценки, независимый от expires_at (актуальность
+    # расчёта, 24ч, глобальный для ВСЕХ строк). НЕ трогает expires_at — см.
+    # migration 240 докстринг. Отдельная колонка, а не подъём expires_at:
+    # expires_at печатается в PDF/UI как «актуальность расчёта» и одинаков
+    # для всех строк, поднять его до года = соврать в документе клиента про
+    # свежесть цифры + нарушить минимизацию ПДн для неоплаченных B2C-адресов.
+    # Единственный источник числа «12 месяцев» на фронте —
+    # `mera-public/content.ts::PAID_REPORT_RETENTION_MONTHS`; текст оферты,
+    # экран после оплаты и SQL продления retain_until при оплате (платёжный
+    # код, отдельный PR) обязаны читать его оттуда, а не хардкодить — иначе
+    # классический исход "в оферте 12 месяцев, в конфиге 365 дней, на экране
+    # «год»". ENV: TRADE_IN_PAID_RETENTION_DAYS.
+    trade_in_paid_retention_days: int = 365
+
     # Батч-размер физического DELETE в purge_expired_trade_in_data (нельзя одним
     # DELETE по всей таблице — долгая блокировка на большом бэклоге). Задача сама
     # крутит цикл батчей за один прогон (см. _DEFAULT_MAX_BATCHES в таске) —

tradein-mvp/backend/app/schemas/trade_in.py

@@ -196,6 +196,10 @@ class AggregatedEstimate(BaseModel):
     analogs: list[AnalogLot]
     actual_deals: list[AnalogLot]  # реальные продажи last 12 mo
     expires_at: datetime
+    # PR-D1: срок жизни ССЫЛКИ/СТРОКИ (оплаченный доступ), НЕ актуальности
+    # расчёта — тот остаётся expires_at (не путать, см. migration 240).
+    # NULL = неоплачено (весь текущий трафик, B2B pilots включительно).
+    retain_until: datetime | None = None
     # ── Дополнительные метаданные ──
     target_address: str | None = None  # geocoded full address
     target_lat: float | None = None

tradein-mvp/backend/app/services/exporters/trade_in_pdf.py

@@ -1050,6 +1050,19 @@ def _build_cover(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> s
     )
     report_num = _report_number(estimate)
 
+    # PR-D1: «Ссылка доступна до …» — срок жизни ОПЛАЧЕННОГО доступа
+    # (retain_until), НЕ путать со «Срок действия данных» (expires_at,
+    # актуальность расчёта) над ней — эта строка не трогается. Рендерится
+    # ТОЛЬКО когда retain_until IS NOT NULL (неоплаченные — весь текущий
+    # трафик — не видят этой строки вообще, поведение бит-в-бит текущее).
+    retain_until_row = (
+        f'<tr><td class="dotted-row">Ссылка доступна до</td>'
+        f'<td class="bold dotted-row">'
+        f"{_mono(estimate.retain_until.date().strftime('%d.%m.%Y'))}</td></tr>"
+        if estimate.retain_until is not None
+        else ""
+    )
+
     # Короткий адрес (для cover): берём первую часть до запятой
     full_address = input_snapshot.get("address", "—")
     address_short = full_address.split(",")[0:3]
@@ -1146,6 +1159,7 @@ def _build_cover(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> s
         <td class="bold dotted-row">{_mono(today.strftime("%d.%m.%Y"))}</td></tr>
     <tr><td class="dotted-row">Срок действия данных</td>
         <td class="bold dotted-row">до {_mono(expires.strftime("%d.%m.%Y"))}</td></tr>
+    {retain_until_row}
     <tr><td class="dotted-row">Адрес</td><td class="bold dotted-row">{address}</td></tr>
     <tr><td class="dotted-row">Год постройки</td>
         <td class="bold dotted-row">{year_label}</td></tr>

tradein-mvp/backend/app/tasks/purge_expired_trade_in_data.py

@@ -54,6 +54,21 @@ BATCHING (не единый DELETE по всей таблице):
     don't match `expires_at < NOW()` on the next run; a mid-run failure leaves earlier
     committed batches deleted (correct, not rolled back) and mark_failed records the
     partial counters reached so far.
+
+Payments retention (PR #2754): the `created_by IS NULL` population above is EXACTLY
+    the future paying-customer population -- the owner sells this report to
+    individuals for money, and a paid row must outlive the 24h `expires_at` link TTL
+    (a separate column, `retain_until`, set by the -- separate, not-yet-existing --
+    payment fulfillment code to now() + settings.trade_in_paid_retention_days, NOT
+    a change to `expires_at` itself). Two independent safeguards were added to
+    `_DELETE_EXPIRED_ESTIMATES_SQL` (retain_until IS NULL + NOT EXISTS payments)
+    plus a pre-flight count in `purge_expired_trade_in_data` that refuses to run at
+    all if it finds an ANOMALOUS paid candidate -- see the SQL constants and
+    `_preflight_paid_candidates` below for the mechanics (deep-review finding
+    2026-08-06 MEDIUM on PR #2754: the pre-flight predicate itself must ALSO carry
+    `retain_until IS NULL`, otherwise a perfectly healthy paid row trips it and
+    wedges the job permanently -- see that function's docstring). No payment code
+    lives in this file.
 """
 
 from __future__ import annotations
@@ -74,6 +89,26 @@ logger = logging.getLogger(__name__)
 # remainder simply drains on the next nightly run (idempotent, no data loss risk).
 _DEFAULT_MAX_BATCHES = 20
 
+#
+# Payments retention (2026-08-06, PR #2754): два независимых предохранителя
+# добавлены к тому же предикату ПЕРЕД тем, как платёжный код появился в
+# проекте (мина уже была заряжена: без них джоба удаляла бы будущих платящих
+# клиентов). Отдельная колонка retain_until (не подъём expires_at) — потому
+# что expires_at глобальный TTL расчёта на ВСЕ строки (включая неоплаченные)
+# и печатается в PDF/UI как «актуальность расчёта»; поднять его до года
+# означало бы одновременно нарушить минимизацию ПДн по 152-ФЗ и соврать в
+# документе клиента про срок актуальности цифры:
+#   1. `retain_until IS NULL` — именно IS NULL, НЕ `< NOW()`. Оплаченная
+#      строка (retain_until IS NOT NULL, migration 240) не удаляется джобой
+#      В ПРИНЦИПЕ, пока не поднято ослабление отдельным PR не раньше чем
+#      через год после первой продажи. `retain_until` ставится сервисным
+#      кодом платёжного контура (ещё не существует в этом PR) на now() +
+#      settings.trade_in_paid_retention_days.
+#   2. `NOT EXISTS (payments)` — независимая страховка на случай, если выдача
+#      забыла проставить retain_until (баг/гонка/ручной INSERT): строка,
+#      которой коснулись деньги, переживёт джобу даже без корректного (1).
+#      `payments` создана migration 233 (payments_estimate_idx — дешёвый терм).
+# См. также _preflight_paid_candidates ниже — та же логика ДО первого батча.
 _DELETE_EXPIRED_ESTIMATES_SQL = text(
     """
     DELETE FROM trade_in_estimates
@@ -81,12 +116,39 @@ _DELETE_EXPIRED_ESTIMATES_SQL = text(
         SELECT id FROM trade_in_estimates
         WHERE expires_at < NOW()
           AND created_by IS NULL
+          AND retain_until IS NULL
+          AND NOT EXISTS (
+              SELECT 1 FROM payments p WHERE p.estimate_id = trade_in_estimates.id
+          )
         ORDER BY expires_at
         LIMIT CAST(:batch_size AS int)
     )
     """
 )
 
+# Pre-flight (см. _preflight_paid_candidates). deep-review finding 2026-08-06
+# MEDIUM (PR #2754): первая редакция считала по БАЗОВОМУ предикату БЕЗ
+# retain_until вообще -- а это ловит и штатно-здоровые оплаченные строки
+# (retain_until проставлен, есть payments) точно так же, как настоящую
+# аномалию (retain_until НЕ проставлен, но payments есть) -- джоба вставала
+# на первой же честной продаже и больше никогда не запускалась (вместе с ней
+# вставало и удаление лидов, вызываемое из той же функции ПОСЛЕ этой
+# проверки -- 180-дневный purge по 152-ФЗ тоже переставал бы работать).
+# Правильная форма: базовый предикат AND "новый предохранитель НЕ сработал
+# бы" (retain_until IS NULL) AND "признак аномалии" (payments всё же есть).
+# Здоровая оплаченная строка (retain_until IS NOT NULL) исключается ЭТИМ
+# термом -- она и так под DELETE не попадает (см. safeguard 1 выше), тревогу
+# поднимать не должна.
+_PREFLIGHT_PAID_CANDIDATES_SQL = text(
+    """
+    SELECT count(*) FROM trade_in_estimates e
+    WHERE e.expires_at < NOW()
+      AND e.created_by IS NULL
+      AND e.retain_until IS NULL
+      AND EXISTS (SELECT 1 FROM payments p WHERE p.estimate_id = e.id)
+    """
+)
+
 _DELETE_EXPIRED_LEADS_SQL = text(
     """
     DELETE FROM trade_in_leads
@@ -135,6 +197,25 @@ def _drain_expired(
             break  # caught up -- fewer expired rows left than one batch
 
 
+def _preflight_paid_candidates(db: Session) -> int:
+    """Safety gate: count ANOMALOUS purge-candidates -- base predicate, retain_until
+    IS NULL (safeguard 1 did NOT protect the row), AND a payments row exists anyway.
+
+    Runs BEFORE any DELETE batch. A non-zero result means fulfillment failed to set
+    `retain_until` on a row money actually touched (bug/race/manual INSERT) -- this
+    run must not delete anything; see `purge_expired_trade_in_data` below, which
+    aborts before the first batch when this returns non-zero.
+
+    MUST include `retain_until IS NULL` (deep-review finding 2026-08-06 MEDIUM, PR
+    #2754): a healthy paid row (retain_until set, has a payments row) is the EXPECTED
+    steady state one day after every sale -- without this term it counts as a "paid
+    candidate" too, so the very first successful sale permanently wedges this job
+    (mark_failed, zero deletions, forever -- and since leads purge runs from the same
+    function AFTER this check, the unrelated 180-day lead retention would also stop).
+    """
+    return db.execute(_PREFLIGHT_PAID_CANDIDATES_SQL).scalar_one()
+
+
 def purge_expired_trade_in_data(
     db: Session,
     run_id: int,
@@ -148,10 +229,32 @@ def purge_expired_trade_in_data(
     deactivate_stale_listings). Finalises the scrape_runs row (mark_done / mark_failed).
 
     Returns {"estimates_deleted": N, "leads_deleted": M}.
+
+    Payments retention pre-flight (see `_preflight_paid_candidates`): if any
+    purge-candidate estimate has `retain_until IS NULL` AND a `payments` row (the
+    ANOMALY -- fulfillment failed to set retain_until on a row money touched), the
+    run aborts BEFORE the first DELETE batch (estimates OR leads) -- zero rows
+    deleted, `mark_failed` records why. Healthy paid rows (retain_until set) do NOT
+    trip this -- they never matched the check to begin with. This is deliberately
+    checked outside the `try` below so it can never be caught and silently
+    re-reported as a generic mid-run failure -- it is a distinct, actionable
+    pre-condition failure.
     """
     batch_size = batch_size or settings.trade_in_purge_batch_size
     max_batches = max_batches or _DEFAULT_MAX_BATCHES
     counters: dict[str, int] = {"estimates_deleted": 0, "leads_deleted": 0}
+
+    paid_candidates = _preflight_paid_candidates(db)
+    if paid_candidates:
+        error = (
+            f"pre-flight abort: {paid_candidates} purge-candidate trade_in_estimates "
+            "row(s) have retain_until IS NULL but a matching payments row -- "
+            "refusing to run, zero rows deleted"
+        )
+        logger.error("purge_expired_trade_in_data run_id=%d %s", run_id, error)
+        runs_mod.mark_failed(db, run_id, error, counters)
+        raise RuntimeError(error)
+
     try:
         _drain_expired(
             db,

tradein-mvp/backend/data/sql/240_trade_in_estimates_retain_until.sql

@@ -0,0 +1,90 @@
+-- 240_trade_in_estimates_retain_until.sql
+-- Платёжный контур МЕРЫ, ретеншен (PR #2754): «оплаченное живёт год, purge
+-- его не трогает». Владелец продаёт отчёт физлицу за 150 ₽ — отчёт должен
+-- жить год на нашей стороне, а не 24ч (см. WHY ниже).
+-- Номер сверен по `forgejo/main` и всем открытым PR-веткам ДВАЖДЫ: сначала
+-- как 234 (последняя занятая на момент ветвления была 233_payments.sql), но
+-- main уехал вперёд и 234 занял `234_scrape_runs_ban_kind_unknown.sql`
+-- (коммит 0de22f4b) — переименовано в 240 (main max на момент повторной
+-- сверки — 239, с дырами 235-237; max+1 безопаснее дыр). Урок пятый за
+-- сутки: сверять номер нужно не только перед первым коммитом, а прямо перед
+-- пушем/мержем — main не стоит на месте.
+--
+-- ── WHY ──────────────────────────────────────────────────────────────────────
+-- purge_expired_trade_in_data (migration 231, seeded enabled=false) удаляет
+-- строки `WHERE expires_at < NOW() AND created_by IS NULL` — это ровно
+-- популяция будущих платящих физлиц (анонимные B2C-оценки). Владелец продаёт
+-- отчёт физлицу за 150 ₽: скачанный файл у клиента бессрочно, но ссылка/строка
+-- на нашей стороне обязана жить дольше 24-часового TTL расчёта — иначе первый
+-- же прогон purge-джобы после запуска продаж физически и безвозвратно удалит
+-- уже оплаченное (PDF нигде не хранится, рендерится на лету).
+--
+-- `expires_at` НЕ трогаем ни на йоту: это единая глобальная настройка
+-- (`trade_in_estimate_retention_hours`), она же — печатаемая в PDF/UI дата
+-- «ДЕЙСТВИТЕЛЕН ДО» (актуальность РАСЧЁТА, а не срок жизни строки), и от неё
+-- зависит вычисление даты расчёта во фронте (`mappers.ts` fmtDateShift(-24)).
+-- Поднять её до года означало бы: (а) дать год хранения ВСЕМ строкам, включая
+-- неоплаченные адреса физлиц — прямое нарушение минимизации по 152-ФЗ;
+-- (б) напечатать в PDF клиента, что расчёт актуален год.
+--
+-- ── WHAT ─────────────────────────────────────────────────────────────────────
+-- Новая, независимая колонка retain_until — срок жизни ДОСТУПА/СТРОКИ:
+-- NULL = неоплаченная строка, поведение (чтение/PDF/purge) бит-в-бит текущее.
+-- Бэкфилла нет — все 1058 существующих строк остаются NULL, ничего не меняется
+-- для уже созданных оценок (весь B2B pilot-трафик в их числе).
+-- При оплате (платёжный код — отдельный PR, здесь его нет) сервисный слой
+-- проставит retain_until = now() + trade_in_paid_retention_days (config.py).
+--
+-- Частичный индекс покрывает predicate purge-джобы (migration 231,
+-- `_DELETE_EXPIRED_ESTIMATES_SQL`) уже С УЧЁТОМ нового терма retain_until —
+-- заведён вместе с колонкой, а не отдельной миграцией, чтобы purge не начал
+-- жить без него хотя бы один деплой.
+--
+-- ── IDEMPOTENCY ──────────────────────────────────────────────────────────────
+-- ADD COLUMN IF NOT EXISTS + CREATE INDEX IF NOT EXISTS — безопасный re-run.
+-- Ничего не удаляет, не бэкфиллит, DDL-only (доли секунды на 1058 строках).
+--
+-- Dependencies: 001_trade_in_estimates.sql, 233_payments.sql (индекс исключает
+-- строки со строкой в payments опосредованно через predicate purge-джобы,
+-- сама таблица payments здесь не читается).
+-- Apply after: 233_payments.sql.
+--
+-- ── lock_timeout — выставлен, хотя гейт (scripts/check-migration-lock-timeout.py)
+-- этот файл не проверяет ────────────────────────────────────────────────────
+-- Порог гейта для tradein (NN >= 250) — артефакт: назначен по номеру аварийной
+-- миграции 250, которую затем сняли с деплоя (#2792, 29f10002). Фактический
+-- максимум применённого на main — 239, то есть НИ ОДНА миграция в диапазоне
+-- 240-249 (этот файл включительно) гейтом не проверяется вообще — "проверено
+-- новых миграций: 0" в логе означает "не проверено ни одного файла", а не
+-- "все чисты". Сама функция scan() внутри гейта, если прогнать её без
+-- порогового отсечения, помечает ALTER TABLE ниже как блокирующий DDL без
+-- lock_timeout. `trade_in_estimates` — самая горячая таблица стека (история,
+-- история сотрудников, каждое чтение/PDF оценки); на этой БД уже наблюдались
+-- открытые транзакции на 46 и 22 часа. Ждущая ACCESS EXCLUSIVE-блокировка
+-- встаёт в очередь ПЕРЕД новыми запросами — за ней начинают ждать обычные
+-- SELECT приложения (см. `sql.md` § lock_timeout). На 1058 строках сам DDL
+-- мгновенный — риск не в исполнении, а в ожидании чужой блокировки. Красный
+-- деплой по таймауту — осознанно принятый в проекте размен (честный отказ
+-- лучше тихой очереди перед приложением). НЕ убирать как "гейт же не просит".
+
+BEGIN;
+
+SET LOCAL lock_timeout = '5s';
+
+ALTER TABLE trade_in_estimates
+    ADD COLUMN IF NOT EXISTS retain_until timestamptz;
+
+COMMENT ON COLUMN trade_in_estimates.retain_until IS
+    'До какого момента строку НЕЛЬЗЯ удалять и ссылка обязана открываться '
+    '(оплаченный доступ). Семантика expires_at не меняется: это дата '
+    'актуальности РАСЧЁТА (24ч), она печатается в PDF. NULL = неоплачено, '
+    'поведение бит-в-бит текущее. Задаётся сервисным кодом платёжного контура '
+    '(отдельный PR) на now() + trade_in_paid_retention_days (config.py).';
+
+-- Частичный индекс под predicate purge-джобы (app/tasks/purge_expired_trade_in_data.py):
+-- WHERE created_by IS NULL AND retain_until IS NULL AND expires_at < NOW().
+CREATE INDEX IF NOT EXISTS trade_in_estimates_purge_idx
+    ON trade_in_estimates (expires_at)
+    WHERE created_by IS NULL AND retain_until IS NULL;
+
+COMMIT;

tradein-mvp/backend/data/sql/_manifest_applied.txt

@@ -241,3 +241,4 @@
 225_listing_source_snapshots_run_id_idx.sql
 233_payments.sql
 234_scrape_runs_ban_kind_unknown.sql
+240_trade_in_estimates_retain_until.sql

tradein-mvp/backend/tests/skip_allowlist.txt

@@ -48,6 +48,13 @@ tests/test_data_erasure.py::test_real_erase_by_phone_finds_differently_formatted
 tests/test_data_erasure.py::test_real_erase_by_phone_finds_ru_trunk_prefix_variant
 tests/test_purge_expired_trade_in_data.py::test_real_purge_deletes_only_anonymous_expired_estimates
 
+# Payments retention (PR #2754), deep-review MEDIUM finding 2026-08-06 — тот же
+# `_live_session()`. Проверяют предполётную проверку purge против реального
+# retain_until/payments состояния (здоровая оплаченная строка не поднимает
+# тревогу и не блокирует джобу), не только SQL-текст.
+tests/test_purge_expired_trade_in_data.py::test_real_preflight_ignores_healthy_paid_row_flags_only_anomaly
+tests/test_purge_expired_trade_in_data.py::test_real_purge_not_wedged_by_healthy_paid_row
+
 # Диагноз оборванного прогона (#2764) — тот же `_live_session()`. Проверяет, что
 # дефолт mark_banned ('unknown') проходит CHECK-констрейнт из миграции 234:
 # на мок-лэйне (deploy-tradein.yml, DSN-заглушка) констрейнта нет вовсе.

tradein-mvp/backend/tests/test_estimate_idor.py

@@ -51,8 +51,13 @@ def trade_in_app() -> FastAPI:
     return application
 
 
-def _make_estimate_row(created_by: str | None) -> SimpleNamespace:
-    """A trade_in_estimates row with the full column set the endpoints read."""
+def _make_estimate_row(created_by: str | None, retain_until: object = None) -> SimpleNamespace:
+    """A trade_in_estimates row with the full column set the endpoints read.
+
+    retain_until defaults to None (PR-D1, migration 240) -- unpaid, matches every
+    row that existed before that migration; explicit param lets retention-gate
+    tests (see test_estimate_retention_gate.py) construct a paid row.
+    """
     from datetime import UTC, datetime, timedelta
 
     return SimpleNamespace(
@@ -69,6 +74,7 @@ def _make_estimate_row(created_by: str | None) -> SimpleNamespace:
         sources_used=["avito"],
         data_freshness_minutes=10,
         expires_at=datetime.now(tz=UTC) + timedelta(hours=12),
+        retain_until=retain_until,
         address="ул. Тестовая, 1",
         lat=56.8,
         lon=60.6,
@@ -597,3 +603,116 @@ def test_get_estimate_imv_benchmark_other_pilot_gets_404(trade_in_app: FastAPI)
         headers={"X-Authenticated-User": "attacker"},
     )
     assert resp.status_code == 404
+
+
+# ── Payments retention: retention gate unification (retain_until, PR #2754) ──
+
+
+def test_estimate_readable_sql_uses_disjunction() -> None:
+    """Single definition — OR retain_until, not a hand-copied expression."""
+    from app.api.v1.trade_in import ESTIMATE_READABLE_SQL
+
+    assert "expires_at > NOW()" in ESTIMATE_READABLE_SQL
+    assert "retain_until > NOW()" in ESTIMATE_READABLE_SQL
+    assert " OR " in ESTIMATE_READABLE_SQL
+
+
+def test_get_estimate_sql_built_from_shared_constant() -> None:
+    """GET /estimate/{id} SQL filter is built FROM ESTIMATE_READABLE_SQL, not a
+    hand-copied literal — regression guard against the two gates drifting apart
+    again (that's exactly what happened before this PR: 404 here, 410 in /pdf)."""
+    import inspect
+
+    from app.api.v1.trade_in import get_estimate
+
+    src = inspect.getsource(get_estimate)
+    assert "ESTIMATE_READABLE_SQL" in src
+    assert "expires_at > NOW()" not in src, "hand-copied predicate, not the shared constant"
+    assert "retain_until" in src, "SELECT must also fetch retain_until"
+
+
+def test_estimate_pdf_select_includes_retain_until_column() -> None:
+    import inspect
+
+    from app.api.v1.trade_in import estimate_pdf
+
+    assert "retain_until" in inspect.getsource(estimate_pdf)
+
+
+@pytest.mark.parametrize(
+    ("expires_delta_hours", "retain_delta_days", "expected"),
+    [
+        (12, None, True),  # not expired, unpaid — current B2B/B2C behaviour, unchanged
+        (-1, None, False),  # expired, unpaid — current behaviour (404/410), unchanged
+        (-1, 365, True),  # expired but PAID — new: readable
+        (12, 365, True),  # not expired AND paid — readable
+        (-1, -1, False),  # expired, and the (hypothetical) retain_until also in the past
+    ],
+)
+def test_estimate_readable_truth_table(
+    expires_delta_hours: int, retain_delta_days: int | None, expected: bool
+) -> None:
+    from datetime import UTC, datetime, timedelta
+
+    from app.api.v1.trade_in import estimate_readable
+
+    expires_at = datetime.now(tz=UTC) + timedelta(hours=expires_delta_hours)
+    retain_until = (
+        datetime.now(tz=UTC) + timedelta(days=retain_delta_days)
+        if retain_delta_days is not None
+        else None
+    )
+    assert estimate_readable(expires_at, retain_until) is expected
+
+
+def test_pdf_expired_but_paid_returns_200(trade_in_app: FastAPI) -> None:
+    """expires_at in the past, retain_until in the future → PDF still downloads
+    (200). Exactly the scenario PR-D1 exists for: a paid report must outlive
+    the 24h expires_at link TTL."""
+    from datetime import UTC, datetime, timedelta
+
+    row = _make_estimate_row(created_by="kopylov")
+    row.expires_at = datetime.now(tz=UTC) - timedelta(hours=1)
+    row.retain_until = datetime.now(tz=UTC) + timedelta(days=300)
+    db_mock = _make_db_mock(row)
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
+        headers={"X-Authenticated-User": "kopylov"},
+    )
+    assert resp.status_code == 200
+    assert resp.headers["content-type"] == "application/pdf"
+
+
+def test_pdf_expired_unpaid_returns_410_without_ttl_text(trade_in_app: FastAPI) -> None:
+    """expires_at in the past, retain_until NULL (unpaid, unchanged behaviour) →
+    410, and the detail text no longer claims a specific '24h TTL' (would be a
+    lie now that retain_until exists for paid rows)."""
+    from datetime import UTC, datetime, timedelta
+
+    row = _make_estimate_row(created_by="kopylov")
+    row.expires_at = datetime.now(tz=UTC) - timedelta(hours=1)
+    row.retain_until = None
+    db_mock = _make_db_mock(row)
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
+        headers={"X-Authenticated-User": "kopylov"},
+    )
+    assert resp.status_code == 410
+    assert resp.json()["detail"] == "estimate expired"
+    assert "24h" not in resp.json()["detail"]
+    assert "TTL" not in resp.json()["detail"]
+
+
+def test_get_estimate_response_includes_retain_until_field(trade_in_app: FastAPI) -> None:
+    """Response schema exposes retain_until (nullable) — schemas/trade_in.py."""
+    row = _make_estimate_row(created_by="kopylov")  # retain_until defaults to None
+    db_mock = _make_db_mock(row)
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
+        headers={"X-Authenticated-User": "kopylov"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["retain_until"] is None

tradein-mvp/backend/tests/test_paid_retention_text_consistency.py

@@ -0,0 +1,103 @@
+"""Payments retention (PR #2754) — "12 месяцев" text sync guard.
+
+WHY:
+    mera-public/content.ts declares itself as the ONE place product promises
+    live (docstring at the top of that file: "ни одного утверждения, которого
+    не делает код"). The public retention promise ("оплаченный отчёт хранится
+    N месяцев") has THREE places it could quietly drift: the backend setting
+    (`settings.trade_in_paid_retention_days`), the frontend constant
+    (`PAID_REPORT_RETENTION_MONTHS` in content.ts), and any page that renders
+    it (today: privacy/page.tsx). Deep-review finding 2026-08-06 MEDIUM on
+    PR #2754 caught exactly this: a comment claimed the number "reads from the
+    same setting" while the page actually hardcoded a `12 месяцев` literal --
+    a comment cannot fail CI, same lesson as
+    test_consent_text_frontend_sync.py's _CONSENT_TEXT_SNAPSHOT guard (which
+    this file mirrors).
+
+WHAT:
+    1. privacy/page.tsx imports PAID_REPORT_RETENTION_MONTHS from content.ts
+       and does NOT hardcode a "N месяцев" literal of its own.
+    2. The frontend months constant and the backend days setting stay within
+       a sane calendar tolerance of each other (28-31 days per month) -- this
+       does NOT enforce byte-identity (days and months are different units by
+       design, see content.ts docstring), only that nobody silently changes
+       one without the other drifting out of "still honestly ~12 months".
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from pathlib import Path
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_FRONTEND_CONTENT = (
+    Path(__file__).resolve().parents[2] / "frontend" / "src" / "app" / "mera-public" / "content.ts"
+)
+_FRONTEND_PRIVACY_PAGE = (
+    Path(__file__).resolve().parents[2]
+    / "frontend"
+    / "src"
+    / "app"
+    / "mera-public"
+    / "privacy"
+    / "page.tsx"
+)
+
+_MONTHS_CONST_RE = re.compile(r"PAID_REPORT_RETENTION_MONTHS\s*=\s*(\d+)\s*;")
+_LITERAL_MONTHS_RE = re.compile(r'"?\d+ месяцев"?')
+
+
+def _extract_months_constant(content_ts_source: str) -> int:
+    match = _MONTHS_CONST_RE.search(content_ts_source)
+    assert match is not None, (
+        "PAID_REPORT_RETENTION_MONTHS not found in content.ts -- "
+        "constant renamed/removed without updating this test"
+    )
+    return int(match.group(1))
+
+
+def test_frontend_files_exist() -> None:
+    assert _FRONTEND_CONTENT.is_file(), f"missing frontend file: {_FRONTEND_CONTENT}"
+    assert _FRONTEND_PRIVACY_PAGE.is_file(), f"missing frontend file: {_FRONTEND_PRIVACY_PAGE}"
+
+
+def test_privacy_page_imports_retention_constant_not_hardcoded() -> None:
+    """The whole point: FAILS if privacy/page.tsx stops importing the shared
+    constant and goes back to a hardcoded '12 месяцев' literal (exactly the
+    drift the deep-review finding caught -- comment said 'reads from content.ts',
+    code said otherwise)."""
+    src = _FRONTEND_PRIVACY_PAGE.read_text(encoding="utf-8")
+    assert "PAID_REPORT_RETENTION_MONTHS" in src, (
+        "privacy/page.tsx no longer references PAID_REPORT_RETENTION_MONTHS -- "
+        "the retention promise must be rendered from the shared content.ts "
+        "constant, not written out again by hand"
+    )
+    assert re.search(r'from\s+"\.\./content"', src), (
+        "privacy/page.tsx must import from '../content' (content.ts), where "
+        "PAID_REPORT_RETENTION_MONTHS is declared"
+    )
+    literal_hits = _LITERAL_MONTHS_RE.findall(src)
+    assert not literal_hits, (
+        "privacy/page.tsx contains a hardcoded 'N месяцев' literal -- render the "
+        "PAID_REPORT_RETENTION_MONTHS constant instead: "
+        f"{literal_hits!r}"
+    )
+
+
+def test_backend_days_setting_matches_frontend_months_within_calendar_tolerance() -> None:
+    """Not byte-identity (days vs months are different units, deliberately --
+    see content.ts docstring on PAID_REPORT_RETENTION_MONTHS): just a sanity
+    bound that `trade_in_paid_retention_days` still honestly rounds to the
+    number of months the public page promises (28-31 days/month, generous)."""
+    from app.core.config import settings
+
+    days = settings.trade_in_paid_retention_days
+    months = _extract_months_constant(_FRONTEND_CONTENT.read_text(encoding="utf-8"))
+
+    assert 28 * months <= days <= 31 * months, (
+        f"settings.trade_in_paid_retention_days={days} no longer honestly rounds to "
+        f"content.ts PAID_REPORT_RETENTION_MONTHS={months} -- update both together "
+        "(and the offer text, when it exists) so the public promise stays true"
+    )

tradein-mvp/backend/tests/test_pdf_security.py

@@ -445,3 +445,36 @@ def test_brand_not_taken_from_query_param_docstring() -> None:
     assert (
         "brand" not in param_names
     ), "estimate_pdf should NOT have a 'brand' query param after #7 fix"
+
+
+# ── PR-D1: retain_until (paid retention) — cover row + valid_until unaffected ──
+
+
+def test_cover_no_retain_until_row_when_unpaid() -> None:
+    """retain_until IS NULL (default, all current traffic) → no 'Ссылка доступна
+    до' row at all — B2B regression guard, cover renders bit-for-bit as before."""
+    est = _estimate()
+    assert est.retain_until is None
+    html = mod._build_cover(est, _SNAPSHOT, _GENERIC)
+    assert "Ссылка доступна до" not in html
+
+
+def test_cover_renders_retain_until_row_when_paid() -> None:
+    """retain_until IS NOT NULL → 'Ссылка доступна до <date>' row present, with
+    its OWN date (not conflated with 'Срок действия данных' / expires_at)."""
+    retain = datetime(2027, 8, 6, tzinfo=UTC)
+    est = _estimate(retain_until=retain)
+    html = mod._build_cover(est, _SNAPSHOT, _GENERIC)
+    assert "Ссылка доступна до" in html
+    assert "06.08.2027" in html
+
+
+def test_expires_date_unaffected_by_retain_until() -> None:
+    """«ДЕЙСТВИТЕЛЕН ДО» (running footer, _expires_date) stays wired to
+    expires_at regardless of retain_until — it is data-actuality, not the
+    paid-access retention window, and must not move when a report is paid."""
+    expires = datetime.now(UTC) + timedelta(hours=24)
+    est_unpaid = _estimate(expires_at=expires)
+    est_paid = _estimate(expires_at=expires, retain_until=expires + timedelta(days=365))
+    assert mod._expires_date(est_unpaid) == expires.date()
+    assert mod._expires_date(est_paid) == expires.date()

tradein-mvp/backend/tests/test_purge_expired_trade_in_data.py

@@ -7,9 +7,22 @@ Covers app/tasks/purge_expired_trade_in_data.py:
   - both tables (trade_in_estimates, trade_in_leads) get drained
   - failure path: rollback + mark_failed with partial counters, exception re-raised
   - SQL shape: DELETE (not UPDATE/deactivate), no psycopg `::` cast trap
+  - Payments retention (PR #2754): retain_until IS NULL + NOT EXISTS(payments)
+    safeguards on the estimates DELETE, plus a pre-flight that refuses to run
+    at all if it finds an ANOMALOUS paid purge-candidate (retain_until unset
+    despite a payments row) -- a healthy paid row (retain_until set) must NOT
+    trip it, see test_preflight_ignores_healthy_paid_row below.
 
 Style mirrors tests/test_deactivate_stale_listings.py (_FakeDB, monkeypatched
 runs_mod.mark_done/mark_failed).
+
+Payments retention note on _FakeDB: purge_expired_trade_in_data now issues ONE
+extra db.execute() call BEFORE any DELETE batch — the pre-flight paid-
+candidates count (_PREFLIGHT_PAID_CANDIDATES_SQL). _FakeDB special-cases that
+statement by identity and answers it from `preflight_count` (default 0 ==
+"no anomalous candidates, proceed exactly as before this PR"). Every
+pre-existing test's `db.executed` index shifted by +1 to account for this;
+`db.commits` is unaffected (the pre-flight is a read, never committed).
 """
 
 from __future__ import annotations
@@ -34,18 +47,30 @@ class _FakeResult:
     def __init__(self, rowcount: int) -> None:
         self.rowcount = rowcount
 
+    def scalar_one(self) -> int:
+        """Supports the PR-D1 pre-flight `SELECT count(*) ... .scalar_one()` call."""
+        return self.rowcount
+
 
 class _FakeDB:
-    """Pops rowcounts in call order -- caller supplies the exact sequence expected."""
+    """Pops rowcounts in call order -- caller supplies the exact sequence expected.
 
-    def __init__(self, rowcounts: list[int]) -> None:
+    PR-D1: the pre-flight paid-candidates count is answered separately, from
+    `preflight_count` (default 0), keyed by statement IDENTITY -- it never
+    consumes an entry off `rowcounts` (that list is DELETE-batch rowcounts only).
+    """
+
+    def __init__(self, rowcounts: list[int], *, preflight_count: int = 0) -> None:
         self._rowcounts = list(rowcounts)
+        self.preflight_count = preflight_count
         self.executed: list[tuple[Any, Any]] = []
         self.commits = 0
         self.rolled_back = False
 
     def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
         self.executed.append((stmt, params))
+        if stmt is task_mod._PREFLIGHT_PAID_CANDIDATES_SQL:
+            return _FakeResult(self.preflight_count)
         return _FakeResult(self._rowcounts.pop(0))
 
     def commit(self) -> None:
@@ -83,7 +108,9 @@ def test_stops_when_batch_below_size(monkeypatch: pytest.MonkeyPatch) -> None:
     db = _FakeDB([3, 0])
     out = task_mod.purge_expired_trade_in_data(db, run_id=1, batch_size=10, max_batches=20)  # type: ignore[arg-type]
     assert out == {"estimates_deleted": 3, "leads_deleted": 0}
-    assert len(db.executed) == 2
+    # +1 vs pre-PR-D1: db.executed now also holds the pre-flight paid-candidates
+    # count (call #1), issued before either DELETE batch.
+    assert len(db.executed) == 3
     assert db.commits == 2
     assert marked["counters"] == out
 
@@ -94,7 +121,7 @@ def test_loops_until_below_batch_size(monkeypatch: pytest.MonkeyPatch) -> None:
     db = _FakeDB([5, 5, 2, 5, 1])
     out = task_mod.purge_expired_trade_in_data(db, run_id=2, batch_size=5, max_batches=20)  # type: ignore[arg-type]
     assert out == {"estimates_deleted": 12, "leads_deleted": 6}
-    assert len(db.executed) == 5
+    assert len(db.executed) == 6  # +1: pre-flight call before the 5 DELETE batches
     assert db.commits == 5, "each batch must commit independently, not one final commit"
 
 
@@ -106,14 +133,16 @@ def test_respects_max_batches_cap(monkeypatch: pytest.MonkeyPatch) -> None:
     db = _FakeDB([5, 5, 5, 5, 5, 5])  # exactly max_batches=3 per table, no more
     out = task_mod.purge_expired_trade_in_data(db, run_id=3, batch_size=5, max_batches=3)  # type: ignore[arg-type]
     assert out == {"estimates_deleted": 15, "leads_deleted": 15}
-    assert len(db.executed) == 6  # 3 (estimates) + 3 (leads), NOT unbounded
+    assert len(db.executed) == 7  # pre-flight + 3 (estimates) + 3 (leads), NOT unbounded
 
 
 def test_default_batch_size_and_max_batches_from_settings(monkeypatch: pytest.MonkeyPatch) -> None:
     _patch_runs(monkeypatch)
     db = _FakeDB([0, 0])  # first batch already empty on both tables -> stop immediately
     task_mod.purge_expired_trade_in_data(db, run_id=4)  # type: ignore[arg-type]
-    _stmt, params = db.executed[0]
+    # db.executed[0] is now the pre-flight call (no batch_size param) -- the
+    # first DELETE-batch call (with batch_size) shifted to index 1.
+    _stmt, params = db.executed[1]
     assert params is not None
     assert params["batch_size"] == task_mod.settings.trade_in_purge_batch_size
 
@@ -125,8 +154,9 @@ def test_drains_both_tables_in_order(monkeypatch: pytest.MonkeyPatch) -> None:
     _patch_runs(monkeypatch)
     db = _FakeDB([0, 0])
     task_mod.purge_expired_trade_in_data(db, run_id=5, batch_size=100, max_batches=1)  # type: ignore[arg-type]
-    first_sql = str(getattr(db.executed[0][0], "text", db.executed[0][0]))
-    second_sql = str(getattr(db.executed[1][0], "text", db.executed[1][0]))
+    # index 0 is now the pre-flight call; DELETE batches shifted to 1/2.
+    first_sql = str(getattr(db.executed[1][0], "text", db.executed[1][0]))
+    second_sql = str(getattr(db.executed[2][0], "text", db.executed[2][0]))
     assert "trade_in_estimates" in first_sql
     assert "trade_in_leads" in second_sql
 
@@ -141,6 +171,86 @@ def test_estimates_sql_is_delete_not_update() -> None:
     assert not re.search(r":\w+::", sql)
 
 
+# ── Payments retention (PR #2754): two independent purge safeguards ─────────
+
+
+def test_estimates_sql_excludes_retain_until_not_null() -> None:
+    """Phase 1: exactly `retain_until IS NULL`, never `< NOW()` -- a paid row
+    (retain_until IS NOT NULL) must never match the DELETE predicate, full stop,
+    regardless of how far in the past that date eventually sits."""
+    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
+    assert "retain_until IS NULL" in sql
+    assert "retain_until <" not in sql, "phase 1 must not weaken to retain_until < NOW()"
+
+
+def test_estimates_sql_has_not_exists_payments_safeguard() -> None:
+    """Independent safeguard: a row with ANY payments row survives even if
+    retain_until failed to be set (fulfillment bug/race/manual INSERT)."""
+    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
+    assert "NOT EXISTS" in sql
+    assert "FROM payments p" in sql
+    assert "p.estimate_id = trade_in_estimates.id" in sql
+
+
+def test_preflight_sql_requires_retain_until_is_null() -> None:
+    """Deep-review finding 2026-08-06 MEDIUM (PR #2754): the pre-flight predicate
+    MUST carry `retain_until IS NULL` -- without it, a perfectly healthy paid row
+    (retain_until set, has a payments row -- the ORDINARY steady state one day
+    after every sale) trips the alarm exactly as hard as the real anomaly
+    (retain_until unset despite a payments row existing), permanently wedging
+    the job on the very first successful sale (and, since leads purge runs from
+    the same function AFTER this check, silently stopping 180-day 152-ФЗ lead
+    retention too). See test_real_preflight_ignores_healthy_paid_row below for
+    the behavioural proof against a real DB."""
+    sql = task_mod._PREFLIGHT_PAID_CANDIDATES_SQL.text
+    assert "expires_at < NOW()" in sql
+    assert "created_by IS NULL" in sql
+    assert "retain_until IS NULL" in sql
+    assert "EXISTS (SELECT 1 FROM payments p WHERE p.estimate_id = e.id)" in sql
+    assert not re.search(r":\w+::", sql)
+
+
+def test_preflight_aborts_before_any_delete_batch(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Non-zero pre-flight count -> RuntimeError, mark_failed, ZERO DELETE batches
+    ever issued (only the pre-flight SELECT itself is in db.executed)."""
+    marked = _patch_runs(monkeypatch)
+    db = _FakeDB([], preflight_count=2)  # rowcounts empty on purpose: must never be popped
+    with pytest.raises(RuntimeError, match="pre-flight abort"):
+        task_mod.purge_expired_trade_in_data(db, run_id=42, batch_size=10, max_batches=20)  # type: ignore[arg-type]
+
+    assert len(db.executed) == 1, "only the pre-flight SELECT -- no DELETE batch was issued"
+    assert db.commits == 0
+    assert marked["kind"] == "failed"
+    assert marked["run_id"] == 42
+    assert marked["counters"] == {"estimates_deleted": 0, "leads_deleted": 0}
+    assert "2" in marked["err"]
+
+
+def test_preflight_zero_candidates_proceeds_as_before(monkeypatch: pytest.MonkeyPatch) -> None:
+    """preflight_count=0 (default) -- the exact pre-PR-D1 behaviour for every
+    row that exists today (all retain_until IS NULL) -- run proceeds normally."""
+    marked = _patch_runs(monkeypatch)
+    db = _FakeDB([0, 0])  # preflight_count defaults to 0
+    out = task_mod.purge_expired_trade_in_data(db, run_id=43, batch_size=10, max_batches=20)  # type: ignore[arg-type]
+    assert out == {"estimates_deleted": 0, "leads_deleted": 0}
+    assert marked["kind"] == "done"
+
+
+def test_leads_sql_unchanged_by_pr_d1() -> None:
+    """Snapshot: _DELETE_EXPIRED_LEADS_SQL byte-for-byte unchanged by payments
+    retention (PR #2754) — leads have their own retention deadline (migration
+    231, no created_by/B2B split, no payments concept) and are explicitly out
+    of scope for the payments-retention safeguards."""
+    expected = (
+        "\n    DELETE FROM trade_in_leads\n    WHERE id IN (\n"
+        "        SELECT id FROM trade_in_leads\n"
+        "        WHERE expires_at < NOW()\n"
+        "        ORDER BY expires_at\n"
+        "        LIMIT CAST(:batch_size AS int)\n    )\n    "
+    )
+    assert task_mod._DELETE_EXPIRED_LEADS_SQL.text == expected
+
+
 def test_leads_sql_is_delete_not_update() -> None:
     sql = task_mod._DELETE_EXPIRED_LEADS_SQL.text
     assert "DELETE FROM trade_in_leads" in sql
@@ -199,13 +309,14 @@ def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch)
 
     class _BoomDB(_FakeDB):
         def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
-            # First batch (estimates) succeeds and commits; second call (still
-            # draining estimates, or first leads call) explodes.
-            if len(self.executed) >= 1:
+            # Call #1 (pre-flight, preflight_count=0) and call #2 (first estimates
+            # batch) succeed and commit; call #3 (still draining estimates, or
+            # first leads call) explodes. +1 vs pre-PR-D1 to admit the pre-flight.
+            if len(self.executed) >= 2:
                 raise RuntimeError("db exploded")
             return super().execute(stmt, params)
 
-    db = _BoomDB([5])  # only ONE successful batch before the boom
+    db = _BoomDB([5])  # only ONE successful DELETE batch before the boom
     with pytest.raises(RuntimeError, match="db exploded"):
         task_mod.purge_expired_trade_in_data(db, run_id=6, batch_size=5, max_batches=20)  # type: ignore[arg-type]
 
@@ -339,3 +450,160 @@ def test_real_purge_deletes_only_anonymous_expired_estimates() -> None:
         )
         db.commit()
         db.close()
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_preflight_ignores_healthy_paid_row_flags_only_anomaly() -> None:
+    """Deep-review finding 2026-08-06 MEDIUM on PR #2754, reproduced exactly
+    against a real DB: a HEALTHY paid row (retain_until set, a payments row
+    exists) is the ordinary steady state one day after every sale and must NOT
+    raise the pre-flight count; a row where fulfillment failed to set
+    retain_until despite a payments row existing is the real ANOMALY and must.
+    Baseline-delta assertions (not absolute counts) so this is safe to run
+    against a dev DB that may already contain unrelated rows."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    healthy_id = uuid4()
+    anomaly_id = uuid4()
+    healthy_order = f"pytest-healthy-{uuid4().hex[:12]}"
+    anomaly_order = f"pytest-anomaly-{uuid4().hex[:12]}"
+    try:
+        baseline = task_mod._preflight_paid_candidates(db)
+
+        # Healthy: retain_until set (paid, safeguard 1 already protects it) +
+        # a payments row -- exactly what every successful sale looks like a day
+        # later. Must NOT move the pre-flight count.
+        db.execute(
+            _t(
+                "INSERT INTO trade_in_estimates "
+                "(id, address, area_m2, rooms, floor, total_floors, "
+                " median_price, range_low, range_high, median_price_per_m2, confidence, "
+                " expires_at, created_by, retain_until) VALUES "
+                "(CAST(:id AS uuid), 'purge-test здоровая оплаченная', 40, 1, 2, 5, "
+                " 5000000, 4500000, 5500000, 125000, 'low', "
+                " NOW() - interval '1 hour', NULL, NOW() + interval '363 days')"
+            ),
+            {"id": str(healthy_id)},
+        )
+        db.execute(
+            _t(
+                "INSERT INTO payments "
+                "(order_id, terminal_key, product_code, amount_kopecks, status, estimate_id) "
+                "VALUES (:order_id, 'pytest_terminal', 'trade_in_report', 15000, 'CONFIRMED', "
+                " CAST(:id AS uuid))"
+            ),
+            {"order_id": healthy_order, "id": str(healthy_id)},
+        )
+        db.commit()
+        assert (
+            task_mod._preflight_paid_candidates(db) == baseline
+        ), "healthy paid row (retain_until set) must NOT raise the pre-flight count"
+
+        # Anomaly: retain_until NULL despite a payments row existing -- exactly
+        # the case the two DELETE safeguards exist for. Must raise by exactly one.
+        db.execute(
+            _t(
+                "INSERT INTO trade_in_estimates "
+                "(id, address, area_m2, rooms, floor, total_floors, "
+                " median_price, range_low, range_high, median_price_per_m2, confidence, "
+                " expires_at, created_by, retain_until) VALUES "
+                "(CAST(:id AS uuid), 'purge-test настоящая аномалия', 40, 1, 2, 5, "
+                " 5000000, 4500000, 5500000, 125000, 'low', "
+                " NOW() - interval '1 hour', NULL, NULL)"
+            ),
+            {"id": str(anomaly_id)},
+        )
+        db.execute(
+            _t(
+                "INSERT INTO payments "
+                "(order_id, terminal_key, product_code, amount_kopecks, status, estimate_id) "
+                "VALUES (:order_id, 'pytest_terminal', 'trade_in_report', 15000, 'CONFIRMED', "
+                " CAST(:id AS uuid))"
+            ),
+            {"order_id": anomaly_order, "id": str(anomaly_id)},
+        )
+        db.commit()
+        assert task_mod._preflight_paid_candidates(db) == baseline + 1, (
+            "anomaly row (retain_until unset, payments row exists) must raise "
+            "the pre-flight count by exactly one"
+        )
+    finally:
+        db.execute(
+            _t("DELETE FROM payments WHERE order_id = ANY(CAST(:orders AS text[]))"),
+            {"orders": [healthy_order, anomaly_order]},
+        )
+        db.execute(
+            _t("DELETE FROM trade_in_estimates WHERE id = ANY(CAST(:ids AS uuid[]))"),
+            {"ids": [str(healthy_id), str(anomaly_id)]},
+        )
+        db.commit()
+        db.close()
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_purge_not_wedged_by_healthy_paid_row() -> None:
+    """Deep-review finding 2026-08-06 MEDIUM on PR #2754: before the fix, a
+    healthy paid row anywhere in the table (retain_until set, has a payments
+    row) permanently wedged the job -- the very first successful sale would
+    have made every subsequent scheduled run abort in mark_failed with zero
+    deletions FOREVER, silently taking 180-day leads purge (152-ФЗ) down with
+    it (leads purge runs from the same function AFTER the pre-flight check).
+    This proves a real end-to-end run completes normally (mark_done) in the
+    presence of such a row."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    healthy_id = uuid4()
+    healthy_order = f"pytest-wedge-{uuid4().hex[:12]}"
+    try:
+        db.execute(
+            _t(
+                "INSERT INTO trade_in_estimates "
+                "(id, address, area_m2, rooms, floor, total_floors, "
+                " median_price, range_low, range_high, median_price_per_m2, confidence, "
+                " expires_at, created_by, retain_until) VALUES "
+                "(CAST(:id AS uuid), 'purge-test не блокирует джобу', 40, 1, 2, 5, "
+                " 5000000, 4500000, 5500000, 125000, 'low', "
+                " NOW() - interval '1 hour', NULL, NOW() + interval '363 days')"
+            ),
+            {"id": str(healthy_id)},
+        )
+        db.execute(
+            _t(
+                "INSERT INTO payments "
+                "(order_id, terminal_key, product_code, amount_kopecks, status, estimate_id) "
+                "VALUES (:order_id, 'pytest_terminal', 'trade_in_report', 15000, 'CONFIRMED', "
+                " CAST(:id AS uuid))"
+            ),
+            {"order_id": healthy_order, "id": str(healthy_id)},
+        )
+        db.commit()
+
+        # Must complete normally -- no RuntimeError, no mark_failed short-circuit
+        # (would raise before reaching this line if the bug were still present).
+        result = task_mod.purge_expired_trade_in_data(
+            db, run_id=999999998, batch_size=100, max_batches=1
+        )
+        assert set(result) == {"estimates_deleted", "leads_deleted"}, (
+            "leads purge must also have run -- it is NOT reachable when the "
+            "pre-flight wrongly aborts first"
+        )
+
+        still_there = db.execute(
+            _t("SELECT id FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
+            {"id": str(healthy_id)},
+        ).fetchone()
+        assert still_there is not None, "healthy paid row must survive the run untouched"
+    finally:
+        db.execute(
+            _t("DELETE FROM payments WHERE order_id = :order_id"), {"order_id": healthy_order}
+        )
+        db.execute(
+            _t("DELETE FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
+            {"id": str(healthy_id)},
+        )
+        db.commit()
+        db.close()

tradein-mvp/frontend/package-lock.json

@@ -6343,6 +6343,19 @@
         "node": ">=8.6"
       }
     },
+    "node_modules/micromatch/node_modules/picomatch": {
+      "version": "2.3.2",
+      "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.2.tgz",
+      "integrity": "sha512-V7+vQEJ06Z+c5tSye8S+nHUfI51xoXIXjHQ99cQtKUkQqqO1kO/KCJUfZXuB47h/YBlDhah2H3hdUGXn8ie0oA==",
+      "dev": true,
+      "license": "MIT",
+      "engines": {
+        "node": ">=8.6"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/jonschlinkert"
+      }
+    },
     "node_modules/mime-db": {
       "version": "1.52.0",
       "resolved": "https://registry.npmjs.org/mime-db/-/mime-db-1.52.0.tgz",
@@ -6803,13 +6816,13 @@
       "license": "ISC"
     },
     "node_modules/picomatch": {
-      "version": "2.3.2",
-      "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.2.tgz",
-      "integrity": "sha512-V7+vQEJ06Z+c5tSye8S+nHUfI51xoXIXjHQ99cQtKUkQqqO1kO/KCJUfZXuB47h/YBlDhah2H3hdUGXn8ie0oA==",
+      "version": "4.0.5",
+      "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.5.tgz",
+      "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==",
       "dev": true,
       "license": "MIT",
       "engines": {
-        "node": ">=8.6"
+        "node": ">=12"
       },
       "funding": {
         "url": "https://github.com/sponsors/jonschlinkert"
@@ -7828,19 +7841,6 @@
         "url": "https://github.com/sponsors/SuperchupuDev"
       }
     },
-    "node_modules/tinyglobby/node_modules/picomatch": {
-      "version": "4.0.5",
-      "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.5.tgz",
-      "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==",
-      "dev": true,
-      "license": "MIT",
-      "engines": {
-        "node": ">=12"
-      },
-      "funding": {
-        "url": "https://github.com/sponsors/jonschlinkert"
-      }
-    },
     "node_modules/tinypool": {
       "version": "1.1.1",
       "resolved": "https://registry.npmjs.org/tinypool/-/tinypool-1.1.1.tgz",

tradein-mvp/frontend/src/app/mera-public/content.ts

@@ -92,6 +92,25 @@ export const LEGAL_ENTITY: {
 /** Внутренний маршрут страницы про обработку персональных данных. */
 export const PRIVACY_PATH = "/mera-public/privacy";
 
+/**
+ * Сколько месяцев на нашей стороне хранится ссылка/строка оплаченного отчёта
+ * после оплаты (`trade_in_estimates.retain_until`, migration 240) — НЕ срок
+ * действия самого расчёта (тот отдельный, `expires_at`, часы).
+ *
+ * ЕДИНСТВЕННОЕ место, где это число хардкодится на фронте — любой другой
+ * текст (оферта, экран после оплаты) обязан импортировать эту константу, а
+ * не писать «12 месяцев» заново (см. `test_paid_retention_text_consistency.py`
+ * — проверяет, что privacy-страница действительно использует эту константу,
+ * а не литерал).
+ *
+ * Источник истины на бэкенде — `settings.trade_in_paid_retention_days = 365`
+ * (`app/core/config.py`). 365 дней округляется до «12 месяцев» для
+ * человекочитаемого текста (12×30=360..12×31=372 — 365 попадает в диапазон);
+ * если бэкендовое число когда-нибудь изменится так, что «12» перестанет быть
+ * честным округлением — обновить оба места руками, тест это тоже проверяет.
+ */
+export const PAID_REPORT_RETENTION_MONTHS = 12;
+
 // ---------------------------------------------------------------------------
 // География
 // ---------------------------------------------------------------------------

tradein-mvp/frontend/src/app/mera-public/privacy/page.tsx

@@ -3,6 +3,7 @@ import Link from "next/link";
 
 import {
   LEGAL_ENTITY,
+  PAID_REPORT_RETENTION_MONTHS,
   PUBLIC_ESTIMATE_ENABLED,
   SUPPORT_TELEGRAM_LABEL,
   SUPPORT_TELEGRAM_URL,
@@ -32,11 +33,26 @@ import { safeUrl } from "@/lib/safeUrl";
  *     квалификация — не наше дело: заявка привязывается к конкретному расчёту
  *     (`TradeInLeadInput.estimate_id`), то есть телефон связывается с ранее
  *     сохранённым адресом. Финальную формулировку даёт юрист.
- *   - «мы удалим ваш телефон и заявку». Механизма удаления в бэкенде НЕТ:
- *     ни `DELETE FROM trade_in_leads/trade_in_estimates` в коде, ни
- *     retention/erasure-джоба среди `app/tasks/**` (проверено grep'ом);
- *     `expires_at` применяется только на чтении. Обещать удаление до появления
- *     процедуры нельзя — это самое дорогое из обещаний.
+ *   - «мы обещаем удалить всё вообще» без оговорок. После #2547 механизм
+ *     удаления в бэкенде ЕСТЬ: `app/services/data_erasure.py` (ручное
+ *     удаление по обращению — по estimate_id/телефону/Telegram chat id) и
+ *     `app/tasks/purge_expired_trade_in_data.py` (автоматическое удаление по
+ *     истечении срока хранения). Прежняя формулировка «механизма нет» стала
+ *     неправдой и здесь больше не пишется. Оговорка, которая остаётся честной:
+ *     копия сообщения в Telegram-группе поддержки этим механизмом не
+ *     стирается (см. докстринг `data_erasure.py`) — сюда её не выносим
+ *     (излишняя техническая деталь для публичной страницы), но это ограничение
+ *     реальное и известное.
+ *   - Срок хранения оплаченного отчёта (PR #2754, `retain_until` /
+ *     `trade_in_paid_retention_days` в backend `config.py`): число месяцев
+ *     ниже — константа `PAID_REPORT_RETENTION_MONTHS` из `../content.ts`,
+ *     НЕ литерал здесь (см. докстринг константы — она единственный источник
+ *     этого числа на фронте; `test_paid_retention_text_consistency.py`
+ *     проверяет, что эта страница действительно её импортирует). Платёжного
+ *     кода в этом PR нет — срок описан на будущее, вместе с честной правкой
+ *     ниже про то, что удаление сегодня описывает установленный ПОРЯДОК, а
+ *     не наблюдаемый на проде автоматический прогон (задача засеяна
+ *     выключенной).
  *
  * Раздел «Что делает эта страница» УСЛОВЕН по `PUBLIC_ESTIMATE_ENABLED`: пока
  * расчёт выключен, адрес действительно не покидает браузер; после включения это
@@ -136,10 +152,17 @@ export default function MeraPublicPrivacyPage() {
           ) : null}
         </p>
         <p>
-          Автоматической кнопки «удалить мои данные» в сервисе пока нет, и мы не
-          обещаем то, чего не умеем: порядок и сроки удаления будут описаны в
-          утверждённой политике обработки, которая появится здесь до открытия
-          публичного доступа.
+          Самостоятельной кнопки «удалить мои данные» в интерфейсе пока нет, но
+          механизм удаления в сервисе есть: обращение в поддержку об удалении мы
+          разбираем вручную и физически стираем телефон, адрес и расчёт из базы,
+          а не просто помечаем запись. Помимо запроса, для каждого типа данных
+          установлен срок хранения, по истечении которого они подлежат
+          удалению. Если результат расчёта оплачен, ссылка на отчёт и
+          связанные с ним данные хранятся на нашей стороне{" "}
+          {PAID_REPORT_RETENTION_MONTHS} месяцев с даты оплаты, а затем
+          подлежат удалению так же, как и остальные данные — на файл, который
+          вы скачали себе, это не влияет: мы его не отзываем, не изменяем и не
+          имеем к нему доступа.
         </p>
 
         <h2>Оператор</h2>

tradein-mvp/frontend/src/app/v2/page.tsx

@@ -123,7 +123,7 @@ const brackets: { key: string; style: CSSProperties }[] = [
 // Honest neutral fallbacks for the meta blocks (HeroBar / Footer) before there
 // is an estimate. Dashes — never the design fixtures (which would read as a fake
 // real report).
-const EMPTY_REPORT: Report = { id: "—", date: "—", validUntil: "—" };
+const EMPTY_REPORT: Report = { id: "—", date: "—", validUntil: "—", retainUntil: null };
 const EMPTY_OBJECT: ObjectInfo = {
   address: "—",
   city: "",

tradein-mvp/frontend/src/components/trade-in/v2/mappers.ts

@@ -782,6 +782,9 @@ export function mapReport(e: AggregatedEstimate): Report {
       ? fmtDate(e.created_at)
       : fmtDateShift(e.expires_at, -24), // pre-BE-1 fallback: no real created_at
     validUntil: fmtDate(e.expires_at),
+    // PR-D1: passthrough only, validUntil above stays on expires_at (this
+    // field is the paid-access retention date, unrelated to report validity).
+    retainUntil: e.retain_until ?? null,
   };
 }
 

tradein-mvp/frontend/src/components/trade-in/v2/types.ts

@@ -7,6 +7,10 @@ export interface Report {
   id: string;
   date: string;
   validUntil: string;
+  // PR-D1: raw ISO retain_until passthrough (paid-access link lifetime), null
+  // when unpaid (current traffic). NOT rendered yet -- no payment UI in this
+  // PR; validUntil stays wired to expires_at, unrelated to this field.
+  retainUntil: string | null;
 }
 
 export interface ObjectInfo {

tradein-mvp/frontend/src/types/trade-in.ts

@@ -172,6 +172,10 @@ export interface AggregatedEstimate {
   analogs: AnalogLot[]; // top 5-10
   actual_deals: AnalogLot[]; // last 12 mo
   expires_at: string; // ISO datetime
+  // PR-D1: срок жизни ссылки/строки (оплаченный доступ, backend migration 240),
+  // НЕ путать с expires_at (актуальность расчёта). null = неоплачено (текущий
+  // трафик целиком).
+  retain_until?: string | null; // ISO datetime
   // ── Метаданные ──
   target_address: string | null;
   target_lat: number | null;