fix(tradein/avito): браузерный путь ходил на каждую карточку холодным (#3251)

Пройденный QRATOR proof-of-work выбрасывался после каждой карточки: `reuse_context`
во всём репозитории передавал ЕДИНСТВЕННЫЙ вызов — domclick_detail_backfill.py:346.
Значит каждый /fetch Авито шёл через sidecar'овский browser.new_page(), то есть
новый изолированный context с пустой банкой кук. В логе прод-сайдкара это видно
прямо: «PoW-челлендж снят за ~1000мс» печатается на КАЖДОЙ успешной карточке —
челлендж решается заново каждый раз, а не один раз на прогон.

Эффект той же правки у Домклика измерен и записан в domclick_detail_backfill.py:331
— 26 последовательных фетчей сайдкара дали 100% блоков, те же карточки в тёплом
контексте 5/5 за ~2с.

Второй холод — сам заход: providers/avito/detail.py звал fetch(full_url) голым,
без origin и без Referer, тогда как Домклик (#3247) идёт fetch(card_url,
origin=SERP, referer=origin). Существующий для Авито прогрев (warm_up_session)
живёт только в curl-пути и с 22.08 в проде мёртв — AVITO_DETAIL_BACKFILL_USE_CURL
выставлен в false.

Что сделано:
- avito_detail_backfill: reuse_context=True + request_context_reset() РОВНО один
  раз за прогон и только на AvitoBlockedError. Не на AvitoSidecarUnavailableError
  (подтип AvitoRateLimitedError — отказ нашего тракта, не бан площадки) и не на
  AvitoListingGoneError. Зеркалит #3212: сброс на каждый блок сам себя
  поддерживает — пропуск живёт в context'е, сброс его выбрасывает, повторная
  проверка с того же IP снова блокируется, одна осечка даёт каскад.
- fetch_detail: необязательные origin/browser_referer (имя referer уже занято под
  Referer curl-пути, это разные фетчеры и разные поля). Дефолт None → payload и
  поведение city_sweep/pipeline/admin не меняются.
- _serp_origin_for: городская SERP из URL карточки, хост берётся из самого url.

Попутно — дефект якорной вкладки сайдкара, найденный при переносе. _ensure_anchor_page
отдавала True на ЛЮБУЮ живую вкладку, не сверяя её с запрошенным origin. А origin у
обоих caller'ов выводится ИЗ URL карточки и меняется вместе с городом (Авито —
сегмент пути, Домклик — поддомен). После первой же карточки другого города Referer
называл выдачу, которую этот контекст никогда не открывал: ни куки её, ни тайминга,
площадка видит заявленный переход без единого следа. Ровно то, что #3258 запретил
делать фолбэкам якорного поиска. Добавлен _anchor_origins: origin сменился — вкладка
переоткрывается. Чинит и Домклик тоже.

