feat(tradein/browser): имитация чтения страницы человеком, целиком opt-in (#3283) (#3308)

Co-authored-by: lekss361 <lekss361@gendsgn.local>
Co-committed-by: lekss361 <lekss361@gendsgn.local>

tradein-mvp/browser/server.py

@@ -141,7 +141,10 @@ import base64
 import gzip
 import logging
 import os
+import random
+import re
 from collections.abc import Callable, Mapping
+from typing import NamedTuple
 from urllib.parse import quote, urlparse
 
 from aiohttp import web
@@ -170,6 +173,31 @@ BROWSER_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_NAV_TIMEOUT_MS", "6000
 # выдача (~50 карточек, 3.2МБ). Подтверждено прод-дебагом 2026-05-31.
 BROWSER_WAIT_MS: int = int(os.environ.get("BROWSER_WAIT_MS", "6000"))
 
+# ── Имитация чтения человеком (#3283, opt-in через body["behaviour"]) ───────────
+# Все константы ниже включаются ТОЛЬКО явным behaviour.scroll/dwell_ms/hop_recommended
+# в теле /fetch — по умолчанию (behaviour отсутствует/пуст) ни одна из них не читается
+# ни разу, поведение сайдкара байт-в-байт то же, что и до #3283. Это осознанное
+# требование задачи: вклад имитации в проходимость анти-бота пока НЕ доказан, есть
+# только косвенные основания, и его нужно честно померить прод-A/B "с behaviour vs
+# без", а не вкатывать по умолчанию.
+#
+# Число и разброс прокруток (500-1400px, 3-6 раз) и пауза между ними (700-1800мс) —
+# из замера, которым это поведение и мерили (см. PR/issue #3283), не придуманы здесь.
+BROWSER_SCROLL_MIN_COUNT: int = int(os.environ.get("BROWSER_SCROLL_MIN_COUNT", "3"))
+BROWSER_SCROLL_MAX_COUNT: int = int(os.environ.get("BROWSER_SCROLL_MAX_COUNT", "6"))
+BROWSER_SCROLL_MIN_PX: int = int(os.environ.get("BROWSER_SCROLL_MIN_PX", "500"))
+BROWSER_SCROLL_MAX_PX: int = int(os.environ.get("BROWSER_SCROLL_MAX_PX", "1400"))
+BROWSER_SCROLL_PAUSE_MIN_MS: int = int(os.environ.get("BROWSER_SCROLL_PAUSE_MIN_MS", "700"))
+BROWSER_SCROLL_PAUSE_MAX_MS: int = int(os.environ.get("BROWSER_SCROLL_PAUSE_MAX_MS", "1800"))
+
+# hop_recommended (#3283): отдельный, НАМЕРЕННО короткий таймаут навигации на
+# рекомендательную ссылку — не наследует BROWSER_NAV_TIMEOUT_MS целиком, чтобы один
+# залипший побочный заход не съедал бюджет основного /fetch. И пауза "подержать
+# страницу пару секунд" перед закрытием.
+BROWSER_HOP_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_HOP_NAV_TIMEOUT_MS", "15000"))
+BROWSER_HOP_DWELL_MIN_MS: int = int(os.environ.get("BROWSER_HOP_DWELL_MIN_MS", "1500"))
+BROWSER_HOP_DWELL_MAX_MS: int = int(os.environ.get("BROWSER_HOP_DWELL_MAX_MS", "3000"))
+
 # 30000: цепочка PoW-челленджа Авито — startPow() в JS, затем setTimeout(3000) на
 # self-reload, затем повторная гидрация страницы. 3с таймера самой площадки — это
 # ТОЛЬКО задержка перед reload, не бюджет на сам расчёт PoW: под headless-браузером
@@ -222,6 +250,33 @@ FETCH_JSON_LOAD_WAIT_MS: int = int(os.environ.get("FETCH_JSON_LOAD_WAIT_MS", "15
 # общем браузере значило бы мешать её сессию с чужими запросами.
 PROVIDERS: tuple[str, ...] = ("avito", "cian", "yandex", "generic", "domclick")
 
+# Ссылки для behaviour.hop_recommended (#3283) — per-provider, тем же механизмом,
+# что PROVIDERS/_provider_for_url, БЕЗ хардкода площадки в общем коде хопа
+# (см. _hop_to_recommended_link).
+#
+# ДВЕ ступени, потому что одной CSS-селектора мало: он сужает до нужного раздела,
+# а отличить карточку объявления от рубрики/пагинации/футера умеет только шаблон
+# самого URL. Без второй ступени "a[href]" на карточке Авито подберёт навигацию,
+# хлебные крошки и футер — то есть хоп уводил бы не на соседнее объявление, а на
+# главную, и смысл следа терялся бы.
+#
+# Шаблон Авито снят на живом замере 31.08.2026 (scratchpad/lifetime.py, 39 карточек):
+# ссылка на объявление о квартире — путь с /kvartiry/ и id из 7+ цифр после "_" в
+# конце (возможен ?context=...). Домклик — по логам сайдкара того же дня:
+# /card/sale__flat__<id>.
+#
+# cian/yandex/generic шаблона НЕ имеют осознанно: их вёрстку я не снимал, а
+# выдуманный шаблон хуже отсутствующего — при None вторая ступень не применяется
+# и работает прежнее поведение (любая ссылка того же хоста).
+_RECOMMENDATION_LINK_SELECTOR_BY_PROVIDER: dict[str, str] = {p: "a[href]" for p in PROVIDERS}
+_RECOMMENDATION_LINK_SELECTOR_BY_PROVIDER["avito"] = 'a[href*="/kvartiry/"]'
+_RECOMMENDATION_LINK_SELECTOR_BY_PROVIDER["domclick"] = 'a[href*="/card/"]'
+
+_RECOMMENDATION_LINK_PATTERN_BY_PROVIDER: dict[str, re.Pattern[str]] = {
+    "avito": re.compile(r"_\d{7,}(?:\?|$)"),
+    "domclick": re.compile(r"/card/sale__flat__\d+"),
+}
+
 
 def _parse_bool(raw: str | None, default: bool) -> bool:
     """Парсит булев env-флаг ("true"/"1"/"yes"/"on" → True), fallback на default."""
@@ -261,6 +316,18 @@ _BLOCKED_TYPES: frozenset[str] = _parse_block_types(
 # прежде чем открыть новую страницу. 0 = без пейсинга.
 BROWSER_MIN_PAGE_INTERVAL_S: float = float(os.environ.get("BROWSER_MIN_PAGE_INTERVAL_S", "2.0"))
 
+# Джиттер поверх интервала пейсинга (#3283) — сейчас пейсинг всегда ждёт РОВНО
+# BROWSER_MIN_PAGE_INTERVAL_S, без разброса, что само по себе регулярный,
+# бото-подобный ритм запросов. Дефолт 0 (БЕЗ джиттера) — намеренно: как и
+# behaviour.* в /fetch, вклад разброса интервала в проходимость анти-бота отдельно
+# не измерен, а включение по умолчанию сломало бы ту же прод-измеримость "с
+# джиттером vs без", которую задача защищает для behaviour. Включается явно через
+# env, без релиза кода — 0 сохраняет прежнее поведение _pace_provider буквально
+# (interval + random.uniform(0, 0) == interval).
+BROWSER_PAGE_INTERVAL_JITTER_S: float = float(
+    os.environ.get("BROWSER_PAGE_INTERVAL_JITTER_S", "0")
+)
+
 # Провайдеры, для которых якорная вкладка (#3251) заходит на origin ЧЕРЕЗ реальный
 # поиск yandex.ru, а не голым goto. Дефолт "domclick" — авито/циан/яндекс проверяются
 # отдельно (#3251), не расширять этот список без замера на них. Пустая строка env =
@@ -1299,6 +1366,46 @@ async def fetch_handler(request: web.Request) -> web.Response:
             status=400,
         )
 
+    # behaviour (опционально, #3283) — имитация чтения страницы человеком: scroll,
+    # dwell_ms, hop_recommended. Отсутствует/пуст → behaviour остаётся None и ни
+    # ОДНА строка нового кода (_apply_scroll_behaviour/_apply_dwell_behaviour/
+    # _hop_to_recommended_link) не исполняется — /fetch ведёт себя байт-в-байт как до
+    # #3283. Это намеренно: вклад имитации в проходимость анти-бота пока НЕ доказан,
+    # и его нужно честно померить прод-A/B "с behaviour vs без" — включать что-либо
+    # из этого по умолчанию значило бы сломать именно этот замер.
+    behaviour_raw = body.get("behaviour") or {}
+    if not isinstance(behaviour_raw, dict):
+        return web.json_response(
+            {"error": f"'behaviour' должен быть object, получено "
+                      f"{type(behaviour_raw).__name__}"},
+            status=400,
+        )
+    behaviour: _Behaviour | None = None
+    if behaviour_raw:
+        dwell_ms_raw = behaviour_raw.get("dwell_ms")
+        dwell_ms: tuple[int, int] | None = None
+        if dwell_ms_raw is not None:
+            is_valid_pair = (
+                isinstance(dwell_ms_raw, (list, tuple))
+                and len(dwell_ms_raw) == 2
+                and all(isinstance(v, int) and not isinstance(v, bool) for v in dwell_ms_raw)
+                and dwell_ms_raw[0] >= 0
+                and dwell_ms_raw[1] >= 0
+                and dwell_ms_raw[0] <= dwell_ms_raw[1]
+            )
+            if not is_valid_pair:
+                return web.json_response(
+                    {"error": f"'behaviour.dwell_ms' должен быть [min, max] — два int "
+                              f">= 0, min <= max, получено {dwell_ms_raw!r}"},
+                    status=400,
+                )
+            dwell_ms = (int(dwell_ms_raw[0]), int(dwell_ms_raw[1]))
+        behaviour = _Behaviour(
+            scroll=bool(behaviour_raw.get("scroll", False)),
+            dwell_ms=dwell_ms,
+            hop_recommended=bool(behaviour_raw.get("hop_recommended", False)),
+        )
+
     provider = _resolve_provider(body, url)
     proxy_override = _resolve_proxy_override(body, provider)
 
@@ -1340,6 +1447,7 @@ async def fetch_handler(request: web.Request) -> web.Response:
                 reuse_context=reuse_context,
                 reset_context=reset_context,
                 fetch_mode=fetch_mode,
+                behaviour=behaviour,
             )
             # Читаем статус под тем же локом, что и сам фетч, — иначе следующий
             # запрос этого провайдера успел бы его перезаписать.
@@ -1488,10 +1596,16 @@ async def _pace_provider(provider: str) -> None:
     Записывает момент ПОСЛЕ возможного sleep (т.е. фактический момент следующего goto),
     чтобы интервал считался между реальными навигациями, а не между входами в функцию.
     No-op при эффективном интервале провайдера <= 0.
+
+    Джиттер (#3283) — BROWSER_PAGE_INTERVAL_JITTER_S добавляется К интервалу на КАЖДЫЙ
+    вызов (``random.uniform(0, JITTER_S)``), а не в остаток ожидания: ровный интервал —
+    сам по себе регулярный, бото-подобный ритм. Дефолт джиттера 0 → uniform(0, 0) == 0.0
+    → interval не меняется, поведение функции идентично прежнему.
     """
-    interval = _MIN_PAGE_INTERVAL_BY_PROVIDER.get(provider, BROWSER_MIN_PAGE_INTERVAL_S)
+    base_interval = _MIN_PAGE_INTERVAL_BY_PROVIDER.get(provider, BROWSER_MIN_PAGE_INTERVAL_S)
-    if interval <= 0:
+    if base_interval <= 0:
         return
+    interval = base_interval + random.uniform(0, BROWSER_PAGE_INTERVAL_JITTER_S)
     now = asyncio.get_event_loop().time()
     last = _last_goto_at.get(provider)
     if last is not None:
@@ -1505,6 +1619,116 @@ async def _pace_provider(provider: str) -> None:
     _last_goto_at[provider] = asyncio.get_event_loop().time()
 
 
+class _Behaviour(NamedTuple):
+    """Разобранное и провалидированное тело body["behaviour"] (#3283, /fetch).
+
+    Собирается ОДИН раз в fetch_handler; None на всём пути ниже (_do_fetch/_fetch_once)
+    означает «behaviour отсутствовал/пуст» — ни одна из функций _apply_scroll_behaviour/
+    _apply_dwell_behaviour/_hop_to_recommended_link не вызывается, поведение то же, что
+    и до #3283.
+    """
+
+    scroll: bool
+    dwell_ms: tuple[int, int] | None
+    hop_recommended: bool
+
+
+async def _apply_scroll_behaviour(page: object, behaviour: _Behaviour) -> None:
+    """3-6 прокруток page.mouse.wheel(0, 500-1400px) с паузами 700-1800мс (behaviour.scroll).
+
+    No-op, если behaviour.scroll не выставлен явным True. Числа — именованные модульные
+    константы BROWSER_SCROLL_* (см. их комментарий рядом с BROWSER_WAIT_MS), не хардкод
+    здесь. Не оборачиваем в try/except: вызывается ТОЛЬКО при явном opt-in, отказ здесь
+    (например, страница уже начала закрываться) — тот же класс сбоя навигации, что и
+    остальной _fetch_once, caller (_do_fetch) уже умеет его обрабатывать.
+    """
+    if not behaviour.scroll:
+        return
+    count = random.randint(BROWSER_SCROLL_MIN_COUNT, BROWSER_SCROLL_MAX_COUNT)
+    for _ in range(count):
+        delta = random.randint(BROWSER_SCROLL_MIN_PX, BROWSER_SCROLL_MAX_PX)
+        await page.mouse.wheel(0, delta)  # type: ignore[attr-defined]
+        pause_ms = random.randint(BROWSER_SCROLL_PAUSE_MIN_MS, BROWSER_SCROLL_PAUSE_MAX_MS)
+        await page.wait_for_timeout(pause_ms)  # type: ignore[attr-defined]
+
+
+async def _apply_dwell_behaviour(page: object, behaviour: _Behaviour) -> None:
+    """Дополнительная случайная пауза [min, max] мс перед content() (behaviour.dwell_ms).
+
+    No-op, если dwell_ms не передан. Диапазон уже провалидирован в fetch_handler
+    (два int >= 0, min <= max) — здесь просто random.randint по нему.
+    """
+    if behaviour.dwell_ms is None:
+        return
+    low, high = behaviour.dwell_ms
+    await page.wait_for_timeout(random.randint(low, high))  # type: ignore[attr-defined]
+
+
+async def _hop_to_recommended_link(page: object, provider: str, url: str) -> None:
+    """Заходит на ОДНУ случайную ссылку того же хоста со страницы (behaviour.hop_recommended).
+
+    Best-effort и полностью изолирован от основного результата /fetch: любой отказ
+    (таймаут, ошибка навигации, бан-страница на побочном заходе) глотается и
+    логируется, наверх ничего не пробрасывается — caller (_fetch_once) вызывает это
+    ПОСЛЕ того, как основной html уже снят и провалидирован.
+
+    Ограничения (см. постановку #3283):
+      * ссылка — ТОЛЬКО с того же host, что и url (никаких внешних переходов);
+      * ссылка != url;
+      * отдельная страница context.new_page() (НЕ переиспользуем ту же page, которую
+        caller вот-вот закроет и чей html уже зафиксирован в ответе);
+      * отдельный, короткий таймаут BROWSER_HOP_NAV_TIMEOUT_MS — не наследует
+        BROWSER_NAV_TIMEOUT_MS целиком, чтобы залипший побочный заход не съедал
+        бюджет основного /fetch.
+
+    Селектор ссылок — per-provider, см. _RECOMMENDATION_LINK_SELECTOR_BY_PROVIDER
+    рядом с PROVIDERS.
+    """
+    try:
+        selector = _RECOMMENDATION_LINK_SELECTOR_BY_PROVIDER.get(provider, "a[href]")
+        hrefs: list[str] = await page.locator(selector).evaluate_all(  # type: ignore[attr-defined]
+            "els => els.map(el => el.href)"
+        )
+        target_host = (urlparse(url).hostname or "").lower()
+        pattern = _RECOMMENDATION_LINK_PATTERN_BY_PROVIDER.get(provider)
+        candidates = [
+            href
+            for href in hrefs
+            if href
+            and href != url
+            and (urlparse(href).hostname or "").lower() == target_host
+            # Вторая ступень: шаблон URL самой карточки. Нет шаблона для площадки
+            # → не сужаем, поведение как без него.
+            and (pattern is None or pattern.search(href) is not None)
+        ]
+        if not candidates:
+            logger.debug(
+                "tradein-browser[%s]: hop_recommended — нет ссылок того же хоста url=%r",
+                provider,
+                url,
+            )
+            return
+        target = random.choice(candidates)
+        hop_page = await page.context.new_page()  # type: ignore[attr-defined]
+        try:
+            await hop_page.goto(  # type: ignore[attr-defined]
+                target, timeout=BROWSER_HOP_NAV_TIMEOUT_MS, wait_until="domcontentloaded"
+            )
+            await hop_page.wait_for_timeout(  # type: ignore[attr-defined]
+                random.randint(BROWSER_HOP_DWELL_MIN_MS, BROWSER_HOP_DWELL_MAX_MS)
+            )
+        finally:
+            await hop_page.close()  # type: ignore[attr-defined]
+    except Exception as exc:
+        logger.warning(
+            "tradein-browser[%s]: hop_recommended отказ (%s: %s) — продолжаю без него url=%r",
+            provider,
+            type(exc).__name__,
+            exc,
+            url,
+        )
+
+
 # ── Режимы получения тела (#3264) ────────────────────────────────────────────────
 # Сайдкар всю жизнь умел ровно одно: navigate — page.goto(url). Для HTML-страницы это
 # и есть то, что делает человек. Но два нужных нам ресурса Домклика страницами НЕ
@@ -1550,6 +1774,7 @@ async def _do_fetch(
     reuse_context: bool = False,
     reset_context: bool = False,
     fetch_mode: str = "navigate",
+    behaviour: "_Behaviour | None" = None,
 ) -> str:
     """Одна попытка навигации; при краше браузера — relaunch и один retry.
 
@@ -1561,7 +1786,9 @@ async def _do_fetch(
     а не обратится к context'у уже мёртвого браузера.
 
     origin/referer/cookies/reuse_context/reset_context — см. _fetch_once. Дефолты не
-    меняют поведение.
+    меняют поведение. behaviour (#3283) — None (дефолт) → ни одна из
+    _apply_scroll_behaviour/_apply_dwell_behaviour/_hop_to_recommended_link не
+    вызывается, поведение то же, что и до #3283.
     """
     try:
         return await _fetch_once(
@@ -1573,6 +1800,7 @@ async def _do_fetch(
             reuse_context=reuse_context,
             reset_context=reset_context,
             fetch_mode=fetch_mode,
+            behaviour=behaviour,
         )
     except Exception as exc:
         if _is_browser_crash(exc):
@@ -1594,6 +1822,7 @@ async def _do_fetch(
                 reuse_context=reuse_context,
                 reset_context=reset_context,
                 fetch_mode=fetch_mode,
+                behaviour=behaviour,
             )
         raise
 
@@ -1984,12 +2213,20 @@ async def _fetch_once(
     reuse_context: bool = False,
     reset_context: bool = False,
     fetch_mode: str = "navigate",
+    behaviour: "_Behaviour | None" = None,
 ) -> str:
     """Открывает страницу, переходит по URL, ждёт JS, возвращает HTML.
 
     Caller держит _locks[provider], поэтому страницы на этом инстансе не
     параллелятся — recycle через _relaunch_browser безопасен прямо здесь.
 
+    behaviour (#3283, опционально, дефолт None) — имитация чтения человеком:
+    scroll (прокрутки перед content()), dwell_ms (доп. пауза перед content()),
+    hop_recommended (побочный заход на рекомендательную ссылку ПОСЛЕ content(), только
+    fetch_mode="navigate" и reuse_context=True — иначе контекста для него нет).
+    None (дефолт) → ни одна из веток ниже не исполняется, поведение то же, что и до
+    #3283.
+
     origin (опционально) — same-site якорь (например SERP), на который камуфокс
     заходит ПЕРЕД целевым url, чтобы получить пропуск QRATOR в контексте (зеркалит
     _fetch_json_once, #1917 — DomClick card-fetch, эмпирически подтверждено вживую
@@ -2107,6 +2344,11 @@ async def _fetch_once(
         _last_response_status[provider] = _status_of(response)
         if BROWSER_WAIT_MS > 0:
             await page.wait_for_timeout(BROWSER_WAIT_MS)  # type: ignore[attr-defined]
+        # Имитация чтения (#3283) — ПОСЛЕ успешной навигации и settle, ДО content():
+        # behaviour=None (дефолт) → обе функции no-op, ни одна строка не исполняется.
+        if behaviour is not None:
+            await _apply_scroll_behaviour(page, behaviour)
+            await _apply_dwell_behaviour(page, behaviour)
         html: str = await page.content()  # type: ignore[attr-defined]
 
         # Бан-страница («проблема с IP») распознаётся и падает СРАЗУ, без траты
@@ -2162,6 +2404,14 @@ async def _fetch_once(
                     f"— отказ площадки, ждать нечего url={url!r}"
                 )
             html = await _wait_out_pow_challenge(page, provider, url)
+
+        # hop_recommended (#3283) — ПОСЛЕ того как html снят и провалидирован (бан/
+        # челлендж уже исключены выше, иначе исключение долетело бы до caller'а раньше
+        # этой строки). Только reuse_context=True — без него контекста, из которого
+        # можно open'нуть побочную вкладку не задев основную page, нет. Best-effort:
+        # см. докстринг _hop_to_recommended_link — отказ на нём не всплывает сюда.
+        if behaviour is not None and behaviour.hop_recommended and reuse_context:
+            await _hop_to_recommended_link(page, provider, url)
     finally:
         await page.close()  # type: ignore[attr-defined]
 

tradein-mvp/browser/test_server.py

@@ -183,6 +183,7 @@ class _OverlapProbe:
         reuse_context: bool = False,
         reset_context: bool = False,
         fetch_mode: str = "navigate",
+        behaviour: object = None,
     ) -> str:
         self.active[provider] = self.active.get(provider, 0) + 1
         self.peak[provider] = max(self.peak.get(provider, 0), self.active[provider])

tradein-mvp/browser/test_server_behaviour.py

@@ -0,0 +1,349 @@
+"""test_server_behaviour.py — имитация чтения страницы человеком (#3283).
+
+По умолчанию (body["behaviour"] отсутствует/пуст) поведение /fetch байт-в-байт то
+же, что и до #3283 — вклад имитации в проходимость анти-бота пока НЕ доказан, и
+прод должен уметь честно померить "с behaviour vs без". Проверяет:
+
+  1. behaviour отсутствует → ни page.mouse.wheel, ни доп. dwell-пауза, ни побочный
+     заход на рекомендательную ссылку НЕ вызываются;
+  2. behaviour.scroll=True → 3-6 вызовов page.mouse.wheel(0, 500-1400px);
+  3. behaviour.dwell_ms — валидный [min, max] отрабатывает; невалидный (не пара
+     int, min>max, отрицательные) — /fetch отвечает 400, как и невалидный
+     fetch_mode;
+  4. behaviour.hop_recommended — заходит на ОДНУ ссылку ТОГО ЖЕ хоста, отказ на
+     этом заходе не роняет основной ответ; ссылка с чужого хоста не берётся;
+     работает только при reuse_context=True.
+  5. body["behaviour"] не-dict → 400.
+
+camoufox НЕ запускается — browser/context/page подделаны, в стиле
+test_server_reuse_context.py.
+
+Запуск (из tradein-mvp/browser/)::
+
+    python -m pytest test_server_behaviour.py -q
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+from aiohttp.test_utils import make_mocked_request
+
+_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
+_spec = importlib.util.spec_from_file_location("tradein_browser_server", _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)
+
+
+def _json_body(response: Any) -> dict[str, Any]:
+    return json.loads(response.body.decode())
+
+
+async def _coro(value: Any) -> Any:
+    return value
+
+
+def _make_fetch_request(body: dict[str, Any]) -> Any:
+    request = make_mocked_request("POST", "/fetch")
+    request.json = lambda: _coro(body)  # type: ignore[method-assign]
+    return request
+
+
+@pytest.fixture(autouse=True)
+def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(server, "_browsers", {})
+    monkeypatch.setattr(server, "_browser_cms", {})
+    monkeypatch.setattr(server, "_contexts", {})
+    monkeypatch.setattr(server, "_anchor_pages", {})
+    monkeypatch.setattr(server, "_page_counters", {})
+    monkeypatch.setattr(server, "_locks", {})
+    monkeypatch.setattr(server, "_last_goto_at", {})
+    monkeypatch.setattr(
+        server, "_RECYCLE_PAGES_BY_PROVIDER", {p: 10_000 for p in server.PROVIDERS}
+    )
+    # Джиттер пейсинга (#3283) не имеет отношения к behaviour — гасим, чтобы не
+    # добавлять произвольный real-time sleep в тест.
+    monkeypatch.setattr(server, "_MIN_PAGE_INTERVAL_BY_PROVIDER", {p: 0.0 for p in server.PROVIDERS})
+
+
+_DEFAULT_HTML = "<html>ok<script>window.__SSR_STATE__={}</script></html>"
+
+
+class _FakeMouse:
+    def __init__(self) -> None:
+        self.wheel_calls: list[tuple[int, int]] = []
+
+    async def wheel(self, dx: int, dy: int) -> None:
+        self.wheel_calls.append((dx, dy))
+
+
+class _FakeLocator:
+    def __init__(self, hrefs: list[str]) -> None:
+        self._hrefs = hrefs
+
+    async def evaluate_all(self, _js: str) -> list[str]:
+        return self._hrefs
+
+
+class _FakePage:
+    """Поддельная page: goto/content/mouse/locator/context, минимальный контракт _fetch_once."""
+
+    def __init__(self, context: "_FakeContext", links: list[str] | None = None) -> None:
+        self.context = context
+        self.mouse = _FakeMouse()
+        self.goto_urls: list[str] = []
+        self.wait_for_timeout_calls: list[int] = []
+        self.closed = 0
+        self._links = links or []
+
+    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)
+
+    async def wait_for_timeout(self, ms: int) -> None:
+        self.wait_for_timeout_calls.append(ms)
+
+    async def content(self) -> str:
+        return _DEFAULT_HTML
+
+    def locator(self, selector: str) -> _FakeLocator:
+        return _FakeLocator(self._links)
+
+    async def close(self) -> None:
+        self.closed += 1
+
+
+class _FailingHopPage(_FakePage):
+    """Побочная (hop) страница, у которой goto всегда падает — для теста best-effort."""
+
+    async def goto(self, url: str, **kwargs: Any) -> None:
+        raise TimeoutError("hop nav timeout (simulated)")
+
+
+class _FakeContext:
+    def __init__(self, links: list[str] | None = None, hop_page_cls: type = _FakePage) -> None:
+        self.add_cookies_calls: list[list[dict]] = []
+        self.pages: list[_FakePage] = []
+        self._links = links or []
+        self._hop_page_cls = hop_page_cls
+        self.closed = 0
+
+    async def add_cookies(self, cookies: list[dict]) -> None:
+        self.add_cookies_calls.append(cookies)
+
+    async def new_page(self) -> _FakePage:
+        # Первая страница — "основная" (несёт links для hop-детекции), последующие —
+        # побочные (hop) вкладки, открытые из _hop_to_recommended_link.
+        cls = _FakePage if not self.pages else self._hop_page_cls
+        page = cls(self, links=self._links)
+        self.pages.append(page)
+        return page
+
+    async def close(self) -> None:
+        self.closed += 1
+
+
+class _FakeReuseBrowser:
+    def __init__(self, links: list[str] | None = None, hop_page_cls: type = _FakePage) -> None:
+        self.contexts_created: list[_FakeContext] = []
+        self._links = links or []
+        self._hop_page_cls = hop_page_cls
+
+    async def new_context(self) -> _FakeContext:
+        ctx = _FakeContext(links=self._links, hop_page_cls=self._hop_page_cls)
+        self.contexts_created.append(ctx)
+        return ctx
+
+    async def new_page(self) -> _FakePage:
+        # Не переиспользуемый путь (reuse_context=False) в этих тестах не участвует.
+        raise AssertionError("browser.new_page() не должен вызываться при reuse_context=True")
+
+
+# ── 1. behaviour отсутствует — ни одна новая ветка не исполняется ────────────────
+
+
+def test_behaviour_absent_is_noop() -> None:
+    browser = _FakeReuseBrowser()
+    server._browsers["avito"] = browser
+
+    html = asyncio.run(
+        server._fetch_once("avito", "https://www.avito.ru/1", reuse_context=True)
+    )
+
+    assert html == _DEFAULT_HTML
+    page = browser.contexts_created[0].pages[0]
+    assert page.mouse.wheel_calls == [], "без behaviour scroll не должен вызываться"
+    # Единственный wait_for_timeout — обязательный BROWSER_WAIT_MS settle, не dwell.
+    assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS]
+    assert len(browser.contexts_created[0].pages) == 1, "hop_recommended не должен заходить"
+
+
+# ── 2. scroll=True — 3-6 вызовов wheel в заданном диапазоне px ───────────────────
+
+
+def test_scroll_true_triggers_wheel_calls_in_range() -> None:
+    browser = _FakeReuseBrowser()
+    server._browsers["avito"] = browser
+    behaviour = server._Behaviour(scroll=True, dwell_ms=None, hop_recommended=False)
+
+    asyncio.run(
+        server._fetch_once(
+            "avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
+        )
+    )
+
+    calls = browser.contexts_created[0].pages[0].mouse.wheel_calls
+    assert server.BROWSER_SCROLL_MIN_COUNT <= len(calls) <= server.BROWSER_SCROLL_MAX_COUNT
+    for dx, dy in calls:
+        assert dx == 0
+        assert server.BROWSER_SCROLL_MIN_PX <= dy <= server.BROWSER_SCROLL_MAX_PX
+
+
+# ── 3. dwell_ms — валидация в fetch_handler (400, как fetch_mode) ────────────────
+
+
+@pytest.mark.parametrize(
+    "dwell_ms",
+    [
+        [500],  # не пара
+        [5000, 100],  # min > max
+        [-10, 100],  # отрицательное
+        ["a", "b"],  # не int
+        "500,1000",  # вообще не список
+    ],
+)
+def test_invalid_dwell_ms_returns_400(dwell_ms: Any) -> None:
+    request = _make_fetch_request(
+        {"url": "https://www.avito.ru/1", "behaviour": {"dwell_ms": dwell_ms}}
+    )
+    response = asyncio.run(server.fetch_handler(request))
+    assert response.status == 400
+    assert "dwell_ms" in _json_body(response)["error"]
+
+
+def test_behaviour_not_dict_returns_400() -> None:
+    request = _make_fetch_request({"url": "https://www.avito.ru/1", "behaviour": "scroll"})
+    response = asyncio.run(server.fetch_handler(request))
+    assert response.status == 400
+
+
+def test_valid_dwell_ms_applies_extra_wait() -> None:
+    browser = _FakeReuseBrowser()
+    server._browsers["avito"] = browser
+    behaviour = server._Behaviour(scroll=False, dwell_ms=(100, 100), hop_recommended=False)
+
+    asyncio.run(
+        server._fetch_once(
+            "avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
+        )
+    )
+
+    page = browser.contexts_created[0].pages[0]
+    # BROWSER_WAIT_MS settle + ровно 100мс dwell (min==max=100 — детерминировано).
+    assert page.wait_for_timeout_calls == [server.BROWSER_WAIT_MS, 100]
+
+
+# ── 4. hop_recommended — same-host only, best-effort, требует reuse_context ──────
+
+
+def test_hop_recommended_visits_same_host_link_only() -> None:
+    links = [
+        "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_505_m_1216_et._8285475966",  # == url, должен быть исключён
+        "https://sub.other.ru/kvartiry/foreign_1234567",  # чужой хост, исключён
+        "https://www.avito.ru/ekaterinburg/kvartiry",  # рубрика, не карточка — исключена
+        "https://www.avito.ru/rossiya/nedvizhimost",  # навигация, исключена
+        "https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_302_m_25_et._8242031370",  # валидный кандидат
+    ]
+    browser = _FakeReuseBrowser(links=links)
+    server._browsers["avito"] = browser
+    behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
+
+    asyncio.run(
+        server._fetch_once("avito", "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_505_m_1216_et._8285475966", reuse_context=True, behaviour=behaviour)
+    )
+
+    ctx = browser.contexts_created[0]
+    assert len(ctx.pages) == 2, "должна открыться ровно одна побочная (hop) страница"
+    hop_page = ctx.pages[1]
+    assert hop_page.goto_urls == ["https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_302_m_25_et._8242031370"]
+    assert hop_page.closed == 1, "hop-страница должна закрыться"
+
+
+def test_hop_recommended_skips_same_host_non_card_links() -> None:
+    """#3283: одного host-фильтра мало — без шаблона URL хоп уводил бы на рубрику.
+
+    Шаблон Авито (путь с /kvartiry/ + id из 7+ цифр) снят на живом замере 31.08.2026.
+    """
+    links = [
+        "https://www.avito.ru/ekaterinburg/kvartiry",
+        "https://www.avito.ru/ekaterinburg/kvartiry/prodam",
+        "https://www.avito.ru/company/about",
+    ]
+    browser = _FakeReuseBrowser(links=links)
+    server._browsers["avito"] = browser
+    behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
+
+    asyncio.run(
+        server._fetch_once("avito", "https://www.avito.ru/ekaterinburg/kvartiry/2-k._kvartira_505_m_1216_et._8285475966", reuse_context=True, behaviour=behaviour)
+    )
+
+    assert len(browser.contexts_created[0].pages) == 1, "рубрика не карточка — хопа быть не должно"
+
+
+def test_hop_recommended_skips_when_only_foreign_host_links() -> None:
+    links = ["https://sub.other.ru/kvartiry/foreign_1234567", "https://another.tld/kvartiry/x_7654321"]
+    browser = _FakeReuseBrowser(links=links)
+    server._browsers["avito"] = browser
+    behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
+
+    asyncio.run(
+        server._fetch_once(
+            "avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
+        )
+    )
+
+    assert len(browser.contexts_created[0].pages) == 1, "чужой хост не должен открывать hop"
+
+
+def test_hop_recommended_failure_does_not_break_main_response() -> None:
+    links = ["https://www.avito.ru/ekaterinburg/kvartiry/1-k._kvartira_302_m_25_et._8242031370"]
+    browser = _FakeReuseBrowser(links=links, hop_page_cls=_FailingHopPage)
+    server._browsers["avito"] = browser
+    behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
+
+    html = asyncio.run(
+        server._fetch_once(
+            "avito", "https://www.avito.ru/1", reuse_context=True, behaviour=behaviour
+        )
+    )
+
+    assert html == _DEFAULT_HTML, "отказ на hop не должен ронять/менять основной ответ"
+
+
+def test_hop_recommended_noop_without_reuse_context() -> None:
+    """hop_recommended=True, но reuse_context=False — hop не выполняется (нет context)."""
+    server._browsers["avito"] = object()
+    behaviour = server._Behaviour(scroll=False, dwell_ms=None, hop_recommended=True)
+
+    # _fetch_once без reuse_context идёт через browser.new_page() — подделаем browser.
+    class _PlainBrowser:
+        async def new_page(self) -> _FakePage:
+            return _FakePage(context=_FakeContext())
+
+    server._browsers["avito"] = _PlainBrowser()
+
+    html = asyncio.run(
+        server._fetch_once("avito", "https://www.avito.ru/1", behaviour=behaviour)
+    )
+    assert html == _DEFAULT_HTML