fix(tradein/privacy): не удалять B2B-строки в purge + находить телефон в другом формате при erasure (#2547)

Deep-review HIGH: purge_expired_trade_in_data удалял trade_in_estimates по
expires_at без разбора B2B/B2C -- эта колонка TTL ссылки/PDF, а не срок
хранения строки, и её единообразно проставляет каждой оценке estimator.py.
Прод-аудит: 1040/1057 строк просрочены, 911 из них у пилотов (admin,
kopylov, brusnika, praktika, pilottest, admintest, user1). DELETE теперь
ограничен created_by IS NULL -- ровно анонимная B2C-популяция (129 строк).
Докстринг миграции 231 переписан: явные цифры аудита, необратимость,
чек-лист (свежий SELECT count + один supervised прогон) перед enable.

Deep-review MEDIUM: erase_person_data сравнивал phone точным =, а lead.py
сохраняет номер как прислали (без нормализации, намеренно) -- разное
форматирование одного и того же номера не находилось, 0 строк удалялось,
но ответ всё равно был 200 "данные удалены". Сравнение переведено на
regexp_replace(x, '\D', '', 'g') с обеих сторон.

Оба фикса проверены живьём (throwaway Postgres 16 в docker, вне обычного
mock-only CI-лейна): без гварда пилотская строка удалялась вместе с
анонимной; без нормализации разноформатный телефон не находился. С
фиксами -- находит/не находит ровно как задумано. Добавлены self-skipping
live-DB тесты (паттерн test_house_dedup_merge.py::_live_session) плюс
статические SQL-guard тесты.

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

@@ -16,6 +16,11 @@ WHO CAN BE IDENTIFIED, HONESTLY:
         * `estimate_ids` -- if they still have the link/PDF from their estimate
           (the UUID in the URL/QR-code IS their proof of "this is mine").
         * `phone` -- if they left a contact-request lead with that phone.
+          Matched by NORMALIZED DIGITS ONLY (regexp_replace strips everything
+          but 0-9 on both sides), not an exact string: lead.py stores
+          `payload.phone` exactly as typed (no E.164 normalization, by
+          design), so the caller's "+7 999 123-45-67" must still find a row
+          saved as "89991234567" or any other formatting of the same digits.
         * `tg_chat_id` -- if they messaged @MERAsupport_bot directly (their own
           Telegram chat id -- not guessable/spoofable by a third party the way
           a name or IP would be).
@@ -119,13 +124,32 @@ def erase_person_data(
 
     # 2. Лиды -- пока estimate_id ещё живой FK (см. п.1), плюс отдельно по
     #    телефону (лид мог быть оставлен без attach к оценке вовсе).
+    #
+    #    ⚠️ Телефон сравнивается по НОРМАЛИЗОВАННЫМ цифрам, не литералом
+    #    (deep-review 2026-08-06, MEDIUM). app/api/v1/lead.py сохраняет
+    #    payload.phone КАК ПРИСЛАЛИ (намеренно -- полная E.164-нормализация
+    #    вне scope MVP, см. lead.py::_PHONE_PATTERN), т.е. одна и та же
+    #    строка может лежать в БД как "+7 999 123-45-67" ИЛИ "89991234567"
+    #    ИЛИ любой другой форматировкой той же маски. Точное `phone = :phone`
+    #    находит строку только если запрашивающий пришлёт БУКВАЛЬНО ТОТ ЖЕ
+    #    формат, каким когда-то ввёл номер -- почти никогда так. Раньше это
+    #    молча удаляло 0 строк и всё равно возвращало 200 "данные удалены":
+    #    для 152-ФЗ ложное подтверждение удаления хуже честной ошибки.
+    #    `regexp_replace(x, '\\D', '', 'g')` с ОБЕИХ сторон сравнения снимает
+    #    форматирование (пробелы/скобки/дефисы/+) и сравнивает голые цифры.
+    #    Параметр -- CAST(:phone AS text), НЕ конкатенация (psycopg v3 / SQL
+    #    injection convention, .claude/rules/backend.md).
     ids_param = [str(i) for i in all_estimate_ids]
     result = db.execute(
         text(
             """
             DELETE FROM trade_in_leads
             WHERE estimate_id = ANY(CAST(:ids AS uuid[]))
-               OR phone = :phone
+               OR (
+                   CAST(:phone AS text) IS NOT NULL
+                   AND regexp_replace(phone, '\\D', '', 'g')
+                       = regexp_replace(CAST(:phone AS text), '\\D', '', 'g')
+               )
             """
         ),
         {"ids": ids_param, "phone": phone},

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

@@ -23,6 +23,27 @@ WHAT:
       it has its OWN retention clock (trade_in_leads.expires_at) and its own PII (phone),
       purged independently below.
 
+⚠️ trade_in_estimates DELETE is scoped to `created_by IS NULL` (deep-review finding,
+   2026-08-06, HIGH): `expires_at` on this table is set UNCONDITIONALLY for every
+   estimate, B2B pilot or anonymous (`now + settings.trade_in_estimate_retention_hours`,
+   see app/services/estimator.py) -- it is a TTL on the ESTIMATE LINK/PDF staying
+   resolvable (GET /estimate/{id}: 404 past expiry, PDF export: 410 past expiry), NOT a
+   declared retention deadline for the ROW. B2B pilots' consent is closed by contract
+   (see migration 229's `consent` column asymmetry: NULL for `created_by IS NOT NULL`,
+   the exact same B2B-vs-B2C split used here) and their estimates are the live basis for
+   /trade-in/history, /team/employees/{id}/history and the cache-stats dashboards (see
+   app/api/v1/trade_in.py, app/api/v1/team.py) -- deleting them past a 24h *link* TTL
+   would be silent, irreversible data loss of pilots' own operational data, not a
+   privacy-driven cleanup. Audited against prod on 2026-08-06: of 1057 rows, 1040 had
+   already crossed `expires_at`, and 911 of THOSE belonged to named pilots
+   (`created_by` set: admin, kopylov, brusnika, praktika, pilottest, admintest, user1).
+   Without the `created_by IS NULL` guard, one unattended run at
+   `batch_size=500, max_batches=20` (the defaults) would have deleted essentially the
+   whole table. `created_by IS NULL` is the honest B2C population -- 129 rows in that
+   same audit. trade_in_leads has no `created_by` column (never had a B2B/B2C split --
+   its own `expires_at` really is a 180-day retention deadline, not a link TTL, see
+   migration 231) so its DELETE below is intentionally NOT scoped the same way.
+
 BATCHING (не единый DELETE по всей таблице):
     Each table is drained in batches of `batch_size` rows (default
     settings.trade_in_purge_batch_size), each batch its OWN statement + its OWN commit
@@ -59,6 +80,7 @@ _DELETE_EXPIRED_ESTIMATES_SQL = text(
     WHERE id IN (
         SELECT id FROM trade_in_estimates
         WHERE expires_at < NOW()
+          AND created_by IS NULL
         ORDER BY expires_at
         LIMIT CAST(:batch_size AS int)
     )

tradein-mvp/backend/data/sql/231_trade_in_privacy_retention.sql

@@ -34,9 +34,35 @@
 --   175_scrape_schedules_seed_domclick_detail_backfill.sql): это ПЕРВАЯ
 --   автоматическая задача физического DELETE персональных данных в trade-in —
 --   заслуживает supervised первого прогона (смотри логи/counters вручную)
---   перед тем, как доверить её расписанию. Включение — отдельный ручной шаг
---   (UPDATE scrape_schedules SET enabled=true WHERE source=
---   'purge_expired_trade_in_data').
+--   перед тем, как доверить её расписанию.
+--
+--   ⚠️ ПЕРЕД ВКЛЮЧЕНИЕМ (deep-review 2026-08-06, HIGH — читай целиком, не
+--   только команду в конце): задача удаляет строки trade_in_estimates
+--   физически и НЕОБРАТИМО. DELETE ограничен `created_by IS NULL` — только
+--   анонимные B2C-оценки (см. докстринг app/tasks/purge_expired_trade_in_
+--   data.py, раздел про асимметрию expires_at = TTL ссылки/PDF, а не срок
+--   хранения строки) — строки B2B-пилотов (`created_by` задан) задача НЕ
+--   трогает НИКОГДА, что бы ни стояло в expires_at. Аудит на проде на момент
+--   этой правки: из 1057 строк trade_in_estimates 1040 уже просрочены
+--   (expires_at < NOW()), но ТОЛЬКО 129 из них имеют created_by IS NULL
+--   (реальная B2C-популяция под удаление) — остальные 911 принадлежат
+--   пилотам (admin/kopylov/brusnika/praktika/pilottest/admintest/user1) и
+--   гвардом защищены от удаления. Эти цифры УСТАРЕЮТ — не включай задачу по
+--   их памяти. Перед `UPDATE scrape_schedules SET enabled=true WHERE
+--   source='purge_expired_trade_in_data'`:
+--     1. Свежий `SELECT count(*) FROM trade_in_estimates WHERE expires_at
+--        < NOW() AND created_by IS NULL` — если число на порядок больше
+--        129 (или created_by-гвард почему-то отсутствует в текущем коде
+--        задачи) — СТОП, разберись, прежде чем включать.
+--     2. Прогони задачу вручную ОДИН раз (app/tasks/
+--        purge_expired_trade_in_data.py, синхронная функция) и сверь
+--        counters с п.1 (задача логирует batch/running_total через
+--        logger.info в _drain_expired) — supervised первый прогон, не
+--        включение вслепую.
+--     3. Только после совпадения счётчиков — enable в scrape_schedules;
+--        расписание подхватит на следующем тике планировщика (крутится в
+--        контейнере tradein-scraper, не tradein-backend).
+--   Включение — отдельный ручной шаг, эта миграция его НЕ делает.
 --
 -- IDEMPOTENCY / SAFETY:
 --   - ADD COLUMN IF NOT EXISTS + UPDATE ... WHERE expires_at IS NULL (no-op на

tradein-mvp/backend/tests/test_data_erasure.py

@@ -15,6 +15,7 @@ Covers app/services/data_erasure.py:
 from __future__ import annotations
 
 import os
+import re
 from types import SimpleNamespace
 from typing import Any
 from unittest.mock import MagicMock
@@ -137,6 +138,26 @@ def test_erase_by_phone_only_touches_only_leads() -> None:
     assert params["ids"] == []
 
 
+def test_phone_delete_normalizes_digits_on_both_sides() -> None:
+    """Regression guard for the deep-review MEDIUM finding (2026-08-06):
+    lead.py stores phone exactly as typed (no E.164 normalization, by
+    design), so a differently-formatted-but-same-number erasure request
+    ('+7 999 123-45-67' vs a stored '89991234567') must still match. The old
+    exact `phone = :phone` comparison silently deleted 0 rows and still
+    returned HTTP 200 'erased' -- worse than an honest error under 152-ФЗ.
+    Both sides of the comparison must go through regexp_replace, and the
+    literal-equality path must be gone."""
+    db = MagicMock()
+    db.execute.side_effect = [_Result(rowcount=1)]
+
+    data_erasure.erase_person_data(db, phone="+7 999 123-45-67")
+
+    sql = _sql_of(db.execute.call_args_list[0])
+    assert sql.count("regexp_replace") == 2
+    assert "phone = :phone" not in sql
+    assert not re.search(r":\w+::", sql)  # psycopg v3 CAST trap
+
+
 def test_erase_by_tg_chat_id_only_touches_only_tg_support() -> None:
     """Anonymous person with NO username, NO estimate link, NO lead phone -- but
     they DID message @MERAsupport_bot -- can still be identified by their own
@@ -158,3 +179,72 @@ def test_erase_by_tg_chat_id_only_touches_only_tg_support() -> None:
     calls = db.execute.call_args_list
     assert "DELETE FROM tg_support_users" in _sql_of(calls[-1])
     assert calls[-1].args[1]["chat_id"] == 123456789
+
+
+# ── Optional real-Postgres behavioural test (self-skips without a DB) ──────────
+# Same pattern as tests/test_house_dedup_merge.py::_live_session -- CI runs the
+# mock-only lane (DATABASE_URL is a placeholder), so this self-skips there; it
+# only executes with a real reachable Postgres (e.g. local dev DB).
+
+
+def _live_session() -> Any | None:
+    """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None."""
+    try:
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+        if not dsn or "localhost:5432/test" in dsn:
+            return None
+        engine = create_engine(dsn, future=True)
+        conn = engine.connect()
+        from sqlalchemy import text as _t
+
+        conn.execute(_t("SELECT 1"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_erase_by_phone_finds_differently_formatted_number() -> None:
+    """End-to-end on a real DB: a lead stored with phone EXACTLY as typed
+    ('89991234567', no separators) must still be found and deleted when the
+    erasure requester supplies the SAME digits in a DIFFERENT format
+    ('8 (999) 123-45-67') -- proves the regexp_replace normalization fix
+    actually matches, not just that the SQL text contains the function name."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    lead_id: Any = None
+    try:
+        row = db.execute(
+            _t(
+                "INSERT INTO trade_in_leads (phone, consent, expires_at) "
+                "VALUES (:phone, TRUE, NOW() + interval '180 days') "
+                "RETURNING id"
+            ),
+            {"phone": "89991234567"},
+        ).fetchone()
+        assert row is not None
+        lead_id = row[0]
+        db.commit()
+
+        out = data_erasure.erase_person_data(db, phone="8 (999) 123-45-67")
+
+        assert out["trade_in_leads_deleted"] == 1
+        remaining = db.execute(
+            _t("SELECT count(*) FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
+            {"id": str(lead_id)},
+        ).scalar()
+        assert remaining == 0
+    finally:
+        if lead_id is not None:
+            db.execute(
+                _t("DELETE FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
+                {"id": str(lead_id)},
+            )
+            db.commit()
+        db.close()

tradein-mvp/backend/tests/test_purge_expired_trade_in_data.py

@@ -18,6 +18,7 @@ import os
 import re
 from pathlib import Path
 from typing import Any
+from uuid import uuid4
 
 import pytest
 
@@ -149,6 +150,35 @@ def test_leads_sql_is_delete_not_update() -> None:
     assert not re.search(r":\w+::", sql)
 
 
+def test_estimates_sql_scopes_delete_to_anonymous_created_by_null() -> None:
+    """Regression guard for the deep-review HIGH finding (2026-08-06):
+    trade_in_estimates.expires_at is set UNCONDITIONALLY on every estimate
+    (B2B pilot or anonymous) as a TTL on the estimate LINK/PDF staying
+    resolvable, NOT a declared row-retention deadline -- see the task's
+    module docstring. Without this guard, prod audit showed 1040/1057 rows
+    past expires_at, 911 of them belonging to named pilots (admin, kopylov,
+    brusnika, praktika, pilottest, admintest, user1); one unattended run at
+    the seeded defaults (batch_size=500, max_batches=20) would have deleted
+    essentially the whole table, including pilots' own operational history
+    (/trade-in/history, /team/employees/{id}/history, cache-stats all read
+    trade_in_estimates without an expires_at filter). The DELETE must stay
+    scoped to created_by IS NULL -- the honest B2C population (129 rows in
+    that same audit)."""
+    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
+    assert "created_by IS NULL" in sql
+    assert not re.search(r":\w+::", sql)
+
+
+def test_leads_sql_has_no_created_by_guard() -> None:
+    """trade_in_leads has NO created_by column at all (never had a B2B/B2C
+    split) -- its expires_at IS a genuine 180-day retention deadline (see
+    migration 231), not a link/PDF-access TTL like trade_in_estimates. This
+    documents the asymmetry explicitly so a future 'fix' doesn't bolt a
+    created_by filter onto a table that doesn't have the column."""
+    sql = task_mod._DELETE_EXPIRED_LEADS_SQL.text
+    assert "created_by" not in sql
+
+
 def test_sql_does_not_delete_whole_table_unbounded() -> None:
     """Neither statement is a bare `DELETE FROM table` -- both scope via a
     subselect + LIMIT batch."""
@@ -230,3 +260,82 @@ def test_migration_231_seeds_purge_schedule_disabled_by_default() -> None:
 def test_migration_231_no_psycopg_trap() -> None:
     sql = _MIGRATION_231.read_text("utf-8")
     assert not re.search(r":\w+::", sql)
+
+
+# ── Optional real-Postgres behavioural test (self-skips without a DB) ──────────
+# Same pattern as tests/test_house_dedup_merge.py::_live_session -- CI runs the
+# mock-only lane (DATABASE_URL is a placeholder), so this self-skips there; it
+# only executes with a real reachable Postgres (e.g. local dev DB).
+
+
+def _live_session() -> Any | None:
+    """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None."""
+    try:
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+        if not dsn or "localhost:5432/test" in dsn:
+            return None
+        engine = create_engine(dsn, future=True)
+        conn = engine.connect()
+        from sqlalchemy import text as _t
+
+        conn.execute(_t("SELECT 1"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_purge_deletes_only_anonymous_expired_estimates() -> None:
+    """End-to-end on a real DB: one expired B2B row (created_by set) and one
+    expired anonymous row (created_by IS NULL) both exist -- a real run of
+    purge_expired_trade_in_data must delete ONLY the anonymous one. This is
+    the exact scenario the deep-review HIGH finding (2026-08-06) flagged:
+    without the created_by IS NULL guard, the pilot row would also be gone."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    anon_id = uuid4()
+    pilot_id = uuid4()
+    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) VALUES "
+                "(CAST(:anon_id AS uuid), 'purge-test аноним', 40, 1, 2, 5, "
+                " 5000000, 4500000, 5500000, 125000, 'low', "
+                " NOW() - interval '1 hour', NULL), "
+                "(CAST(:pilot_id AS uuid), 'purge-test пилот', 40, 1, 2, 5, "
+                " 5000000, 4500000, 5500000, 125000, 'low', "
+                " NOW() - interval '1 hour', 'pytest_purge_guard')"
+            ),
+            {"anon_id": str(anon_id), "pilot_id": str(pilot_id)},
+        )
+        db.commit()
+
+        task_mod.purge_expired_trade_in_data(db, run_id=999999999, batch_size=100, max_batches=5)
+
+        remaining_ids = {
+            str(r)
+            for r in db.execute(
+                _t("SELECT id FROM trade_in_estimates " "WHERE id = ANY(CAST(:ids AS uuid[]))"),
+                {"ids": [str(anon_id), str(pilot_id)]},
+            )
+            .scalars()
+            .all()
+        }
+        assert str(anon_id) not in remaining_ids, "anonymous expired row must be purged"
+        assert str(pilot_id) in remaining_ids, "B2B pilot row must survive despite expires_at"
+    finally:
+        db.execute(
+            _t("DELETE FROM trade_in_estimates WHERE id = ANY(CAST(:ids AS uuid[]))"),
+            {"ids": [str(anon_id), str(pilot_id)]},
+        )
+        db.commit()
+        db.close()