diff --git a/tradein-mvp/backend/app/api/v1/trade_in.py b/tradein-mvp/backend/app/api/v1/trade_in.py
index 42b75cd3..8e814002 100644
--- a/tradein-mvp/backend/app/api/v1/trade_in.py
+++ b/tradein-mvp/backend/app/api/v1/trade_in.py
@@ -44,6 +44,53 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def _assert_estimate_access(created_by: str | None, x_authenticated_user: str | None) -> None:
+    """IDOR guard (#690): только владелец оценки или admin могут её читать.
+
+    Зеркало скоупинга /history (#656). 401 — нет заголовка X-Authenticated-User
+    (Caddy basic_auth обязателен); 403 — юзер отсутствует в roles.yaml; 404 —
+    pilot читает чужую оценку (скрываем существование, не подтверждаем чужой id).
+    """
+    if not x_authenticated_user:
+        raise HTTPException(
+            status_code=401,
+            detail="no authenticated user (Caddy basic_auth required)",
+        )
+
+    from app.core.auth import get_role
+
+    try:
+        role = get_role(x_authenticated_user)
+    except KeyError:
+        logger.warning(
+            "user %r authenticated via Caddy but missing from roles.yaml",
+            x_authenticated_user,
+        )
+        raise HTTPException(status_code=403, detail="user not in roles config") from None
+
+    if role == "admin":
+        return
+    if created_by is None or created_by != x_authenticated_user:
+        raise HTTPException(status_code=404, detail="estimate not found")
+
+
+def _assert_estimate_access_by_id(
+    db: Session, estimate_id: UUID, x_authenticated_user: str | None
+) -> None:
+    """IDOR guard для derived-роутов, не читающих саму оценку (#690).
+
+    Тянет created_by оценки и применяет _assert_estimate_access. 404 если оценки
+    нет (как и owner-mismatch — существование не подтверждаем).
+    """
+    row = db.execute(
+        text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
+        {"id": str(estimate_id)},
+    ).fetchone()
+    if row is None:
+        raise HTTPException(status_code=404, detail="estimate not found")
+    _assert_estimate_access(row.created_by, x_authenticated_user)
+
+
 @router.post("/estimate", response_model=AggregatedEstimate)
 async def estimate(
     payload: TradeInEstimateInput,
@@ -100,6 +147,7 @@ def get_quota(
 def get_estimate(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> AggregatedEstimate:
     """Получить сохранённую оценку по UUID (для генерации PDF).
 
@@ -118,7 +166,7 @@ def get_estimate(
                    dadata_qc_geo, dadata_metro,
                    expected_sold_price, expected_sold_range_low,
                    expected_sold_range_high, expected_sold_per_m2,
-                   asking_to_sold_ratio, ratio_basis
+                   asking_to_sold_ratio, ratio_basis, created_by
             FROM trade_in_estimates
             WHERE id = CAST(:id AS uuid)
               AND expires_at > NOW()
@@ -130,6 +178,8 @@ def get_estimate(
     if row is None:
         raise HTTPException(status_code=404, detail="estimate not found or expired")
 
+    _assert_estimate_access(row.created_by, x_authenticated_user)
+
     from app.services.estimator import _qc_geo_to_precision
 
     analogs = [AnalogLot(**a) for a in (row.analogs or [])]
@@ -190,6 +240,7 @@ def estimate_pdf(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
     brand: str | None = None,
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> Response:
     """Скачать 4-страничный PDF-отчёт для оценки trade-in.
 
@@ -210,7 +261,7 @@ def estimate_pdf(
                    dadata_qc_geo, dadata_metro,
                    expected_sold_price, expected_sold_range_low,
                    expected_sold_range_high, expected_sold_per_m2,
-                   asking_to_sold_ratio, ratio_basis
+                   asking_to_sold_ratio, ratio_basis, created_by
             FROM trade_in_estimates
             WHERE id = CAST(:id AS uuid)
             """
@@ -221,6 +272,8 @@ def estimate_pdf(
     if row is None:
         raise HTTPException(status_code=404, detail="estimate not found")
 
+    _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)")
 
@@ -275,12 +328,15 @@ def estimate_pdf(
     }
 
     from app.services.brand import get_brand as _resolve_brand
+
     brand_obj = _resolve_brand(brand, db)
     pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj)
     filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf"
     logger.info(
         "PDF generated estimate_id=%s brand=%s size=%d",
-        estimate_id, brand_obj.slug, len(pdf_bytes),
+        estimate_id,
+        brand_obj.slug,
+        len(pdf_bytes),
     )
     return Response(
         content=pdf_bytes,
@@ -290,7 +346,7 @@ def estimate_pdf(
 
 
 # ── Фото квартиры (#394) ─────────────────────────────────────────────────────
-_MAX_PHOTO_BYTES = 10 * 1024 * 1024            # 10 МБ на фото
+_MAX_PHOTO_BYTES = 10 * 1024 * 1024  # 10 МБ на фото
 _MAX_PHOTOS_PER_ESTIMATE = 12
 _ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/heic"}
 
@@ -300,14 +356,16 @@ async def upload_photo(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
     file: Annotated[UploadFile, File()],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> PhotoMeta:
     """Загрузить фото квартиры к оценке (#394). Хранение в estimate_photos (bytea)."""
     estimate = db.execute(
-        text("SELECT 1 FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
+        text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
         {"id": str(estimate_id)},
     ).fetchone()
     if estimate is None:
         raise HTTPException(status_code=404, detail="estimate not found")
+    _assert_estimate_access(estimate.created_by, x_authenticated_user)
 
     ctype = (file.content_type or "").lower()
     if ctype not in _ALLOWED_IMAGE_TYPES:
@@ -337,27 +395,35 @@ async def upload_photo(
     except ImageSanitizationError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
 
-    row = db.execute(
-        text(
-            """
+    row = (
+        db.execute(
+            text(
+                """
             INSERT INTO estimate_photos
                 (estimate_id, filename, content_type, content, size_bytes)
             VALUES (CAST(:eid AS uuid), :fn, :ct, :content, :sz)
             RETURNING id, filename, content_type, size_bytes, uploaded_at
             """
-        ),
-        {
-            "eid": str(estimate_id),
-            "fn": file.filename,
-            "ct": sanitized_ctype,
-            "content": sanitized_bytes,
-            "sz": len(sanitized_bytes),
-        },
-    ).mappings().fetchone()
+            ),
+            {
+                "eid": str(estimate_id),
+                "fn": file.filename,
+                "ct": sanitized_ctype,
+                "content": sanitized_bytes,
+                "sz": len(sanitized_bytes),
+            },
+        )
+        .mappings()
+        .fetchone()
+    )
     db.commit()
     logger.info(
         "photo uploaded: estimate=%s photo=%s orig_size=%d sanitized_size=%d ctype=%s",
-        estimate_id, row["id"], len(content), len(sanitized_bytes), sanitized_ctype,
+        estimate_id,
+        row["id"],
+        len(content),
+        len(sanitized_bytes),
+        sanitized_ctype,
     )
     return PhotoMeta(**row)
 
@@ -366,19 +432,25 @@ async def upload_photo(
 def list_photos(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> list[PhotoMeta]:
     """Список фото оценки — метаданные, без содержимого (#394)."""
-    rows = db.execute(
-        text(
-            """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
+    rows = (
+        db.execute(
+            text(
+                """
             SELECT id, filename, content_type, size_bytes, uploaded_at
             FROM estimate_photos
             WHERE estimate_id = CAST(:id AS uuid)
             ORDER BY uploaded_at
             """
-        ),
-        {"id": str(estimate_id)},
-    ).mappings().all()
+            ),
+            {"id": str(estimate_id)},
+        )
+        .mappings()
+        .all()
+    )
     return [PhotoMeta(**r) for r in rows]
 
 
@@ -387,8 +459,10 @@ def get_photo(
     estimate_id: UUID,
     photo_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> Response:
     """Отдать содержимое фото — image bytes (#394)."""
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
     row = db.execute(
         text(
             """
@@ -452,9 +526,10 @@ def estimate_history(
         where = "WHERE created_by = :owner"
         params["owner"] = x_authenticated_user
 
-    rows = db.execute(
-        text(
-            f"""
+    rows = (
+        db.execute(
+            text(
+                f"""
             SELECT id, address, rooms, area_m2, median_price,
                    confidence, n_analogs, created_at
             FROM trade_in_estimates
@@ -462,18 +537,22 @@ def estimate_history(
             ORDER BY created_at DESC
             LIMIT :limit
             """
-        ),
-        params,
-    ).mappings().all()
+            ),
+            params,
+        )
+        .mappings()
+        .all()
+    )
     return [dict(r) for r in rows]
 
 
 @router.get("/cache-stats")
 def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
     """Состояние данных и кэшей (#399) — для страницы «Кэш»."""
-    row = db.execute(
-        text(
-            """
+    row = (
+        db.execute(
+            text(
+                """
             SELECT
               (SELECT count(*) FROM geocode_cache) AS geocode_cache,
               (SELECT count(*) FROM geocode_cache WHERE expires_at > NOW())
@@ -485,8 +564,11 @@ def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]:
               (SELECT count(*) FROM house_metadata) AS house_metadata,
               (SELECT count(*) FROM trade_in_estimates) AS estimates_total
             """
+            )
         )
-    ).mappings().fetchone()
+        .mappings()
+        .fetchone()
+    )
     return dict(row) if row else {}
 
 
@@ -507,6 +589,7 @@ _HOUSE_SELECT_COLS = """
 def get_estimate_houses(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> list[HouseInfoForEstimate]:
     """House(s) информация для estimate.
 
@@ -517,6 +600,7 @@ def get_estimate_houses(
     Возвращаем прямой матч + nearby (dedup by id), up to ~6 домов.
     Пустой список если нет matches.
     """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
     target = db.execute(
         text(
             """
@@ -546,7 +630,9 @@ def get_estimate_houses(
                     """
                 ),
                 {"addr": target.address},
-            ).mappings().all()
+            )
+            .mappings()
+            .all()
         )
 
     # Path 2: geo-nearby (any source, 500м radius)
@@ -573,7 +659,9 @@ def get_estimate_houses(
                     """
                 ),
                 {"lat": target.lat, "lon": target.lon},
-            ).mappings().all()
+            )
+            .mappings()
+            .all()
         )
 
     # Merge: direct первым, затем nearby (dedup by house_id)
@@ -590,12 +678,14 @@ def get_estimate_houses(
 def get_estimate_placement_history(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> list[PlacementHistoryEntry]:
     """Историческая продажная активность по дому(ам) target estimate.
 
     Возвращает rows из house_placement_history для всех houses связанных с
     target адресом. Сортировано по last_price_date DESC.
     """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
     target = db.execute(
         text(
             """
@@ -645,9 +735,10 @@ def get_estimate_placement_history(
     if not house_ids:
         return []
 
-    history = db.execute(
-        text(
-            """
+    history = (
+        db.execute(
+            text(
+                """
             SELECT id, source, house_id, ext_item_id, title, rooms, area_m2,
                    floor, total_floors, start_price, start_price_date,
                    last_price, last_price_date, removed_date, exposure_days,
@@ -657,9 +748,12 @@ def get_estimate_placement_history(
             ORDER BY COALESCE(last_price_date, start_price_date) DESC NULLS LAST
             LIMIT 50
             """
-        ),
-        {"house_ids": house_ids},
-    ).mappings().all()
+            ),
+            {"house_ids": house_ids},
+        )
+        .mappings()
+        .all()
+    )
 
     return [PlacementHistoryEntry(**dict(r)) for r in history]
 
@@ -668,12 +762,14 @@ def get_estimate_placement_history(
 def get_estimate_house_analytics(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> HouseAnalyticsResponse:
     """House-level analytics from house_placement_history backfill.
 
     Resolves target house(s) — если в самом доме <8 hist rows — расширяем поиск до 300м.
     Возвращает: price-history by year (median ₽/м²), recent sold (12mo), KPI.
     """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
     target = db.execute(
         text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
         {"id": str(estimate_id)},
@@ -740,9 +836,10 @@ def get_estimate_house_analytics(
         )
 
     # 3. Price history by year × source (median ₽/м²)
-    price_history_rows = db.execute(
-        text(
-            """
+    price_history_rows = (
+        db.execute(
+            text(
+                """
             SELECT
                 EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year,
                 source,
@@ -758,14 +855,18 @@ def get_estimate_house_analytics(
             GROUP BY year, source
             ORDER BY year ASC, source ASC
             """
-        ),
-        {"ids": house_ids},
-    ).mappings().all()
+            ),
+            {"ids": house_ids},
+        )
+        .mappings()
+        .all()
+    )
 
     # 4. Recent sold (12 months, with removed_date)
-    recent_sold_rows = db.execute(
-        text(
-            """
+    recent_sold_rows = (
+        db.execute(
+            text(
+                """
             SELECT id, source, rooms, area_m2, floor, start_price, last_price,
                    removed_date, exposure_days,
                    CASE WHEN start_price > 0 AND last_price IS NOT NULL
@@ -778,14 +879,18 @@ def get_estimate_house_analytics(
             ORDER BY removed_date DESC
             LIMIT 20
             """
-        ),
-        {"ids": house_ids},
-    ).mappings().all()
+            ),
+            {"ids": house_ids},
+        )
+        .mappings()
+        .all()
+    )
 
     # 5. KPI aggregate
-    kpi_row = db.execute(
-        text(
-            """
+    kpi_row = (
+        db.execute(
+            text(
+                """
             SELECT
                 COUNT(*) AS total_lots,
                 COUNT(*) FILTER (WHERE removed_date IS NOT NULL) AS sold_count,
@@ -811,9 +916,12 @@ def get_estimate_house_analytics(
             FROM house_placement_history
             WHERE house_id = ANY(:ids)
             """
-        ),
-        {"ids": house_ids},
-    ).mappings().first()
+            ),
+            {"ids": house_ids},
+        )
+        .mappings()
+        .first()
+    )
 
     return HouseAnalyticsResponse(
         house_ids=house_ids,
@@ -845,6 +953,7 @@ def get_estimate_house_analytics(
 def get_estimate_cian_price_changes(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> list[CianPriceChangeStats]:
     """История изменений цены для Cian-аналогов из estimate.
 
@@ -857,9 +966,11 @@ def get_estimate_cian_price_changes(
     Возвращает только аналоги с хотя бы одним изменением цены.
     Пустой список если нет cian-аналогов или нет истории.
     """
-    rows = db.execute(
-        text(
-            """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
+    rows = (
+        db.execute(
+            text(
+                """
             WITH cian_analogs AS (
                 SELECT DISTINCT
                     substring(a->>'source_url' from '/sale/flat/(\\d+)/') AS cian_id
@@ -916,9 +1027,12 @@ def get_estimate_cian_price_changes(
             WHERE ca.n_changes > 0
             ORDER BY ca.last_change_time DESC
             """
-        ),
-        {"eid": str(estimate_id)},
-    ).mappings().all()
+            ),
+            {"eid": str(estimate_id)},
+        )
+        .mappings()
+        .all()
+    )
 
     return [CianPriceChangeStats(**dict(r)) for r in rows]
 
@@ -930,6 +1044,7 @@ def get_estimate_cian_price_changes(
 def get_estimate_sell_time_sensitivity(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> SellTimeSensitivityResponse:
     """Срок продажи в зависимости от цены к медиане дома/района.
 
@@ -937,6 +1052,7 @@ def get_estimate_sell_time_sensitivity(
     Filter last_price > start_price * 0.7 — отбрасываем подозрительно
     заниженные лоты (выбросы, ошибки парсинга).
     """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
     # 1. Resolve house_ids (same logic as house-analytics)
     target = db.execute(
         text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
@@ -1014,9 +1130,10 @@ def get_estimate_sell_time_sensitivity(
     ).scalar()
 
     # 3. Per-year median (для расчёта premium per lot); используем CTE для bucket-расчёта
-    bucket_rows = db.execute(
-        text(
-            """
+    bucket_rows = (
+        db.execute(
+            text(
+                """
             WITH year_medians AS (
                 SELECT
                     EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year,
@@ -1070,9 +1187,12 @@ def get_estimate_sell_time_sensitivity(
             WHERE bucket IS NOT NULL
             GROUP BY bucket
             """
-        ),
-        {"ids": house_ids},
-    ).mappings().all()
+            ),
+            {"ids": house_ids},
+        )
+        .mappings()
+        .all()
+    )
 
     # 4. Build buckets — гарантируем все 4 даже если данных нет в bucket
     bucket_map = {r["bucket"]: dict(r) for r in bucket_rows}
@@ -1108,6 +1228,7 @@ def get_estimate_sell_time_sensitivity(
 def get_estimate_imv_benchmark(
     estimate_id: UUID,
     db: Annotated[Session, Depends(get_db)],
+    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> IMVBenchmarkResponse:
     """Avito IMV benchmark для estimate (для UI badge «наша 6.4М · Avito 6.29М»).
 
@@ -1115,6 +1236,7 @@ def get_estimate_imv_benchmark(
       1. avito_imv_evaluations WHERE estimate_id = :id (если linked в estimator)
       2. Если не linked — fallback: most recent IMV для same address (TTL 24h)
     """
+    _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
     # Сначала пытаемся найти directly linked
     row = db.execute(
         text(
@@ -1238,9 +1360,10 @@ def get_street_deals(
     area_min = area_m2 * (1.0 - area_tolerance)
     area_max = area_m2 * (1.0 + area_tolerance)
 
-    rows = db.execute(
-        text(
-            """
+    rows = (
+        db.execute(
+            text(
+                """
             SELECT address, area_m2, rooms, floor, total_floors,
                    price_rub, price_per_m2, deal_date, source
             FROM deals
@@ -1252,20 +1375,26 @@ def get_street_deals(
               AND price_rub > 0
             ORDER BY deal_date DESC
             """
-        ),
-        {
-            "street_pattern": "%" + street_name + "%",
-            "rooms": rooms,
-            "area_min": area_min,
-            "area_max": area_max,
-            "period_months": period_months,
-        },
-    ).mappings().all()
+            ),
+            {
+                "street_pattern": "%" + street_name + "%",
+                "rooms": rooms,
+                "area_min": area_min,
+                "area_max": area_max,
+                "period_months": period_months,
+            },
+        )
+        .mappings()
+        .all()
+    )
 
     if not rows:
         logger.info(
             "street-deals: no rows found street=%r rooms=%d area=%.1f±%.0f%%",
-            street_name, rooms, area_m2, area_tolerance * 100,
+            street_name,
+            rooms,
+            area_m2,
+            area_tolerance * 100,
         )
         return StreetDealsResponse(
             street=street_name,
@@ -1294,7 +1423,11 @@ def get_street_deals(
 
     logger.info(
         "street-deals: street=%r rooms=%d area=%.1f count=%d median_ppm2=%.0f",
-        street_name, rooms, area_m2, count, median_ppm2,
+        street_name,
+        rooms,
+        area_m2,
+        count,
+        median_ppm2,
     )
 
     return StreetDealsResponse(
@@ -1357,9 +1490,10 @@ def get_sales_vs_listings(
         logger.warning("sales-vs-listings: cannot extract street from %r", address)
         return _empty()
 
-    rows = db.execute(
-        text(
-            """
+    rows = (
+        db.execute(
+            text(
+                """
             SELECT
                 deal_id, deal_date, deal_price_rub, deal_price_per_m2,
                 deal_area_m2, deal_rooms, deal_floor, deal_address,
@@ -1375,21 +1509,27 @@ def get_sales_vs_listings(
                 CAST(:period_months AS integer)
             )
             """
-        ),
-        {
-            "street_pattern": "%" + street_name + "%",
-            "area_m2": area_m2,
-            "rooms": rooms,
-            "window_days": window_days,
-            "area_tolerance": area_tolerance,
-            "period_months": period_months,
-        },
-    ).mappings().all()
+            ),
+            {
+                "street_pattern": "%" + street_name + "%",
+                "area_m2": area_m2,
+                "rooms": rooms,
+                "window_days": window_days,
+                "area_tolerance": area_tolerance,
+                "period_months": period_months,
+            },
+        )
+        .mappings()
+        .all()
+    )
 
     if not rows:
         logger.info(
             "sales-vs-listings: no deals street=%r rooms=%d area=%.1f period_months=%d",
-            street_name, rooms, area_m2, period_months,
+            street_name,
+            rooms,
+            area_m2,
+            period_months,
         )
         return _empty(reason_street=street_name)
 
@@ -1411,35 +1551,30 @@ def get_sales_vs_listings(
                 int(r["listing_price_rub"]) if r["listing_price_rub"] is not None else None
             ),
             listing_price_per_m2=(
-                int(r["listing_price_per_m2"])
-                if r["listing_price_per_m2"] is not None
-                else None
+                int(r["listing_price_per_m2"]) if r["listing_price_per_m2"] is not None else None
             ),
             listing_area_m2=(
                 float(r["listing_area_m2"]) if r["listing_area_m2"] is not None else None
             ),
             days_listing_to_deal=r["days_listing_to_deal"],
-            discount_pct=(
-                float(r["discount_pct"]) if r["discount_pct"] is not None else None
-            ),
+            discount_pct=(float(r["discount_pct"]) if r["discount_pct"] is not None else None),
         )
         for r in rows
     ]
 
     total_deals = len(pairs)
     deals_with_listings = sum(1 for p in pairs if p.listing_id is not None)
-    linkage_rate_pct = (
-        round(deals_with_listings / total_deals * 100, 1) if total_deals else 0.0
-    )
+    linkage_rate_pct = round(deals_with_listings / total_deals * 100, 1) if total_deals else 0.0
 
     discounts = sorted(p.discount_pct for p in pairs if p.discount_pct is not None)
-    median_discount = (
-        round(_percentile(discounts, 0.5), 2) if discounts else None
-    )
+    median_discount = round(_percentile(discounts, 0.5), 2) if discounts else None
 
     logger.info(
         "sales-vs-listings: street=%r deals=%d with_listings=%d linkage=%.1f%% median_disc=%s",
-        street_name, total_deals, deals_with_listings, linkage_rate_pct,
+        street_name,
+        total_deals,
+        deals_with_listings,
+        linkage_rate_pct,
         f"{median_discount:+.2f}%" if median_discount is not None else "n/a",
     )
 
diff --git a/tradein-mvp/backend/tests/test_estimate_idor.py b/tradein-mvp/backend/tests/test_estimate_idor.py
new file mode 100644
index 00000000..26f316ac
--- /dev/null
+++ b/tradein-mvp/backend/tests/test_estimate_idor.py
@@ -0,0 +1,476 @@
+"""Tests for GET /estimate/{id} and /estimate/{id}/pdf ownership scoping (#690 IDOR).
+
+Closes cross-pilot data-leak: any authenticated pilot could read/download another
+pilot's estimate (leaking client_name/client_phone). The endpoints now scope to the
+owner (trade_in_estimates.created_by) OR admin, mirroring /history (#656).
+
+Реальная БД не нужна: DB + get_role мокируются. Для 200-кейсов подменяем
+_qc_geo_to_precision и generate_trade_in_pdf, чтобы не тащить тяжёлый estimator/PDF.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+# psycopg v3 driver required; stub DATABASE_URL before any app import
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+# WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import.
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+sys.modules.setdefault("weasyprint.CSS", _wp_mock)
+sys.modules.setdefault("weasyprint.HTML", _wp_mock)
+
+import pytest  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+_ESTIMATE_ID = "11111111-1111-1111-1111-111111111111"
+
+
+@pytest.fixture(autouse=True)
+def _restore_get_role():
+    """Restore app.core.auth.get_role after each test (mirror test_history_scope)."""
+    from app.core import auth as auth_mod
+
+    original = auth_mod.get_role
+    yield
+    auth_mod.get_role = original
+
+
+@pytest.fixture()
+def trade_in_app() -> FastAPI:
+    """Minimal FastAPI app mounting only the trade-in router."""
+    from app.api.v1 import trade_in as trade_in_module
+
+    application = FastAPI()
+    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
+    return application
+
+
+def _make_estimate_row(created_by: str | None) -> SimpleNamespace:
+    """A trade_in_estimates row with the full column set the endpoints read."""
+    from datetime import UTC, datetime, timedelta
+
+    return SimpleNamespace(
+        id=_ESTIMATE_ID,
+        median_price=5_000_000,
+        range_low=4_500_000,
+        range_high=5_500_000,
+        median_price_per_m2=100_000,
+        confidence="medium",
+        confidence_explanation="ok",
+        n_analogs=7,
+        analogs=[],
+        actual_deals=[],
+        sources_used=["avito"],
+        data_freshness_minutes=10,
+        expires_at=datetime.now(tz=UTC) + timedelta(hours=12),
+        address="ул. Тестовая, 1",
+        lat=56.8,
+        lon=60.6,
+        area_m2=50.0,
+        rooms=2,
+        floor=3,
+        total_floors=9,
+        year_built=2010,
+        house_type="монолит",
+        repair_state="хороший",
+        has_balcony=True,
+        canonical_address="ул. Тестовая, 1",
+        house_cadnum=None,
+        house_fias_id=None,
+        dadata_qc_geo=0,
+        dadata_metro=[],
+        expected_sold_price=4_600_000,
+        expected_sold_range_low=4_200_000,
+        expected_sold_range_high=5_000_000,
+        expected_sold_per_m2=92_000,
+        asking_to_sold_ratio=0.85,
+        ratio_basis="per_rooms",
+        created_by=created_by,
+    )
+
+
+def _make_db_mock(row: SimpleNamespace | None) -> MagicMock:
+    """DB session mock returning *row* from .execute(...).fetchone()."""
+    db = MagicMock()
+    execute_result = MagicMock()
+    execute_result.fetchone.return_value = row
+    db.execute.return_value = execute_result
+    return db
+
+
+def _client_with(app: FastAPI, db_mock: MagicMock, role: str | None) -> TestClient:
+    """Override get_db with *db_mock*; patch get_role to return *role* (or raise KeyError)."""
+    from app.core.db import get_db
+
+    def _override_db():
+        yield db_mock
+
+    app.dependency_overrides[get_db] = _override_db
+
+    auth_mod = sys.modules["app.core.auth"]
+    if role is None:
+
+        def _raise_keyerror(_u: str):
+            raise KeyError(_u)
+
+        auth_mod.get_role = _raise_keyerror  # type: ignore[assignment]
+    else:
+        auth_mod.get_role = lambda _u: role  # type: ignore[assignment]
+    return TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def _stub_precision_and_pdf():
+    """Stub _qc_geo_to_precision + generate_trade_in_pdf so 200-paths don't pull heavy deps."""
+    from app.api.v1 import trade_in as trade_in_module
+
+    estimator_stub = SimpleNamespace(_qc_geo_to_precision=lambda _qc: "house")
+    real_estimator = sys.modules.get("app.services.estimator")
+    sys.modules["app.services.estimator"] = estimator_stub  # type: ignore[assignment]
+
+    original_pdf = trade_in_module.generate_trade_in_pdf
+    trade_in_module.generate_trade_in_pdf = lambda *a, **k: b"%PDF-fake"  # type: ignore[assignment]
+
+    # brand resolution also imports lazily — stub the module it imports from.
+    brand_stub = SimpleNamespace(get_brand=lambda _b, _db: SimpleNamespace(slug="generic"))
+    real_brand = sys.modules.get("app.services.brand")
+    sys.modules["app.services.brand"] = brand_stub  # type: ignore[assignment]
+
+    yield
+
+    trade_in_module.generate_trade_in_pdf = original_pdf
+    if real_estimator is not None:
+        sys.modules["app.services.estimator"] = real_estimator
+    else:
+        sys.modules.pop("app.services.estimator", None)
+    if real_brand is not None:
+        sys.modules["app.services.brand"] = real_brand
+    else:
+        sys.modules.pop("app.services.brand", None)
+
+
+# ── GET /estimate/{id} ───────────────────────────────────────────────────────
+
+
+def test_get_estimate_owner_can_read(trade_in_app: FastAPI) -> None:
+    """Owner pilot reads own estimate → 200."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
+    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()["estimate_id"] == _ESTIMATE_ID
+
+
+def test_get_estimate_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """Non-owner pilot must NOT read someone else's estimate → 404 (hide existence)."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="victim"))
+    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": "attacker"},
+    )
+    assert resp.status_code == 404
+
+
+def test_get_estimate_admin_can_read_any(trade_in_app: FastAPI) -> None:
+    """Admin reads any estimate regardless of owner → 200."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="someone_else"))
+    client = _client_with(trade_in_app, db_mock, role="admin")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
+        headers={"X-Authenticated-User": "admin"},
+    )
+    assert resp.status_code == 200
+
+
+def test_get_estimate_requires_authenticated_user(trade_in_app: FastAPI) -> None:
+    """No X-Authenticated-User header → 401."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}")
+    assert resp.status_code == 401
+
+
+def test_get_estimate_unknown_user_gets_403(trade_in_app: FastAPI) -> None:
+    """Authenticated via Caddy but missing from roles.yaml → 403."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
+    client = _client_with(trade_in_app, db_mock, role=None)
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}",
+        headers={"X-Authenticated-User": "ghost"},
+    )
+    assert resp.status_code == 403
+
+
+# ── GET /estimate/{id}/pdf ───────────────────────────────────────────────────
+
+
+def test_pdf_owner_can_download(trade_in_app: FastAPI) -> None:
+    """Owner pilot downloads own PDF → 200."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
+    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_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """Non-owner pilot must NOT download someone else's PDF → 404."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="victim"))
+    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": "attacker"},
+    )
+    assert resp.status_code == 404
+
+
+def test_pdf_admin_can_download_any(trade_in_app: FastAPI) -> None:
+    """Admin downloads any PDF regardless of owner → 200."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="someone_else"))
+    client = _client_with(trade_in_app, db_mock, role="admin")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
+        headers={"X-Authenticated-User": "admin"},
+    )
+    assert resp.status_code == 200
+
+
+def test_pdf_requires_authenticated_user(trade_in_app: FastAPI) -> None:
+    """No X-Authenticated-User header → 401."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf")
+    assert resp.status_code == 401
+
+
+def test_pdf_unknown_user_gets_403(trade_in_app: FastAPI) -> None:
+    """/pdf: authenticated via Caddy but missing from roles.yaml → 403 (reviewer gap)."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by="kopylov"))
+    client = _client_with(trade_in_app, db_mock, role=None)
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/pdf",
+        headers={"X-Authenticated-User": "ghost"},
+    )
+    assert resp.status_code == 403
+
+
+# ── Reviewer-named gap: legacy NULL created_by ───────────────────────────────
+
+
+def test_get_estimate_legacy_null_owner_non_admin_gets_404(trade_in_app: FastAPI) -> None:
+    """Legacy estimate with NULL created_by: non-admin pilot must NOT read it → 404."""
+    db_mock = _make_db_mock(_make_estimate_row(created_by=None))
+    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 == 404
+
+
+# ── Photo endpoints (#690 — raw apartment-interior bytes are enumerable PII) ──
+
+
+def _make_photo_row() -> SimpleNamespace:
+    """A minimal estimate_photos row for get_photo (content + content_type)."""
+    return SimpleNamespace(content=b"\x89PNG-fake", content_type="image/png")
+
+
+def _make_db_mock_seq(*rows: SimpleNamespace | None) -> MagicMock:
+    """DB session mock whose successive .execute(...).fetchone() yield *rows* in order.
+
+    Derived/photo routes run the guard SELECT (created_by) first, then their own
+    queries; configure side_effect so the guard row comes first.
+    """
+    db = MagicMock()
+
+    def _execute(*_a, **_k):
+        result = MagicMock()
+        result.fetchone.return_value = next(_iter)
+        return result
+
+    _iter = iter(rows)
+    db.execute.side_effect = _execute
+    return db
+
+
+def test_get_photo_owner_can_read(trade_in_app: FastAPI) -> None:
+    """Owner pilot reads photo bytes → 200."""
+    db_mock = _make_db_mock_seq(
+        SimpleNamespace(created_by="kopylov"),  # guard query
+        _make_photo_row(),  # route query
+    )
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
+        headers={"X-Authenticated-User": "kopylov"},
+    )
+    assert resp.status_code == 200
+    assert resp.content == b"\x89PNG-fake"
+
+
+def test_get_photo_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """Non-owner pilot must NOT read someone else's photo → 404 (guard raises first)."""
+    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
+        headers={"X-Authenticated-User": "attacker"},
+    )
+    assert resp.status_code == 404
+
+
+def test_get_photo_admin_can_read_any(trade_in_app: FastAPI) -> None:
+    """Admin reads any photo → 200."""
+    db_mock = _make_db_mock_seq(
+        SimpleNamespace(created_by="someone_else"),
+        _make_photo_row(),
+    )
+    client = _client_with(trade_in_app, db_mock, role="admin")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
+        headers={"X-Authenticated-User": "admin"},
+    )
+    assert resp.status_code == 200
+
+
+def test_get_photo_requires_authenticated_user(trade_in_app: FastAPI) -> None:
+    """get_photo: no X-Authenticated-User header → 401."""
+    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="kopylov"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos/{_ESTIMATE_ID}",
+    )
+    assert resp.status_code == 401
+
+
+def test_list_photos_owner_can_read(trade_in_app: FastAPI) -> None:
+    """Owner pilot lists photos → 200 (empty list ok)."""
+    db = MagicMock()
+    guard_result = MagicMock()
+    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
+    list_result = MagicMock()
+    list_result.mappings.return_value.all.return_value = []
+    db.execute.side_effect = [guard_result, list_result]
+    client = _client_with(trade_in_app, db, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
+        headers={"X-Authenticated-User": "kopylov"},
+    )
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+def test_list_photos_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """Non-owner pilot must NOT list someone else's photos → 404."""
+    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
+        headers={"X-Authenticated-User": "attacker"},
+    )
+    assert resp.status_code == 404
+
+
+def test_upload_photo_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """Write-side IDOR: non-owner pilot must NOT upload to someone else's estimate → 404.
+
+    Guard runs on the existence SELECT (created_by) before any write — so a single
+    fetchone returning created_by='admin' + a non-owner header is enough.
+    """
+    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="admin"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.post(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
+        headers={"X-Authenticated-User": "attacker"},
+        files={"file": ("x.png", b"\x89PNG-fake", "image/png")},
+    )
+    assert resp.status_code == 404
+
+
+def test_upload_photo_owner_can_upload(trade_in_app: FastAPI, monkeypatch) -> None:
+    """Owner pilot uploads → 200. Stub sanitize_image + the INSERT mapping row."""
+    from app.api.v1 import trade_in as trade_in_module
+
+    monkeypatch.setattr(
+        trade_in_module, "sanitize_image", lambda _c: (b"\x89PNG-clean", "image/png")
+    )
+
+    from datetime import UTC, datetime
+
+    db = MagicMock()
+    guard_result = MagicMock()
+    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
+    count_result = MagicMock()
+    count_result.scalar_one.return_value = 0
+    insert_result = MagicMock()
+    insert_result.mappings.return_value.fetchone.return_value = {
+        "id": _ESTIMATE_ID,
+        "filename": "x.png",
+        "content_type": "image/png",
+        "size_bytes": 9,
+        "uploaded_at": datetime.now(tz=UTC),
+    }
+    db.execute.side_effect = [guard_result, count_result, insert_result]
+    client = _client_with(trade_in_app, db, role="pilot")
+    resp = client.post(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/photos",
+        headers={"X-Authenticated-User": "kopylov"},
+        files={"file": ("x.png", b"\x89PNG-fake", "image/png")},
+    )
+    assert resp.status_code == 200
+
+
+# ── Derived analytics routes (representative: /houses, /imv-benchmark) ────────
+
+
+def test_get_estimate_houses_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """Derived route: non-owner pilot blocked by guard → 404 (before any house query)."""
+    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/houses",
+        headers={"X-Authenticated-User": "attacker"},
+    )
+    assert resp.status_code == 404
+
+
+def test_get_estimate_houses_owner_can_read(trade_in_app: FastAPI) -> None:
+    """Derived route: owner pilot passes guard → 200. Stub target with no address/geo."""
+    db = MagicMock()
+    guard_result = MagicMock()
+    guard_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
+    target_result = MagicMock()
+    target_result.fetchone.return_value = SimpleNamespace(lat=None, lon=None, address=None)
+    db.execute.side_effect = [guard_result, target_result]
+    client = _client_with(trade_in_app, db, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/houses",
+        headers={"X-Authenticated-User": "kopylov"},
+    )
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+def test_get_estimate_imv_benchmark_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
+    """imv-benchmark: non-owner pilot blocked by guard → 404."""
+    db_mock = _make_db_mock_seq(SimpleNamespace(created_by="victim"))
+    client = _client_with(trade_in_app, db_mock, role="pilot")
+    resp = client.get(
+        f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}/imv-benchmark",
+        headers={"X-Authenticated-User": "attacker"},
+    )
+    assert resp.status_code == 404