Merge remote-tracking branch 'upstream/dev' into feature-indeed-jobs-scraper

# Conflicts:
#	README.es.md
#	README.hi.md
#	README.md
#	README.pt-BR.md
#	README.zh-CN.md
#	surfsense_backend/tests/unit/capabilities/google_maps/test_registry.py
#	surfsense_backend/tests/unit/capabilities/reddit/test_registry.py
#	surfsense_backend/tests/unit/capabilities/youtube/test_registry.py
#	surfsense_mcp/mcp_server/features/scrapers/__init__.py
#	surfsense_web/content/docs/connectors/index.mdx
#	surfsense_web/content/docs/connectors/native/index.mdx
#	surfsense_web/content/docs/connectors/native/meta.json
#	surfsense_web/content/docs/how-to/mcp-server.mdx
#	surfsense_web/lib/connectors-marketing/index.ts
#	surfsense_web/lib/playground/catalog.ts

surfsense_backend/scripts/e2e_amazon_scraper.py

@@ -0,0 +1,84 @@
+"""Manual end-to-end check for the public Amazon scraper.
+
+Run from the backend directory:
+
+    uv run python scripts/e2e_amazon_scraper.py
+    uv run python scripts/e2e_amazon_scraper.py --refresh-fixtures
+
+The script requires live network access and the configured residential proxy.
+The optional flag replaces the product and search parser fixtures with current
+live responses. The script is intentionally excluded from pytest.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+_BACKEND_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_BACKEND_ROOT))
+for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"):
+    if _candidate.exists():
+        load_dotenv(_candidate)
+        break
+
+from app.proprietary.platforms.amazon import (  # noqa: E402
+    AmazonScrapeInput,
+    scrape_products,
+)
+from app.proprietary.platforms.amazon.fetch import fetch_page  # noqa: E402
+
+_PRODUCT_URL = "https://www.amazon.com/dp/B09V3KXJPB"
+_SEARCH_URL = "https://www.amazon.com/s?k=wireless+mouse"
+_FIXTURE_DIR = _BACKEND_ROOT / "tests" / "unit" / "platforms" / "amazon" / "fixtures"
+
+
+def _check(label: str, passed: bool) -> bool:
+    print(f"[{'PASS' if passed else 'FAIL'}] {label}")
+    return passed
+
+
+async def main() -> int:
+    product_items = await scrape_products(
+        AmazonScrapeInput(categoryOrProductUrls=[{"url": _PRODUCT_URL}]),
+        limit=1,
+    )
+    product = product_items[0] if product_items else {}
+    print(json.dumps(product, indent=2, ensure_ascii=False)[:3000])
+    product_ok = _check(
+        "product detail has identity and title",
+        bool(product.get("asin") and product.get("title")),
+    )
+
+    search_items = await scrape_products(
+        AmazonScrapeInput(
+            categoryOrProductUrls=[{"url": _SEARCH_URL}],
+            maxItemsPerStartUrl=3,
+            scrapeProductDetails=False,
+        )
+    )
+    search_ok = _check(
+        "search returns product cards",
+        bool(search_items)
+        and all(
+            item.get("asin") and item.get("categoryPageData") for item in search_items
+        ),
+    )
+    fixture_ok = True
+    if "--refresh-fixtures" in sys.argv:
+        _FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
+        for name, url in (("product.html", _PRODUCT_URL), ("search.html", _SEARCH_URL)):
+            response = await fetch_page(url)
+            saved = response is not None and response.status == 200
+            if saved:
+                (_FIXTURE_DIR / name).write_text(response.html, encoding="utf-8")
+            fixture_ok &= _check(f"refreshed {name}", saved)
+    return 0 if product_ok and search_ok and fixture_ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))

surfsense_backend/scripts/e2e_google_search.py

@@ -28,9 +28,12 @@ sys.path.insert(0, str(_ROOT))
 load_dotenv(_ROOT / ".env")
 
 logging.basicConfig(level=logging.WARNING)
-logging.getLogger("app.proprietary.platforms.google_search.scraper").setLevel(
-    logging.INFO
-)
+for _name in (
+    "app.proprietary.platforms.google_search.scraper",
+    "app.proprietary.platforms.google_search.fetch",
+    "app.proprietary.platforms.google_search.captcha",
+):
+    logging.getLogger(_name).setLevel(logging.INFO)
 
 from app.proprietary.platforms.google_search import (  # noqa: E402
     GoogleSearchScrapeInput,
@@ -146,6 +149,9 @@ async def run(
 
 _CASES = {
     "plain": lambda: run("plain query", queries="python asyncio tutorial"),
+    # Prod incident 2026-07-17: single-word brand/navigational queries were
+    # exhausting all 24 IPs (precheck 200 but the browser render 429-walled).
+    "brand": lambda: run("brand query (notebooklm, prod repro)", queries="notebooklm"),
     "site": lambda: run("site: filter", queries="machine learning", site="arxiv.org"),
     "ads": lambda: run("text ads", queries="car insurance quotes", expect_ads=True),
     "products": lambda: run(

surfsense_backend/scripts/e2e_instagram_scraper.py

@@ -51,8 +51,14 @@ from app.proprietary.platforms.instagram.fetch import (  # noqa: E402
 )
 from app.proprietary.platforms.instagram.url_resolver import resolve_url  # noqa: E402
 
-_PROFILE = "natgeo"
-_SEARCH_TERM = "national geographic"
+# Canonical public targets. Override from the CLI to test any real-world case:
+#   python scripts/e2e_instagram_scraper.py <profile> [search term]
+# Note: web_profile_info intermittently 400s for *business/creator* accounts
+# (IG server bug on the ig_business_category_subvertical schema); a regular
+# public account is the reliable smoke target.
+_DEFAULT_PROFILE = "natgeo"
+_PROFILE = sys.argv[1] if len(sys.argv) > 1 else _DEFAULT_PROFILE
+_SEARCH_TERM = sys.argv[2] if len(sys.argv) > 2 else "national geographic"
 
 _FIXTURE_DIR = _BACKEND_ROOT / "tests" / "unit" / "platforms" / "instagram" / "fixtures"
 
@@ -179,6 +185,12 @@ async def step5_search() -> bool:
 
 async def step6_dump_fixtures(post_url: str | None) -> bool:
     _hr("STEP 6 — dump trimmed, anonymized fixtures for offline tests")
+    if _PROFILE != _DEFAULT_PROFILE:
+        return _check(
+            "dumped fixtures",
+            True,
+            f"skipped (custom profile {_PROFILE!r} would clobber committed fixtures)",
+        )
     profile = await fetch_json("api/v1/users/web_profile_info/", {"username": _PROFILE})
     _FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
     wrote = []

surfsense_backend/scripts/e2e_reddit_scraper.py

@@ -9,10 +9,10 @@ This is NOT a pytest test (it needs live network + a residential/custom proxy).
 It:
 
   Step 0 — go/no-go probe (folds in the old scripts/reddit_probe.py): open a
-      proxy session, warm a ``loid`` (svc/shreddit first, old.reddit fallback),
-      then do sequential ``.json`` fetches on the SAME sticky IP and assert each
-      returns a Reddit Listing. If this fails the whole approach is invalid —
-      later steps are skipped.
+      proxy session, browser-warm a ``loid`` cookie jar on the sticky exit IP,
+      then do sequential plain-HTTP ``.json`` fetches on the SAME sticky IP
+      (replaying the jar) and assert each returns a Reddit Listing. If this fails
+      the whole approach is invalid — later steps are skipped.
   Step 1 — scrape a discovered post URL (post + a few comments).
   Step 2 — scrape a subreddit listing.
   Step 3 — run a search query.
@@ -96,8 +96,11 @@ async def step0_probe() -> bool:
             return _check(
                 "proxy configured", False, "no proxy -> set PROXY_PROVIDER + creds"
             )
-        minted = await warm_session(holder.session)
+        jar = await warm_session(holder.proxy)
+        if jar:
+            holder.cookies = jar
         holder.warmed = True  # don't let fetch_json re-warm; we just warmed it
+        minted = bool(jar)
         _check("loid warm-up minted a session", minted)
         oks: list[bool] = []
         for path in (f"r/{_SUBREDDIT}/hot", "r/programming/new", f"r/{_SUBREDDIT}/hot"):

surfsense_backend/scripts/e2e_tiktok_scrape.py

@@ -39,6 +39,12 @@ from urllib.parse import urlsplit
 from dotenv import load_dotenv
 
 # --- bootstrap: load .env and put the backend root on sys.path before app.* ---
+# Scraped captions carry arbitrary Unicode (emoji, CJK, decorative glyphs); the
+# Windows cp1252 console can't encode it and would abort the whole run on a single
+# character. Emit UTF-8 and replace anything un-encodable instead of crashing.
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+
 _BACKEND_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(_BACKEND_ROOT))
 for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"):
@@ -50,7 +56,7 @@ _FIXTURES = _BACKEND_ROOT / "tests" / "fixtures" / "tiktok"
 
 # Evergreen public targets: a regular high-volume creator, a broad hashtag, and
 # a common search term.
-_PROFILE = "nasa"
+_PROFILE = "tiktok"
 _HASHTAG = "food"
 _SEARCH = "meal prep"
 _COUNT = 5

surfsense_backend/scripts/scale_google_search.py

@@ -0,0 +1,244 @@
+"""Zero-cost concurrency simulation for the Google Search fetch seam.
+
+Drives many concurrent ``fetch_serp_html`` calls with the network I/O stubbed
+out (no live Google, no proxy spend, no captcha solves) so we can measure the
+*structure* of the scraper under load without paying for it:
+
+* **throughput** — completions/sec at steady state (the per-process ceiling is
+  the render gate divided by warm-render latency; nothing above it is real),
+* **latency** — p50/p95/p99 wait as the gate queues excess renders,
+* **solves** — how many paid solves the run would cost (the number we most want
+  the sticky-IP pool to shrink), and
+* **per-IP peak concurrency** — the funneling metric: how many renders pile onto
+  a single sticky IP at once (a hot IP is what Google re-walls).
+
+The stub keeps the *real* ``fetch_serp_html`` loop intact (sticky reuse,
+exemption skip-precheck, the render gate, inflight accounting) and only replaces
+``_precheck`` / ``_get_session`` / ``get_proxy_url``. Timings are compressed
+(warm≈0.1 s vs ~14 s live) and extrapolated to real seconds via the warm-render
+ratio, since the system is gate-bounded sleeps + a semaphore (delays scale
+linearly).
+
+    .venv/Scripts/python.exe scripts/scale_google_search.py --count 400 --rate 60
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+import threading
+import time
+from pathlib import Path
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
+_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_ROOT))
+
+from app.proprietary.platforms.google_search import fetch  # noqa: E402
+
+# Live-measured render latencies (README "Timings"); used only to translate the
+# compressed sim clock back to real seconds for the human-readable report.
+_REAL_WARM_S = 14.0
+_REAL_SOLVE_S = 45.0
+
+_RESULTS_HTML = '<div id="rso">ok</div>'  # trips fetch._has_results
+
+
+class _Metrics:
+    """Browser-loop-side counters (fetch runs on two loops → guard with a lock)."""
+
+    def __init__(self) -> None:
+        self.lock = threading.Lock()
+        self.solves = 0
+        self.renders = 0
+        self.ip_hits: dict[str, int] = {}
+        self.ip_now: dict[str, int] = {}
+        self.ip_peak: dict[str, int] = {}
+
+    def enter(self, proxy: str, solved: bool) -> None:
+        with self.lock:
+            self.renders += 1
+            if solved:
+                self.solves += 1
+            self.ip_hits[proxy] = self.ip_hits.get(proxy, 0) + 1
+            n = self.ip_now.get(proxy, 0) + 1
+            self.ip_now[proxy] = n
+            self.ip_peak[proxy] = max(self.ip_peak.get(proxy, 0), n)
+
+    def exit(self, proxy: str) -> None:
+        with self.lock:
+            self.ip_now[proxy] = self.ip_now.get(proxy, 1) - 1
+
+
+class _FakePage:
+    def __init__(self) -> None:
+        self.status = 200
+        self.html_content = _RESULTS_HTML
+
+
+class _FakeSession:
+    """Stands in for AsyncStealthySession: sleeps like a render, simulates the
+    solve on a cold IP by seeding fetch._exemption_jar (what page_action does)."""
+
+    def __init__(self, metrics: _Metrics, warm_s: float, solve_s: float) -> None:
+        self.m = metrics
+        self.warm_s = warm_s
+        self.solve_s = solve_s
+
+    async def fetch(self, url, proxy=None, **kwargs):
+        key = proxy or ""
+        # Cold IP (no cached exemption) pays the solve and warms the jar; a warm
+        # IP just pays the render. Mirrors _make_page_action's cache write.
+        cold = key not in fetch._exemption_jar
+        self.m.enter(key, solved=cold)
+        try:
+            if cold:
+                await asyncio.sleep(self.solve_s)
+                fetch._exemption_jar[key] = [{"name": "GOOGLE_ABUSE_EXEMPTION"}]
+            await asyncio.sleep(self.warm_s)
+            return _FakePage()
+        finally:
+            self.m.exit(key)
+
+
+def _install_stubs(metrics: _Metrics, warm_s: float, solve_s: float, precheck_s: float):
+    base = "http://u:p@gw.dataimpulse.com:823"  # hostname in fetch._STICKY_HOSTS
+    session = _FakeSession(metrics, warm_s, solve_s)
+
+    async def fake_precheck(url, proxy):
+        await asyncio.sleep(precheck_s)
+        return True
+
+    async def fake_get_session(mobile):
+        return session
+
+    fetch.get_proxy_url = lambda: base
+    fetch._precheck = fake_precheck
+    fetch._get_session = fake_get_session
+    fetch.captcha_enabled = lambda: True
+    fetch.get_captcha_config = lambda: object()
+    fetch._captcha.solver_latched = lambda: False
+    # The pool's backpressure poll is negligible vs a real 14 s render; keep it
+    # proportional under the compressed sim clock so waiters don't idle.
+    fetch._POOL_WAIT_S = max(warm_s * 0.05, 0.001)
+
+    # Isolate the sim to the LOCAL pool: the cross-process store (Redis) has its
+    # own unit test and would otherwise add ping latency / cross-run bleed here.
+    async def _no_adopt(exclude):
+        return None
+
+    async def _noop(*a, **k):
+        return None
+
+    fetch._store.adopt = _no_adopt
+    fetch._store.publish = _noop
+    fetch._store.evict = _noop
+
+
+def _reset_state() -> None:
+    fetch._exemption_jar.clear()
+    for name in ("_pool", "_pool_inflight"):
+        obj = getattr(fetch, name, None)
+        if isinstance(obj, dict):
+            obj.clear()
+    if hasattr(fetch, "_pool_pending"):
+        fetch._pool_pending = 0
+    if hasattr(fetch, "_last_good_proxy"):  # pre-pool revision
+        fetch._last_good_proxy = None
+
+
+async def _drive(count: int, rate: float) -> list[float]:
+    """Fire ``count`` fetches at ``rate``/sec (steady arrival), return latencies."""
+    lat: list[float] = []
+    interval = 1.0 / rate if rate > 0 else 0.0
+
+    async def one() -> None:
+        t0 = time.perf_counter()
+        await fetch.fetch_serp_html("https://www.google.com/search?q=notebooklm")
+        lat.append(time.perf_counter() - t0)
+
+    tasks: list[asyncio.Task] = []
+    for _ in range(count):
+        tasks.append(asyncio.create_task(one()))
+        if interval:
+            await asyncio.sleep(interval)
+    await asyncio.gather(*tasks)
+    return lat
+
+
+def _pct(xs: list[float], p: float) -> float:
+    if not xs:
+        return 0.0
+    xs = sorted(xs)
+    return xs[min(len(xs) - 1, int(p / 100 * len(xs)))]
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--count", type=int, default=400)
+    ap.add_argument("--rate", type=float, default=60.0, help="arrivals/sec (sim clock)")
+    ap.add_argument(
+        "--warm", type=float, default=0.10, help="warm render seconds (sim)"
+    )
+    ap.add_argument("--solve", type=float, default=0.45, help="solve seconds (sim)")
+    ap.add_argument("--precheck", type=float, default=0.01)
+    args = ap.parse_args()
+
+    metrics = _Metrics()
+    _install_stubs(metrics, args.warm, args.solve, args.precheck)
+    _reset_state()
+
+    scale = _REAL_WARM_S / args.warm  # sim seconds → real seconds
+    gate = fetch._MAX_CONCURRENT_PAGES
+    pool = getattr(fetch, "_WARM_POOL_TARGET", None)
+
+    wall0 = time.perf_counter()
+    lat = await _drive(args.count, args.rate)
+    wall = time.perf_counter() - wall0
+    await fetch._in_browser_loop(asyncio.sleep(0))  # let last exit() settle
+
+    hot_peak = max(metrics.ip_peak.values()) if metrics.ip_peak else 0
+    top_hits = max(metrics.ip_hits.values()) if metrics.ip_hits else 0
+    top_share = 100.0 * top_hits / metrics.renders if metrics.renders else 0.0
+    thru_sim = args.count / wall if wall else 0.0
+    thru_real = thru_sim / scale
+
+    print("\n=== Google Search scale simulation ===")
+    print(
+        f"  gate (_MAX_CONCURRENT_PAGES) = {gate}"
+        + (f", warm-pool target = {pool}" if pool else "  (single-slot sticky IP)")
+    )
+    print(f"  requests={args.count} arrival_rate={args.rate}/s (sim)")
+    print("  --- structure (scale-free) ---")
+    print(
+        f"  paid solves           = {metrics.solves}  (want ~pool size, not ~requests)"
+    )
+    print(f"  distinct sticky IPs   = {len(metrics.ip_hits)}")
+    print(
+        f"  busiest IP carried    = {top_hits}/{metrics.renders} renders "
+        f"({top_share:.0f}%)  (funneling; want ~even spread)"
+    )
+    print(f"  peak renders on 1 IP  = {hot_peak}  (concurrency; capped by gate)")
+    print(
+        f"  --- throughput / latency (extrapolated to live @ warm={_REAL_WARM_S}s) ---"
+    )
+    print(f"  throughput  = {thru_real * 60:.0f} SERP/min  ({thru_real:.2f}/s)")
+    print(
+        f"  latency p50 = {_pct(lat, 50) * scale:6.1f}s   p95 = {_pct(lat, 95) * scale:6.1f}s   "
+        f"p99 = {_pct(lat, 99) * scale:6.1f}s"
+    )
+    print(
+        f"  ceiling (gate/warm) = {gate / _REAL_WARM_S * 60:.0f} SERP/min per process"
+    )
+    need = 500 / (gate / _REAL_WARM_S * 60)
+    print(
+        f"  -> to sustain 500 SERP/min you need ~{need:.0f} such processes "
+        f"(or a larger gate)\n"
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

surfsense_backend/scripts/stress_google_search.py

@@ -0,0 +1,215 @@
+"""Live stress test for the Google Search scraper (REAL solves + proxy spend).
+
+Simulates a burst of concurrent "users" hitting the real ``scrape_serps``
+pipeline with a mix of single-query and multi-query requests. Unlike the offline
+sim (``scale_google_search.py``), this exercises the actual DataImpulse IPs,
+CapSolver solves, and the shared browser — so it surfaces what the sim can't:
+real warm-vs-cold latency, whether the warm-IP pool amortizes solves across real
+IPs without Google re-walling them, and browser stability under concurrency.
+
+    .venv/Scripts/python.exe scripts/stress_google_search.py --users 30 --gate 8
+
+It COSTS money (each cold IP = one CapSolver solve, bounded by the pool target)
+and proxy bandwidth. Metrics are tallied from the live perf/captcha logs.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import os
+import random
+import re
+import sys
+import time
+from pathlib import Path
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
+_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_ROOT))
+
+from dotenv import load_dotenv  # noqa: E402
+
+load_dotenv(_ROOT / ".env")
+
+# Brand/navigational terms reliably trip the reCAPTCHA wall (a solve/pool grow);
+# informational terms are the lighter, often-warm path. A realistic mix.
+_BRAND = ["notebooklm", "figma", "notion", "linear app", "vercel", "perplexity ai"]
+_INFO = [
+    "python asyncio tutorial",
+    "best mechanical keyboard 2026",
+    "kubernetes vs docker",
+    "typescript generics explained",
+    "how to make sourdough bread",
+    "rust ownership model",
+]
+
+
+class _LogTally(logging.Handler):
+    """Counts solves / renders / walls / pool-reuse from the live log stream."""
+
+    _RENDER = re.compile(r"has_results=(\w+).*from_pool=(\w+) pool=(\d+)")
+    _SOLVE = re.compile(r"\[captcha\] solve (OK|did not)")
+
+    def __init__(self) -> None:
+        super().__init__(level=logging.INFO)
+        self.solves_ok = 0
+        self.solves_fail = 0
+        self.renders_ok = 0
+        self.renders_walled = 0
+        self.reuse = 0
+        self.grow = 0
+        self.max_pool = 0
+
+    def emit(self, record: logging.LogRecord) -> None:
+        msg = record.getMessage()
+        m = self._RENDER.search(msg)
+        if m:
+            ok, from_pool, pool = m.group(1), m.group(2), int(m.group(3))
+            if ok == "True":
+                self.renders_ok += 1
+            else:
+                self.renders_walled += 1
+            if from_pool == "True":
+                self.reuse += 1
+            else:
+                self.grow += 1
+            self.max_pool = max(self.max_pool, pool)
+            return
+        s = self._SOLVE.search(msg)
+        if s:
+            if s.group(1) == "OK":
+                self.solves_ok += 1
+            else:
+                self.solves_fail += 1
+
+
+def _make_request(rng: random.Random, multi_ratio: float) -> tuple[str, int]:
+    """Build one user's request: a single term or a 2-3 term multi-query.
+    Returns (newline-joined queries, expected SERP count)."""
+    pool = _BRAND + _INFO
+    if rng.random() < multi_ratio:
+        n = rng.randint(2, 3)
+        terms = rng.sample(pool, n)
+        return "\n".join(terms), n
+    return rng.choice(pool), 1
+
+
+async def _user(uid: int, queries: str, want: int, results: list[dict]) -> None:
+    from app.proprietary.platforms.google_search import (
+        GoogleSearchScrapeInput,
+        scrape_serps,
+    )
+
+    t0 = time.perf_counter()
+    inp = GoogleSearchScrapeInput(queries=queries, countryCode="us", languageCode="en")
+    try:
+        items = await scrape_serps(inp, limit=want)
+        got = len(items)
+        err = None
+    except Exception as e:  # a user request should never crash the whole run
+        got, err = 0, repr(e)
+    results.append(
+        {
+            "uid": uid,
+            "want": want,
+            "got": got,
+            "secs": time.perf_counter() - t0,
+            "err": err,
+        }
+    )
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--users", type=int, default=30, help="concurrent requests")
+    ap.add_argument("--gate", type=int, default=8, help="MAX_CONCURRENT_PAGES for run")
+    ap.add_argument("--multi-ratio", type=float, default=0.5)
+    ap.add_argument("--ramp", type=float, default=0.3, help="secs between user starts")
+    ap.add_argument("--seed", type=int, default=7)
+    args = ap.parse_args()
+
+    # Must be set BEFORE fetch.py is imported (the gate is read at import).
+    os.environ["GOOGLE_SEARCH_MAX_CONCURRENT_PAGES"] = str(args.gate)
+
+    logging.basicConfig(level=logging.WARNING)
+    tally = _LogTally()
+    for name in (
+        "app.proprietary.platforms.google_search.fetch",
+        "app.proprietary.platforms.google_search.captcha",
+    ):
+        lg = logging.getLogger(name)
+        lg.setLevel(logging.INFO)
+        lg.addHandler(tally)
+
+    from app.proprietary.platforms.google_search.fetch import (
+        _WARM_POOL_TARGET,
+        close_sessions,
+    )
+
+    rng = random.Random(args.seed)
+    plans = [_make_request(rng, args.multi_ratio) for _ in range(args.users)]
+    total_serps = sum(w for _, w in plans)
+    n_multi = sum(1 for _, w in plans if w > 1)
+
+    print(
+        f"\n=== LIVE stress: {args.users} users "
+        f"({args.users - n_multi} single / {n_multi} multi), "
+        f"~{total_serps} SERPs, gate={args.gate}, pool_target={_WARM_POOL_TARGET} ==="
+    )
+    print("  (real solves + proxy spend; warming up...)\n")
+
+    results: list[dict] = []
+    wall0 = time.perf_counter()
+    tasks = []
+    for uid, (queries, want) in enumerate(plans):
+        tasks.append(asyncio.create_task(_user(uid, queries, want, results)))
+        await asyncio.sleep(args.ramp)  # gentle ramp, not a thundering herd
+    await asyncio.gather(*tasks)
+    wall = time.perf_counter() - wall0
+
+    await close_sessions()
+
+    lat = sorted(r["secs"] for r in results)
+    got_serps = sum(r["got"] for r in results)
+    fails = [r for r in results if r["err"]]
+    short = [r for r in results if not r["err"] and r["got"] < r["want"]]
+
+    def pct(p: float) -> float:
+        return lat[min(len(lat) - 1, int(p / 100 * len(lat)))] if lat else 0.0
+
+    print("=== results ===")
+    print(f"  wall time            = {wall:.0f}s")
+    print(f"  requests ok/failed   = {len(results) - len(fails)}/{len(fails)}")
+    print(
+        f"  SERPs got/expected   = {got_serps}/{total_serps}"
+        + (f"  ({len(short)} short)" if short else "")
+    )
+    print(f"  throughput           = {got_serps / wall * 60:.0f} SERP/min")
+    print(
+        f"  request latency p50={pct(50):.0f}s  p95={pct(95):.0f}s  "
+        f"max={lat[-1] if lat else 0:.0f}s"
+    )
+    print("  --- pipeline (from live logs) ---")
+    print(
+        f"  paid solves ok/fail  = {tally.solves_ok}/{tally.solves_fail}  "
+        f"(bounded by pool target {_WARM_POOL_TARGET})"
+    )
+    print(f"  renders ok/walled    = {tally.renders_ok}/{tally.renders_walled}")
+    print(
+        f"  pool reuse/grow      = {tally.reuse}/{tally.grow}  "
+        f"(reuse share {100 * tally.reuse / max(1, tally.reuse + tally.grow):.0f}%)"
+    )
+    print(f"  peak pool size       = {tally.max_pool}")
+    if fails:
+        print("  --- failures ---")
+        for r in fails[:5]:
+            print(f"    user{r['uid']}: {r['err']}")
+    print()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())