feat(tradein/payments): оплаченный отчёт хранится год — retain_until и предохранители в задаче удаления

Мина: purge_expired_trade_in_data (сейчас enabled=false) удаляет строки
WHERE expires_at < NOW() AND created_by IS NULL — это ровно популяция
будущих платящих физлиц (владелец продаёт отчёт за 150 руб., отчёт должен
жить год на нашей стороне, а не 24ч). Первый прогон после запуска продаж
безвозвратно снёс бы оплаченное.

Делается ДО платёжного кода, которого в этом PR нет:
- migration 234: колонка trade_in_estimates.retain_until (NULL = неоплачено,
  бэкенд-бита-в-бит не меняется) + частичный индекс под purge-предикат.
- config.py: trade_in_paid_retention_days=365 (ENV) — единственный источник
  "12 месяцев" для будущей оферты/экрана/SQL продления.
- Единый гейт чтения ESTIMATE_READABLE_SQL + estimate_readable() — раньше
  SQL-фильтр (404) и Python-проверка (410) в trade_in.py уже разошлись по
  тексту ответа; текст "estimate expired (24h TTL)" убран (стал бы ложью при
  годовом хранении).
- purge_expired_trade_in_data: retain_until IS NULL (не < NOW() — оплаченное
  не удаляем в принципе) + NOT EXISTS(payments) как независимая страховка +
  pre-flight, который считает оплаченных кандидатов и падает в mark_failed
  ДО первого батча при ненулевом результате.
- PDF: "Ссылка доступна до …" только при retain_until IS NOT NULL;
  "ДЕЙСТВИТЕЛЕН ДО" (expires_at, актуальность расчёта) не тронут.
- Фронт: retain_until прокинут в mapper (validUntil остаётся на expires_at).
- privacy-страница: убрано устаревшее "механизма удаления нет" (неправда
  после #2547), добавлен срок 12 месяцев для оплаченных отчётов.

Ни строчки платёжного кода. expires_at, trade_in_estimate_retention_hours,
_DELETE_EXPIRED_LEADS_SQL не тронуты.

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) поведение не
+# меняется вообще. Не копировать это выражение по месту — только через
+# константу/хелпер ниже. См. `mera-pr-d-spec.md` §1.3 в корне репо.
+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,17 @@ class Settings(BaseSettings):
     # срок — решение DPO/юриста, не инженера). ENV: TRADE_IN_LEAD_RETENTION_DAYS.
     trade_in_lead_retention_days: int = 180
 
+    # ── PR-D1: платный отчёт живёт год (retain_until, migration 234) ────────
+    # trade_in_estimates.retain_until TTL (дни ОТ ОПЛАТЫ) — срок жизни ССЫЛКИ/
+    # СТРОКИ для оплаченной оценки, независимый от expires_at (актуальность
+    # расчёта, 24ч). НЕ трогает expires_at — см. migration 234 докстринг и
+    # `mera-pr-d-spec.md` §1.1/§1.2 в корне репо. Единственный источник числа
+    # «12 месяцев»: текст оферты (content.ts), текст экрана S4 и 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 234).
+    # 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,17 @@ 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.
+
+PR-D1 (2026-08-06, payments retention -- see `mera-pr-d-spec.md` §1 at repo root):
+    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. 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 a paid candidate -- see the SQL constants and
+    `_preflight_paid_candidates` below for the mechanics. No payment code lives in
+    this file; `retain_until` is set by the (separate, not-yet-existing) payment
+    fulfillment code.
 """
 
 from __future__ import annotations
@@ -74,6 +85,21 @@ logger = logging.getLogger(__name__)
 # remainder simply drains on the next nightly run (idempotent, no data loss risk).
 _DEFAULT_MAX_BATCHES = 20
 
+#
+# PR-D1 (2026-08-06): два независимые предохранителя добавлены к тому же
+# предикату, ПЕРЕД тем как платёжный код появился в проекте (мина уже была
+# заряжена, см. `mera-pr-d-spec.md` §1 в корне репо):
+#   1. `retain_until IS NULL` — именно IS NULL, НЕ `< NOW()`. Оплаченная
+#      строка (retain_until IS NOT NULL, migration 234) не удаляется джобой
+#      В ПРИНЦИПЕ, пока не поднято ослабление отдельным 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 +107,31 @@ _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)
     )
     """
 )
 
+# PR-D1 pre-flight (см. _preflight_paid_candidates): считает по БАЗОВОМУ
+# (пред-PR-D1) предикату purge -- `expires_at < NOW() AND created_by IS NULL`,
+# БЕЗ retain_until/NOT EXISTS -- специально ШИРЕ итогового DELETE-предиката
+# выше, чтобы поймать именно случай "retain_until не проставлен, а деньги
+# были" (а не только штатно защищённые retain_until IS NOT NULL строки,
+# которые и так не попали бы под DELETE).
+_PREFLIGHT_PAID_CANDIDATES_SQL = text(
+    """
+    SELECT count(*) FROM trade_in_estimates e
+    WHERE e.expires_at < NOW()
+      AND e.created_by 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 +180,19 @@ def _drain_expired(
             break  # caught up -- fewer expired rows left than one batch
 
 
+def _preflight_paid_candidates(db: Session) -> int:
+    """PR-D1 safety gate: count purge-candidates (base predicate) that have a payments row.
+
+    Runs BEFORE any DELETE batch. A non-zero result means at least one estimate that
+    would have matched the OLD (pre-PR-D1) purge predicate was actually touched by
+    money -- either `retain_until` failed to be set (fulfillment bug/race/manual
+    INSERT) or something inconsistent happened. Either way this run must not delete
+    anything; see `purge_expired_trade_in_data` below, which aborts before the first
+    batch when this returns non-zero.
+    """
+    return db.execute(_PREFLIGHT_PAID_CANDIDATES_SQL).scalar_one()
+
+
 def purge_expired_trade_in_data(
     db: Session,
     run_id: int,
@@ -148,10 +206,28 @@ 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}.
+
+    PR-D1 pre-flight (see `_preflight_paid_candidates`): if any purge-candidate
+    estimate has a `payments` row, the run aborts BEFORE the first DELETE batch --
+    zero rows deleted, `mark_failed` records why. 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 a matching payments row (retain_until may be unset) -- "
+            "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/234_trade_in_estimates_retain_until.sql

@@ -0,0 +1,65 @@
+-- 234_trade_in_estimates_retain_until.sql
+-- PR-D1 «Ретеншен: оплаченное живёт год, purge его не трогает» — см.
+-- `mera-pr-d-spec.md` §1 в корне репо (обоснования там, здесь только SQL).
+-- Номер сверен и по `forgejo/main`, и по всем открытым PR-веткам на момент
+-- написания (последняя занятая — 233_payments.sql) — см. урок в шапке того
+-- же файла про то, как коллизия 228/229/231/232 обнаруживается поздно.
+--
+-- ── 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.
+
+BEGIN;
+
+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

@@ -231,3 +231,4 @@
 # поддержки, #2532/#2533) откладывались до подтверждения, что они осели на
 # проде в финальном виде. Они в _schema_migrations — условие выполнено.
 233_payments.sql
+234_trade_in_estimates_retain_until.sql

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 234) -- 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
+
+
+# ── PR-D1: retention gate unification (retain_until, mera-pr-d-spec.md §1.3) ──
+
+
+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_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,20 @@ 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
+  - PR-D1 (payments retention, mera-pr-d-spec.md §1.4): 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 a paid purge-candidate.
 
 Style mirrors tests/test_deactivate_stale_listings.py (_FakeDB, monkeypatched
 runs_mod.mark_done/mark_failed).
+
+PR-D1 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 paid
+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 +45,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 +106,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 +119,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 +131,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 +152,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 +169,81 @@ def test_estimates_sql_is_delete_not_update() -> None:
     assert not re.search(r":\w+::", sql)
 
 
+# ── PR-D1 (mera-pr-d-spec.md §1.4): 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_is_wider_than_delete_predicate() -> None:
+    """Pre-flight counts by the BASE (pre-PR-D1) predicate -- expires_at < NOW()
+    AND created_by IS NULL, WITHOUT retain_until/NOT EXISTS -- so it also catches
+    the case those two terms exist specifically to guard against (retain_until
+    unset despite a payments row existing)."""
+    sql = task_mod._PREFLIGHT_PAID_CANDIDATES_SQL.text
+    assert "expires_at < NOW()" in sql
+    assert "created_by IS NULL" in sql
+    assert "retain_until" not 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 PR-D1 —
+    leads have their own retention deadline (migration 231) and are explicitly
+    out of scope (mera-pr-d-spec.md §1.4: '_DELETE_EXPIRED_LEADS_SQL — оставить
+    дословно')."""
+    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 +302,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]
 

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

@@ -32,11 +32,21 @@ 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-D1 (`retain_until`, `trade_in_paid_retention_days`): срок «12 месяцев»
+ *     ниже читается из той же настройки, что и оферта/SQL продления —
+ *     см. `mera-pr-d-spec.md` §1.2 в корне репо. Платёжного кода в этом PR
+ *     нет — срок описан на будущее, синхронно с privacy-обязательством #1.7
+ *     того же дока, а не «потом».
  *
  * Раздел «Что делает эта страница» УСЛОВЕН по `PUBLIC_ESTIMATE_ENABLED`: пока
  * расчёт выключен, адрес действительно не покидает браузер; после включения это
@@ -136,10 +146,16 @@ export default function MeraPublicPrivacyPage() {
           ) : null}
         </p>
         <p>
-          Автоматической кнопки «удалить мои данные» в сервисе пока нет, и мы не
-          обещаем то, чего не умеем: порядок и сроки удаления будут описаны в
-          утверждённой политике обработки, которая появится здесь до открытия
-          публичного доступа.
+          Самостоятельной кнопки «удалить мои данные» в интерфейсе пока нет, но
+          механизм удаления в сервисе есть: обращение в поддержку об удалении мы
+          разбираем вручную и физически стираем телефон, адрес и расчёт из базы,
+          а не просто помечаем запись. Помимо запроса, у данных есть собственный
+          срок хранения, по истечении которого они удаляются без обращения с
+          вашей стороны. Если результат расчёта оплачен, ссылка на отчёт и
+          связанные с ним данные хранятся на нашей стороне 12 месяцев с даты
+          оплаты, после чего удаляются точно так же — на файл, который вы
+          скачали себе, это не влияет: мы его не отзываем, не изменяем и не
+          имеем к нему доступа.
         </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/fixtures.ts

@@ -30,6 +30,7 @@ export const report: Report = {
   id: "9b9e2737",
   date: "20.06.2026",
   validUntil: "21.06.2026",
+  retainUntil: null,
 };
 
 export const object: ObjectInfo = {

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

@@ -780,6 +780,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

@@ -5,6 +5,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 234),
+  // НЕ путать с expires_at (актуальность расчёта). null = неоплачено (текущий
+  // трафик целиком).
+  retain_until?: string | null; // ISO datetime
   // ── Метаданные ──
   target_address: string | null;
   target_lat: number | null;