Заход через поиск Яндекса (BROWSER_ANCHOR_VIA_SEARCH) для Авито НЕ включается —
расширять этот список без отдельного замера запрещает комментарий у самой константы.
Хранилище авторизованных сессий Авито (#3179/#3180) этой правкой не заменяется.

База для сравнения снята ДО выката и записана в #3251: доля блоков от попыток
46.9% / 74.2% / 96.4% / 69.8% / 52.9% / 62.3% по суткам 25-30.08. Сравнивать после
деплоя по доле блоков и обогащению за сутки, а НЕ по статусу прогона: ratio-критерий
обрывает КАЖДЫЙ прогон, и статус banned про площадку ничего не говорит.

Тесты: backend 5150 passed / 37 skipped, сайдкар 211 passed (было 206 + 5 новых на
переезд якоря), ruff чист.

Refs #3251, #3180, #3118, #3212, #3247, #3258

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

@@ -73,6 +73,7 @@ from scraper_kit.orchestration.pipeline import CITY_LOCATIONS, ban_kind_of_excep
 from scraper_kit.providers._base import DEFAULT_IMPERSONATE, DOCUMENT_HEADERS, http_proxies
 from scraper_kit.providers.avito.detail import (
     _AVITO_WARM_SEARCH_URL,
+    _serp_origin_for,
     build_warmed_session,
     fetch_detail,
     research_in_session,
@@ -293,6 +294,13 @@ async def run_avito_detail_backfill(
             # Замер 27.08: прогон 5098 — mode=browser, proxy_lease_id=None,
             # 5 блоков подряд из 5 попыток, enriched=0. При этом пул здоров
             # (4 узла, все ok), а тот же URL через прокси отдаёт 200 и 3.3 МБ.
+            # reuse_context=True (#3180/#3251): без него sidecar's browser.new_page()
+            # создаёт НОВЫЙ изолированный context на КАЖДЫЙ /fetch — пройденный
+            # QRATOR-PoW предыдущей карточки выбрасывается, и следующий запрос снова
+            # холодный. У DomClick (#3118) это измеренно давало 100% блоков (26/26)
+            # на изолированных контекстах против 5/5 успехов в тёплом контексте.
+            # Сброс сожжённого context'а — ниже, через bf.request_context_reset()
+            # (#3212: один раз за прогон, не на каждый блок — см. except-ветку).
             _cfg = RealScraperConfig()
             browser_fetcher = BrowserFetcher(
                 source="avito",
@@ -302,6 +310,7 @@ async def run_avito_detail_backfill(
                 # Без environment отказ «пул пуст» на этом пути мёртв — фетчер
                 # молча ушёл бы на env-прокси сайдкара (#2616 шаг 1).
                 environment=_cfg.environment,
+                reuse_context=True,
             )
             await browser_fetcher.__aenter__()
             own_browser = True
@@ -439,6 +448,13 @@ async def run_avito_detail_backfill(
         abort_reason: str | None = None
         do_sleep = False
         items_since_warm = 0
+        # #3251: сброс переиспользуемого browser-context'а (reuse_context=True выше)
+        # разрешён РОВНО один раз за прогон — зеркалит domclick_detail_backfill (#3212).
+        # Сброс на КАЖДЫЙ блок сам себя поддерживает: пройденный QRATOR-PoW живёт в
+        # context'е, сброс его выбрасывает, повторная проверка с того же IP сразу
+        # после принятой снова блокируется — одна осечка превращается в необратимый
+        # каскад блоков (см. except-ветку ниже).
+        context_reset_used = False
         # Перепись причин (блоки + отказы) — переживает пересоздание контейнера,
         # в отличие от логов; см. _failure_signature.
         failure_census: Counter[str] = Counter()
@@ -544,6 +560,14 @@ async def run_avito_detail_backfill(
                 # #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не
                 # должен блокировать loop навсегда (иначе budget-guard/heartbeat молчат
                 # и run reaped как zombie). wait_for отменяет fetch → TimeoutError.
+                # #3251: same-site SERP-якорь. Смысл он имеет ТОЛЬКО в browser-
+                # режиме: сайдкар держит выдачу открытой якорной вкладкой и шлёт
+                # её как Referer целевой навигации. На curl/own-session пути
+                # сессия прогрета warm-batch'ем (см. referer= ниже, это ДРУГОЕ
+                # поле и другой фетчер) — там origin остаётся None и fetch_detail
+                # ведёт себя ровно как раньше. None также если source_url не
+                # распарсился на город+категорию (см. docstring _serp_origin_for).
+                serp_origin = _serp_origin_for(source_url) if browser_mode else None
                 enrichment = await asyncio.wait_for(
                     fetch_detail(
                         item_url,
@@ -558,6 +582,8 @@ async def run_avito_detail_backfill(
                         # напрямую). Не влияет на use_curl=True (прод-дефолт) —
                         # там reconnect_on_block=False уже гасит backconnect.
                         config=RealScraperConfig(),
+                        origin=serp_origin,
+                        browser_referer=serp_origin,
                     ),
                     timeout=fetch_timeout_s,
                 )
@@ -627,6 +653,22 @@ async def run_avito_detail_backfill(
                 failure_census[_failure_signature(e)] += 1
                 block_ban_kinds[ban_kind_of_exception(e)] += 1
                 do_sleep = False
+                # #3251: сброс тёплого browser-context'а — ровно один раз за прогон,
+                # и только на настоящий бан ПЛОЩАДКОЙ (AvitoBlockedError и подтипы:
+                # AvitoContentBlockedError, AvitoWarmupCookiesMissingError). НЕ на
+                # AvitoSidecarUnavailableError (подтип AvitoRateLimitedError) — это
+                # отказ НАШЕГО тракта, площадка тут ни при чём, context цел и сбрасывать
+                # его незачем. См. domclick_detail_backfill.py (#3212) — тот же приём:
+                # сброс на каждый блок сам себя поддерживает (пройденный QRATOR-PoW
+                # живёт в context'е, сброс его выбрасывает → следующая проверка с того
+                # же IP снова блокируется).
+                if (
+                    browser_fetcher is not None
+                    and isinstance(e, AvitoBlockedError)
+                    and not context_reset_used
+                ):
+                    context_reset_used = True
+                    browser_fetcher.request_context_reset()
                 logger.warning(
                     "avito_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s",
                     run_id,

tradein-mvp/backend/tests/scrapers/test_3251_avito_fetch_detail_origin.py

@@ -0,0 +1,83 @@
+"""fetch_detail(#3251): same-site SERP origin/referer в browser-режиме Avito.
+
+Переносит на Avito рецепт, измеренно поднявший добор карточек у DomClick (#3247/
+#3118): переход на карточку идёт из SERP-выдачи (origin), а не голым goto(url).
+Контракт формулируется от того, чего требует площадка (переход должен выглядеть
+как реальная навигация с выдачи), а не от того, что уже делает код — см. #3262,
+где обратный подход закрепил баг на месяцы.
+
+_serp_origin_for — чистый URL-хелпер, тестируется отдельно ниже без сети/моков.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.providers.avito.detail import _serp_origin_for, fetch_detail
+
+_MIN_HTML = '<html><body><div data-marker="item-view/item-id">№ 1234567890</div></body></html>'
+
+
+# ── fetch_detail: origin/browser_referer прокидываются в browser_fetcher.fetch ──
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_passes_origin_and_referer_when_given() -> None:
+    """Площадка должна увидеть переход С ВЫДАЧИ: origin/referer уходят в fetch()."""
+    bf = MagicMock()
+    bf.fetch = AsyncMock(return_value=_MIN_HTML)
+    card_url = "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_1234567890"
+    origin = "https://www.avito.ru/ekaterinburg/kvartiry/prodam"
+
+    await fetch_detail(card_url, browser_fetcher=bf, origin=origin, browser_referer=origin)
+
+    bf.fetch.assert_called_once_with(card_url, origin=origin, referer=origin)
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_omits_origin_and_referer_when_not_given() -> None:
+    """Дефолт (origin/browser_referer не переданы) — payload как раньше, без ключей.
+
+    Это путь ВСЕХ прочих browser-вызовов (city_sweep, pipeline, admin) — они не
+    знают о #3251 и не должны получить изменённый payload.
+    """
+    bf = MagicMock()
+    bf.fetch = AsyncMock(return_value=_MIN_HTML)
+    card_url = "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_1234567890"
+
+    await fetch_detail(card_url, browser_fetcher=bf)
+
+    bf.fetch.assert_called_once_with(card_url, origin=None, referer=None)
+
+
+# ── _serp_origin_for: чистый URL-хелпер ─────────────────────────────────────────
+
+
+def test_serp_origin_for_normal_card_url() -> None:
+    url = "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_45_m_kv_1234567890"
+    assert _serp_origin_for(url) == "https://www.avito.ru/ekaterinburg/kvartiry/prodam"
+
+
+def test_serp_origin_for_other_city_and_host() -> None:
+    """Хост берётся из card_url, не хардкодится 'www.avito.ru'."""
+    url = "https://m.avito.ru/tyumen/kvartiry/1-k._kvartira_30_m_kv_987654321"
+    assert _serp_origin_for(url) == "https://m.avito.ru/tyumen/kvartiry/prodam"
+
+
+@pytest.mark.parametrize(
+    "garbage",
+    [
+        "not a url at all",
+        "https://www.avito.ru/",
+        "https://www.avito.ru/ekaterinburg",
+        "",
+        "avito.ru/ekaterinburg/kvartiry/foo",  # без scheme -> netloc пуст
+    ],
+)
+def test_serp_origin_for_garbage_returns_none(garbage: str) -> None:
+    assert _serp_origin_for(garbage) is None

tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py

@@ -735,6 +735,10 @@ async def test_backfill_fetch_timeout_skips_and_continues() -> None:
         referer: object = None,
         reconnect_on_block: bool = True,
         config: object = None,
+        # Стаб терпит рост сигнатуры fetch_detail: тест про hard-timeout, а не про
+        # набор аргументов. Без этого любой новый kwarg (напр. origin/browser_referer
+        # из #3251) валит его TypeError'ом, хотя к таймауту отношения не имеет.
+        **_extra: object,
     ):
         call_urls.append(url)
         if len(call_urls) == 1:

tradein-mvp/backend/tests/test_3251_avito_warm_context.py

@@ -0,0 +1,394 @@
+"""Тёплый переиспользуемый browser-context для Avito detail-бэкфилла (#3180/#3251).
+
+Переносит на Avito рецепт, измеренно поднявший добор карточек у DomClick (#3118):
+sidecar's browser.new_page() создаёт НОВЫЙ изолированный context на КАЖДЫЙ /fetch —
+пройденный QRATOR-подобный анти-бот проход площадки выбрасывается вместе с ним, и
+каждая следующая карточка снова холодная. reuse_context=True держит ОДИН sidecar-
+context живым на весь прогон; сброс сожжённого context'а оркестратор просит через
+bf.request_context_reset() — как и DomClick, ровно один раз за прогон (#3212), а не
+на каждый обнаруженный блок (иначе сброс сам себя поддерживает — см. domclick
+docstring). Reset относится ТОЛЬКО к реальному бану площадкой (AvitoBlockedError и
+подтипы), не к отказу нашего тракта (AvitoSidecarUnavailableError) и не к мёртвому
+листингу (AvitoListingGoneError) — площадка тут ни при чём в обоих случаях, context
+цел.
+
+Контракт тестов сформулирован от требования площадки/архитектуры (что ДОЛЖНО
+происходить при блоке/не-блоке), а не от текущей реализации — см. #3262, где
+обратный подход закрепил баг тестами на месяцы.
+
+Зеркалит tests/test_3118_domclick_warm_context.py (module-level patch-target
+строки, helpers).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+import pytest  # noqa: E402
+from scraper_kit.avito_exceptions import (  # noqa: E402
+    AvitoBlockedError,
+    AvitoContentBlockedError,
+    AvitoListingGoneError,
+    AvitoSidecarUnavailableError,
+)
+
+from app.core import shutdown as _sd  # noqa: E402
+from app.tasks.avito_detail_backfill import run_avito_detail_backfill  # noqa: E402
+
+_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
+_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
+_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
+_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
+_SETTINGS = "app.tasks.avito_detail_backfill.settings"
+_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
+_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
+_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"
+
+
+@pytest.fixture(autouse=True)
+def _reset_shutdown() -> None:
+    """shutdown -- module-global Event: чистим вокруг каждого теста (изоляция #1182)."""
+    _sd.reset_shutdown()
+    yield
+    _sd.reset_shutdown()
+
+
+def _fake_settings(**overrides: object) -> MagicMock:
+    defaults: dict[str, object] = {
+        "scraper_fetch_mode": "browser",
+        "avito_detail_backfill_use_curl": False,
+        "detail_backfill_block_ratio_window": 20,
+        "detail_backfill_block_ratio_threshold": 0.7,
+        "browser_http_endpoint": "http://browser:9000",
+    }
+    defaults.update(overrides)
+    return MagicMock(**defaults)
+
+
+def _make_snapshot(n: int, *, absolute: bool = True) -> list[dict]:
+    if absolute:
+        return [
+            {
+                "id": i + 1,
+                "source_url": (
+                    f"https://www.avito.ru/ekaterinburg/kvartiry/"
+                    f"1-k._kvartira_{i + 1}"
+                ),
+            }
+            for i in range(n)
+        ]
+    return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)]
+
+
+def _mock_db(snapshot: list[dict]) -> MagicMock:
+    db = MagicMock()
+    sel = MagicMock()
+    sel.mappings.return_value.all.return_value = snapshot
+    db.execute.return_value = sel
+    return db
+
+
+def _mock_browser_fetcher_cls() -> tuple[MagicMock, AsyncMock]:
+    """Возвращает (класс-мок, instance-мок) -- instance с реальным request_context_reset."""
+    instance = AsyncMock()
+    instance.__aenter__ = AsyncMock(return_value=instance)
+    instance.__aexit__ = AsyncMock(return_value=False)
+    # request_context_reset -- синхронный метод (не awaited caller'ом).
+    instance.request_context_reset = MagicMock()
+    return MagicMock(return_value=instance), instance
+
+
+# ── BrowserFetcher(reuse_context=True) ──────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_backfill_constructs_browser_fetcher_with_reuse_context() -> None:
+    """BrowserFetcher поднимается с reuse_context=True (#3180) -- тёплый sidecar-context."""
+    snapshot = _make_snapshot(1)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(return_value=MagicMock())
+    fake_settings = _fake_settings()
+    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, return_value=True),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=200, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    mock_bf_cls.assert_called_once()
+    _, call_kwargs = mock_bf_cls.call_args
+    assert call_kwargs["source"] == "avito"
+    assert call_kwargs["reuse_context"] is True
+
+
+# ── reset: ровно один раз за прогон, только на реальный бан площадкой ──────────
+
+
+@pytest.mark.asyncio
+async def test_context_reset_requested_at_most_once_per_run() -> None:
+    """3 подряд AvitoBlockedError -> request_context_reset() запрошен ровно 1 раз.
+
+    Зеркалит domclick #3212: сброс на КАЖДЫЙ блок сам себя поддерживает (пройденный
+    анти-бот проход живёт в context'е, сброс его выбрасывает -> следующая проверка с
+    того же IP снова блокируется) -- одна осечка становится необратимым каскадом.
+    """
+    snapshot = _make_snapshot(10)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    blocked_exc = AvitoBlockedError("firewall/soft-block")
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = _fake_settings()
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=201,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
+        )
+
+    assert result.blocked == 3
+    assert instance.request_context_reset.call_count == 1, (
+        "ожидалась ровно одна попытка начать с чистого листа за весь прогон"
+    )
+
+
+@pytest.mark.asyncio
+async def test_content_blocked_subtype_also_requests_reset() -> None:
+    """AvitoContentBlockedError (подтип AvitoBlockedError) тоже считается блоком площадки."""
+    snapshot = _make_snapshot(3)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(side_effect=AvitoContentBlockedError("0 карточек, captcha"))
+    fake_settings = _fake_settings()
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=202,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
+        )
+
+    assert result.blocked == 3
+    instance.request_context_reset.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_sidecar_unavailable_does_not_request_context_reset() -> None:
+    """AvitoSidecarUnavailableError -- НАШ тракт, площадка ни при чём: reset НЕ звать.
+
+    Context цел, сбрасывать его на отказе сайдкара (503/timeout ДО площадки) незачем
+    и вредно -- анти-бот проход площадки внутри context'а тут ни при чём.
+    """
+    snapshot = _make_snapshot(3)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(side_effect=AvitoSidecarUnavailableError("browser unavailable"))
+    fake_settings = _fake_settings()
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db,
+            run_id=203,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
+        )
+
+    assert result.blocked == 3
+    instance.request_context_reset.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_listing_gone_does_not_request_context_reset() -> None:
+    """AvitoListingGoneError (мёртвый 404-листинг) -- не блок, reset НЕ звать."""
+    snapshot = _make_snapshot(3)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(side_effect=AvitoListingGoneError("404 gone"))
+    fake_settings = _fake_settings()
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db, run_id=204, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    assert result.gone == 3
+    assert result.blocked == 0
+    instance.request_context_reset.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_successful_fetch_does_not_request_context_reset() -> None:
+    """Успешный прогон без блоков -> request_context_reset() ни разу не вызывается."""
+    snapshot = _make_snapshot(3)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(return_value=MagicMock())
+    fake_settings = _fake_settings()
+    mock_bf_cls, instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, return_value=True),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_avito_detail_backfill(
+            db, run_id=205, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    assert result.blocked == 0
+    instance.request_context_reset.assert_not_called()
+
+
+# ── origin/browser_referer wiring в fetch_detail ────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_backfill_passes_serp_origin_to_fetch_detail_in_browser_mode() -> None:
+    """browser-режим: fetch_detail получает origin=browser_referer=SERP-URL карточки."""
+    snapshot = _make_snapshot(1, absolute=True)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(return_value=MagicMock())
+    fake_settings = _fake_settings()
+    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, return_value=True),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=206, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    mock_fetch.assert_called_once()
+    _, kwargs = mock_fetch.call_args
+    expected = "https://www.avito.ru/ekaterinburg/kvartiry/prodam"
+    assert kwargs.get("origin") == expected
+    assert kwargs.get("browser_referer") == expected
+
+
+@pytest.mark.asyncio
+async def test_backfill_omits_origin_when_url_unparseable() -> None:
+    """source_url без города/категории (relative path) -> origin/browser_referer=None.
+
+    Хелпер не смог вывести SERP-URL -> caller НЕ выдумывает его, просто не передаёт
+    (fetch_detail тогда идёт как до #3251).
+    """
+    snapshot = _make_snapshot(1, absolute=False)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(return_value=MagicMock())
+    fake_settings = _fake_settings()
+    mock_bf_cls, _instance = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, return_value=True),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=207, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    mock_fetch.assert_called_once()
+    _, kwargs = mock_fetch.call_args
+    assert kwargs.get("origin") is None
+    assert kwargs.get("browser_referer") is None
+
+
+@pytest.mark.asyncio
+async def test_curl_mode_does_not_pass_origin() -> None:
+    """use_curl=True (прод-дефолт warm-batch путь) -- origin/browser_referer НЕ передаются.
+
+    Тот путь уже прогрет warm-batch'ем (referer=), origin -- ТОЛЬКО browser-режим (см.
+    fetch_detail docstring): в curl-режиме передача origin стоила бы лишней полной
+    навигации без реального смысла (browser_fetcher там вообще не участвует).
+    """
+    snapshot = _make_snapshot(1, absolute=True)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(return_value=MagicMock())
+    fake_settings = _fake_settings(avito_detail_backfill_use_curl=True)
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION),
+        patch(_SCRAPER),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER) as mock_bf_cls,
+        patch(_FETCH, mock_fetch),
+        patch(
+            "app.tasks.avito_detail_backfill.build_warmed_session",
+            AsyncMock(return_value=AsyncMock()),
+        ),
+        patch(_SAVE, return_value=True),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=208, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    mock_bf_cls.assert_not_called()
+    mock_fetch.assert_called_once()
+    _, kwargs = mock_fetch.call_args
+    assert kwargs.get("origin") is None
+    assert kwargs.get("browser_referer") is None

tradein-mvp/browser/server.py

@@ -746,6 +746,16 @@ _contexts: dict[str, object] = {}
 # Живёт только при reuse_context=True: без переиспользуемого context'а держать
 # вкладку негде и незачем.
 _anchor_pages: dict[str, object] = {}
+# provider -> origin, НА КОТОРОМ якорная вкладка стоит СЕЙЧАС. Без этой памяти
+# _ensure_anchor_page отдавала любую живую вкладку как годную, не глядя на
+# запрошенный origin, — а origin у обоих caller'ов выводится ИЗ URL карточки и
+# меняется вместе с городом (Авито: /ekaterinburg/... vs /verhnyaya_pyshma/...;
+# Домклик: ekaterinburg.domclick.ru vs иной поддомен). Тогда Referer целевой
+# навигации называл выдачу, которую этот контекст НИКОГДА не открывал: ни куки
+# её, ни тайминга — площадка видит заявленный переход без единого следа. Ровно
+# то, что #3258 запретил делать фолбэкам якорного поиска. Origin сменился —
+# вкладку переоткрываем на новый.
+_anchor_origins: dict[str, str] = {}
 _page_counters: dict[str, int] = {}      # provider → страниц с момента launch'а
 _locks: dict[str, asyncio.Lock] = {}     # provider → Lock (весь доступ к инстансу)
 _retry_tasks: dict[str, asyncio.Task[None]] = {}  # provider → фоновая retry-задача
@@ -867,11 +877,28 @@ async def _ensure_anchor_page(provider: str, origin: str) -> bool:
     page = _anchor_pages.get(provider)
     if page is not None:
         try:
-            if not page.is_closed():  # type: ignore[attr-defined]
-                return True
+            alive = not page.is_closed()  # type: ignore[attr-defined]
         except Exception:
-            pass
+            alive = False
+        # Живая вкладка годится ТОЛЬКО если стоит на запрошенном origin: иначе
+        # Referer назовёт выдачу, которой в этом контексте не открывали (см.
+        # комментарий у _anchor_origins). Чужой origin — закрываем и поднимаем
+        # заново; это дороже на одну навигацию, но честно.
+        if alive and _anchor_origins.get(provider) == origin:
+            return True
+        if alive:
+            logger.info(
+                "tradein-browser[%s]: якорь стоял на %s, запрошен %s — переоткрываем",
+                provider,
+                _anchor_origins.get(provider),
+                origin,
+            )
+            try:
+                await page.close()  # type: ignore[attr-defined]
+            except Exception:
+                pass
         _anchor_pages.pop(provider, None)
+        _anchor_origins.pop(provider, None)
 
     try:
         page = await ctx.new_page()  # type: ignore[attr-defined]
@@ -907,6 +934,7 @@ async def _ensure_anchor_page(provider: str, origin: str) -> bool:
         return False
 
     _anchor_pages[provider] = page
+    _anchor_origins[provider] = origin
     logger.info("tradein-browser[%s]: якорная вкладка открыта на %s", provider, origin)
     return True
 
@@ -921,6 +949,7 @@ async def _close_reusable_context(provider: str) -> None:
     # Якорная вкладка принадлежит этому context'у и умирает вместе с ним — важно
     # снять ссылку ДО close(), иначе следующий запрос возьмёт закрытую страницу.
     _anchor_pages.pop(provider, None)
+    _anchor_origins.pop(provider, None)
     ctx = _contexts.pop(provider, None)
     if ctx is None:
         return

tradein-mvp/browser/test_server_anchor_origin_switch.py

@@ -0,0 +1,140 @@
+"""test_server_anchor_origin_switch.py — якорь обязан стоять на ЗАПРОШЕННОМ origin.
+
+Контракт формулируется от того, чего требует площадка, а не от того, что делал код:
+Referer целевой навигации называет страницу выдачи, поэтому эта выдача обязана быть
+в контексте реально открыта. До этой правки `_ensure_anchor_page` возвращала True на
+ЛЮБУЮ живую вкладку, не сверяя её origin с запрошенным. Оба нынешних caller'а выводят
+origin из URL карточки, и он меняется вместе с городом (Авито — сегмент пути, Домклик —
+поддомен), так что после первой же карточки другого города Referer называл выдачу,
+которой этот контекст никогда не открывал: ни куки её, ни тайминга. Ровно тот вид
+«заявленного, но не случившегося перехода», который запретил #3258.
+
+camoufox НЕ запускается — context/page подделаны.
+
+Запуск (из tradein-mvp/browser/)::
+
+    python -m pytest test_server_anchor_origin_switch.py -q
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
+_spec = importlib.util.spec_from_file_location("tradein_browser_server_anchor_switch", _SERVER_PATH)
+assert _spec is not None and _spec.loader is not None
+server = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(server)
+
+_PROVIDER = "avito"
+_EKB = "https://www.avito.ru/ekaterinburg/kvartiry/prodam"
+_PYSHMA = "https://www.avito.ru/verhnyaya_pyshma/kvartiry/prodam"
+
+
+@pytest.fixture(autouse=True)
+def _reset_state() -> None:
+    server._contexts = {}
+    server._anchor_pages = {}
+    server._anchor_origins = {}
+
+
+class _Page:
+    def __init__(self) -> None:
+        self.goto_urls: list[str] = []
+        self.closed = 0
+        self.url = ""
+
+    def is_closed(self) -> bool:
+        return self.closed > 0
+
+    async def route(self, pattern: str, handler: Any) -> None:
+        return None
+
+    async def goto(self, url: str, **kwargs: Any) -> None:
+        self.goto_urls.append(url)
+        self.url = url
+
+    async def wait_for_timeout(self, ms: int) -> None:
+        return None
+
+    async def close(self) -> None:
+        self.closed += 1
+
+
+class _Context:
+    """new_page() отдаёт КАЖДЫЙ раз новую вкладку — как настоящий BrowserContext."""
+
+    def __init__(self) -> None:
+        self.pages: list[_Page] = []
+
+    async def new_page(self) -> _Page:
+        page = _Page()
+        self.pages.append(page)
+        return page
+
+    async def close(self) -> None:
+        return None
+
+
+def test_same_origin_reuses_the_open_tab() -> None:
+    """Повторный запрос того же origin не платит за навигацию второй раз."""
+    ctx = _Context()
+    server._contexts[_PROVIDER] = ctx
+
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
+
+    assert len(ctx.pages) == 1, "вторая вкладка не нужна — origin тот же"
+    assert ctx.pages[0].goto_urls == [_EKB]
+    assert server._anchor_origins[_PROVIDER] == _EKB
+
+
+def test_changed_origin_reopens_the_tab_on_the_new_origin() -> None:
+    """Другой город — якорь ОБЯЗАН переехать, иначе Referer называет чужую выдачу."""
+    ctx = _Context()
+    server._contexts[_PROVIDER] = ctx
+
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _PYSHMA)) is True
+
+    assert len(ctx.pages) == 2, "на новый origin поднимается новая вкладка"
+    assert ctx.pages[0].closed == 1, "старая вкладка не должна течь"
+    assert ctx.pages[1].goto_urls == [_PYSHMA]
+    assert server._anchor_origins[_PROVIDER] == _PYSHMA
+    assert server._anchor_pages[_PROVIDER] is ctx.pages[1]
+
+
+def test_dead_tab_is_replaced_even_on_the_same_origin() -> None:
+    """Вкладку закрыли извне (recycle/краш) — поднимаем заново, не отдаём мёртвую."""
+    ctx = _Context()
+    server._contexts[_PROVIDER] = ctx
+
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
+    asyncio.run(ctx.pages[0].close())
+
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
+    assert len(ctx.pages) == 2
+    assert server._anchor_pages[_PROVIDER] is ctx.pages[1]
+
+
+def test_closing_context_forgets_the_anchor_origin() -> None:
+    """Сброс context'а обнуляет и память об origin — иначе новый context унаследует
+    чужое значение и вернёт «годен» на вкладку, которой уже нет."""
+    ctx = _Context()
+    server._contexts[_PROVIDER] = ctx
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is True
+
+    asyncio.run(server._close_reusable_context(_PROVIDER))
+
+    assert _PROVIDER not in server._anchor_origins
+    assert _PROVIDER not in server._anchor_pages
+
+
+def test_no_context_means_no_anchor() -> None:
+    """Без переиспользуемого context'а держать вкладку негде — прежнее поведение."""
+    assert asyncio.run(server._ensure_anchor_page(_PROVIDER, _EKB)) is False

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/detail.py

@@ -467,6 +467,30 @@ def _is_detail_not_found(html: str) -> bool:
     return "страница не найдена" in head or "ошибка 404" in head
 
 
+def _serp_origin_for(card_url: str) -> str | None:
+    """URL городской SERP-выдачи для карточки — same-site якорь origin/referer (#3251).
+
+    Из '<scheme>://<host>/<city>/<category>/<slug>...' достаёт первые ДВА сегмента
+    пути (город, категория) и добавляет фиксированный хвост '/prodam' — так Avito
+    формирует URL списка объявлений о продаже. Хост берётся из самого card_url (НЕ
+    хардкодится 'www.avito.ru' — сайдкар мог прийти с мобильного/иного поддомена).
+
+    Путь короче двух сегментов, url без scheme/netloc, либо не распарсился —
+    возвращает None: caller тогда просто не передаёт origin/referer дальше
+    (поведение как до #3251, ровно один голый goto(url)).
+    """
+    try:
+        parsed = urlparse(card_url)
+    except ValueError:
+        return None
+    if not parsed.scheme or not parsed.netloc:
+        return None
+    segments = [s for s in parsed.path.split("/") if s]
+    if len(segments) < 2:
+        return None
+    return f"{parsed.scheme}://{parsed.netloc}/{segments[0]}/{segments[1]}/prodam"
+
+
 # ── fetch_detail ──────────────────────────────────────────────────────────────
 async def fetch_detail(
     item_url: str,
@@ -476,17 +500,25 @@ async def fetch_detail(
     referer: str | None = None,
     reconnect_on_block: bool = True,
     config: ScraperConfig | None = None,
+    origin: str | None = None,
+    browser_referer: str | None = None,
 ) -> DetailEnrichment:
     """GET <avito>/{item_url} → parse HTML via selectolax → DetailEnrichment.
 
     Если browser_fetcher передан — использует браузерный fetch (browser mode).
     Если cffi_session не передана — создаёт новую (impersonate=DEFAULT_IMPERSONATE).
     referer — если задан, шлётся в Referer-заголовке detail-GET'а (warm-batch #1551:
-    referer = URL avito-search страницы, на которой прогрелась сессия).
+    referer = URL avito-search страницы, на которой прогрелась сессия). Это ТОЛЬКО
+    curl-путь — НЕ путать с browser_referer ниже, они разных фетчеров и разных полей.
     reconnect_on_block — если False, 403/firewall поднимает AvitoBlockedError СРАЗУ
     (без холодных reconnect'ов — ими займётся outer-loop через rebuild+rewarm прогретой
     сессии); 429 после short-retry поднимает AvitoRateLimitedError. Дефолт True
     сохраняет старое поведение own-session / shared-session (run_avito_city_sweep) путей.
+    origin / browser_referer (#3180, #3251) — ТОЛЬКО browser-режим: same-site SERP-якорь,
+    на который сайдкар зайдёт ПЕРЕД карточкой (см. BrowserFetcher.fetch), и Referer
+    целевой навигации. Рецепт зеркалит DomClick (#3247): переход на карточку логически
+    и есть переход из выдачи. None (дефолт) → оба поля не кладутся в /fetch payload,
+    поведение остальных вызовов (city_sweep, pipeline, admin) не меняется НИ НА БАЙТ.
     Raises:
         httpx.HTTPError — если status != 200 (через raise_for_status-like).
         ValueError — если item_id не извлечён из HTML.
@@ -503,7 +535,7 @@ async def fetch_detail(
         # дальше, НЕ трогая curl.
         full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
         try:
-            html = await browser_fetcher.fetch(full_url)
+            html = await browser_fetcher.fetch(full_url, origin=origin, referer=browser_referer)
         except (AvitoBlockedError, AvitoRateLimitedError):
             raise
         except Exception as exc: