diff --git a/.github/proxy_traffic_summary.py b/.github/proxy_traffic_summary.py
new file mode 100644
index 0000000..bd7db83
--- /dev/null
+++ b/.github/proxy_traffic_summary.py
@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+"""Canary step: one line about a run's proxy traffic for the job summary.
+
+    python3 .github/proxy_traffic_summary.py <out> <label> [budget_mb]
+
+Reads `<out>.meta.json` and `<out>.latest_attempt.meta.json` (the newer of
+the two when both exist), appends `**<label>**: proxy traffic X.X MB of
+<budget> MB` to $GITHUB_STEP_SUMMARY, and exits 1 with an `::error::` line
+when the run stopped at its proxy budget. A missing or unreadable sidecar,
+or a run that used no local proxy, is reported and is not a failure here:
+the canary's own check step judges the run itself.
+"""
+import json
+import os
+import sys
+from datetime import datetime
+
+
+def _load(path):
+    try:
+        with open(path, encoding="utf-8") as stream:
+            data = json.load(stream)
+        return data if isinstance(data, dict) else None
+    except FileNotFoundError:
+        return None
+    except (OSError, ValueError) as exc:
+        print("::warning::%s is unreadable: %s" % (path, exc))
+        return None
+
+
+def _finished_at(meta):
+    """A sidecar's finished_at as a number: epoch seconds or an ISO-8601
+    string (repos write either), 0.0 when absent or unreadable."""
+    value = meta.get("finished_at")
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        pass
+    try:
+        return datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp()
+    except ValueError:
+        return 0.0
+
+
+def summarise(out, label, budget_mb=10.0):
+    """Return (line, whether the run stopped at its proxy budget)."""
+    metas = [m for m in (_load(out + ".meta.json"), _load(out + ".latest_attempt.meta.json")) if m]
+    meta = max(metas, key=_finished_at) if metas else None
+    budget = "%g MB" % budget_mb if budget_mb else "no"
+    if meta is None:
+        line = "**%s**: proxy traffic not recorded (no sidecar for %s)" % (label, out)
+    elif "proxy_bytes" not in meta:
+        line = "**%s**: proxy traffic not recorded (the sidecar has no proxy_bytes)" % label
+    elif meta.get("proxy_bytes") is None:
+        line = "**%s**: proxy traffic: none (no local proxy on this run)" % label
+    else:
+        line = "**%s**: proxy traffic %.1f MB of %s" % (label, (meta.get("proxy_bytes") or 0) / 1e6, budget)
+    exceeded = bool(meta and meta.get("proxy_budget_exceeded"))
+    return line, exceeded
+
+
+def main(argv):
+    if len(argv) < 3:
+        print("usage: proxy_traffic_summary.py <out> <label> [budget_mb]", file=sys.stderr)
+        return 2
+    try:
+        budget_mb = float(argv[3]) if len(argv) > 3 else 10.0
+    except ValueError:
+        budget_mb = 10.0
+    line, exceeded = summarise(argv[1], argv[2], budget_mb)
+    print(line)
+    summary = os.environ.get("GITHUB_STEP_SUMMARY")
+    if summary:
+        try:
+            with open(summary, "a", encoding="utf-8") as stream:
+                stream.write(line + "\n")
+        except OSError as exc:
+            print("::warning::could not write the job summary: %s" % exc)
+    if exceeded:
+        print("::error::stopped: proxy budget %g MB exceeded" % budget_mb)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))
diff --git a/.github/workflows/canary.yml b/.github/workflows/canary.yml
index 455f896..610a768 100644
--- a/.github/workflows/canary.yml
+++ b/.github/workflows/canary.yml
@@ -171,9 +171,17 @@ jobs:
           set +e
           python3 playwright_scraper.py \
             --url "https://www.farfetch.com/shopping/kids/girls-clothing-4/items.aspx" \
-            --pages 3 --out canary_run --dump-html canary_run.html
+            --pages 3 --out canary_run --dump-html canary_run.html --proxy-budget-mb 10
           echo "exit_code=$?" >> "$GITHUB_OUTPUT"
 
+      - name: Proxy traffic
+        # What the run sent through FARFETCH_PROXY, read from its sidecar,
+        # into the job summary. Red when the run stopped at the 10 MB budget
+        # (stop_reason proxy_budget_exceeded); a missing sidecar is reported,
+        # not failed — the next step judges the run itself.
+        if: always() && env.HAVE_PROXY == 'true'
+        run: python3 .github/proxy_traffic_summary.py canary_run production-like 10
+
       - name: Interpret the exit code
         if: env.HAVE_PROXY == 'true'
         # No --allow-block here: from a residential exit, a block is news.
@@ -192,6 +200,7 @@ jobs:
             canary_run.json
             canary_run.csv
             canary_run.meta.json
+            canary_run.latest_attempt.meta.json
             canary_run.html*
             canary_run_page*_debug.*
           if-no-files-found: ignore
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index c309b7e..48de6b8 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -251,8 +251,10 @@ jobs:
     # (which proves each requirements file is self-consistent, a thing nothing
     # checked before), and is exactly what the README tells users to do.
     #
-    # Browser BINARIES are still not installed: the offline suite imports the
-    # libraries and never launches one.
+    # Browser BINARIES: the offline suite imports the libraries and never
+    # launches one. The proxy-meter step at the end does launch a real
+    # browser against a stub proxy on 127.0.0.1 (Playwright's own Chromium,
+    # the runner's Chrome for pyppeteer and Selenium).
     strategy:
       fail-fast: false
       matrix:
@@ -327,6 +329,18 @@ jobs:
       - name: This engine's CLI answers --help
         run: .venv-${{ matrix.engine }}/bin/python ${{ matrix.engine }}_scraper.py --help
 
+      - name: Install Playwright's Chromium (playwright only)
+        if: matrix.engine == 'playwright'
+        run: .venv-playwright/bin/playwright install --with-deps chromium
+
+      - name: The proxy meter against a stub proxy, in a real browser
+        # tests/test_proxy_meter.py compares this engine's byte count with
+        # what a local stub proxy saw. PROXY_METER_REQUIRE makes a missing
+        # engine or browser a failure here instead of a skip.
+        env:
+          PROXY_METER_REQUIRE: ${{ matrix.library }}
+        run: .venv-${{ matrix.engine }}/bin/python -m unittest -v tests.test_proxy_meter
+
       - name: The console scripts install and resolve
         # A [project.scripts] entry naming a missing target installs happily
         # and fails only when a user runs it. Installing the package and
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4bf8bec..80962a4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,7 +16,12 @@ a promise that every flag and exit code is contractually frozen.
 > files and `<out>.meta.json` are left untouched. (2) Behind a local proxy,
 > the browser no longer downloads images, video/audio or web fonts.
 > `--load-media` restores the old behaviour. Nothing changes with
-> `--cdp-endpoint` or without a proxy.
+> `--cdp-endpoint` or without a proxy. (3) Behind a local proxy a run now
+> stops once about 10 MB have gone through it (`--proxy-budget-mb`, `0` =
+> unlimited): the products read so far are kept and the run exits 6 (5 with
+> none), `stop_reason: proxy_budget_exceeded`. A proxied run that writes no
+> products now writes `<out>.latest_attempt.meta.json` with its outcome and
+> byte count.
 
 ### Added
 
@@ -28,6 +33,19 @@ a promise that every flag and exit code is contractually frozen.
   (`context.route`) and pyppeteer (CDP `Fetch`), and by file extension in
   Selenium (`Network.setBlockedURLs`). `--load-media` opts out on all three.
   Playwright turns off its HTTP cache while a route is installed.
+- Proxy traffic meter (`proxy_meter.py`) in all three engines, active only
+  behind a local proxy: Playwright sums `request.sizes()`, pyppeteer and
+  Selenium read CDP `Network.loadingFinished.encodedDataLength` plus the
+  request sizes, and the pre-check's `HEAD` counts too. It is an estimate of
+  the tunnelled payload (TLS/`CONNECT` overhead not included). The sidecar
+  records `proxy_bytes`, `proxy_budget_bytes` and `proxy_budget_exceeded`, and
+  a proxied run ends with one `Proxy traffic: X.X MB of Y MB budget` line. New
+  flag `--proxy-budget-mb` (default 10) on every engine.
+- CI only: the production-like canary passes `--proxy-budget-mb 10`, writes
+  `**production-like**: proxy traffic X.X MB of 10 MB` to the job summary and
+  fails with `::error::stopped: proxy budget 10 MB exceeded` when the run hit
+  the budget. The engine jobs run the meter against a stub proxy in a real
+  browser.
 
 ### Changed
 
diff --git a/Dockerfile b/Dockerfile
index fc59420..275c5d3 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -24,7 +24,7 @@ RUN pip install --no-cache-dir -r requirements.txt -r requirements-playwright.tx
 # level, so the image died with ModuleNotFoundError on every invocation,
 # `--help` included. Nothing noticed, because CI never built the image.
 COPY arg_types.py captcha_solver.py notify.py run_state.py env_config.py fingerprint_client.py output_writer.py \
-     playwright_scraper.py product_parser.py product_detail_parser.py diff_runs.py proxy_pool.py ./
+     playwright_scraper.py product_parser.py product_detail_parser.py diff_runs.py proxy_pool.py proxy_meter.py ./
 
 ENTRYPOINT ["python3", "playwright_scraper.py"]
 CMD ["--help"]
diff --git a/README.md b/README.md
index bd0218e..a0638b5 100644
--- a/README.md
+++ b/README.md
@@ -250,6 +250,7 @@ The three browser engines share these:
 | `--cdp-endpoint` | – | Attach to a running browser instead of launching one |
 | `--proxy` | – | Proxy for a self-launched browser; ignored with `--cdp-endpoint`. Pre-checked before the browser starts ([traffic savers](#traffic-savers-pre-check-and-media-blocking)) |
 | `--load-media` | off | Load images, video and web fonts behind `--proxy` too (by default they are blocked there) |
+| `--proxy-budget-mb` | 10 | Stop once this many MB have gone through `--proxy` / `--proxy-file`; `0` = unlimited (see [Traffic budget](#traffic-budget)) |
 | `--retries` | `3` | Attempts per page load, pause doubling each time |
 | `--concurrency` | `1` | Playwright only — fetch pages through N parallel workers ([details](#concurrency)) |
 | `--twocaptcha-key` | – | 2Captcha API key (or set `TWOCAPTCHA_KEY`) |
@@ -284,8 +285,8 @@ A contract, not decoration — the harness and any pipeline can branch on these.
 | `2` | Bad usage |
 | `3` | **Blocked before parsing** — a challenge page, or an outright refusal |
 | `4` | Fetched the page fine, parsed **0 products**, and the page links to no products either |
-| `5` | **No content obtained** — navigation timeout, dead proxy, a proxy gateway that refused the pre-check with 407/402 (`stop_reason: proxy_auth_failed`), 4xx/5xx, a failed `--cdp-endpoint` connect, a Scraper API error, or a page that links to products and parsed to none (`stop_reason: parse_drift` — a parser problem, not a network one) |
-| `6` | **Partial run** — products written, but the page loop stopped early |
+| `5` | **No content obtained** — navigation timeout, dead proxy, a proxy gateway that refused the pre-check with 407/402 (`stop_reason: proxy_auth_failed`), 4xx/5xx, a failed `--cdp-endpoint` connect, a Scraper API error, or a page that links to products and parsed to none (`stop_reason: parse_drift` — a parser problem, not a network one), or the proxy budget ran out before any product was read (`stop_reason: proxy_budget_exceeded`) |
+| `6` | **Partial run** — products written, but the page loop stopped early (including at `--proxy-budget-mb`: `stop_reason: proxy_budget_exceeded`) |
 | `124` | Selenium only: chromedriver could not be **started** within the watchdog |
 
 **Exit 4 means the page loaded.** This is the one code that says something
@@ -841,6 +842,39 @@ off.
 Debug screenshots (`*_debug.png`) taken while blocking is on show the page
 without its images.
 
+#### Traffic budget
+
+Also only for a browser this repo launches behind `--proxy` / `--proxy-file`:
+every byte that browser sends or receives through the proxy is counted, and
+`--proxy-budget-mb` (default `10`; `0` = unlimited) caps a run. Once the
+running total passes the budget:
+
+- no further listing page, worker page or product page is opened, and the
+  browser aborts any request it still tries;
+- the products already read are kept, and the run ends with
+  `stop_reason: proxy_budget_exceeded` — exit 6 with products, exit 5
+  without. It never goes on silently, and a block or an empty page seen on
+  the way does not change that code.
+
+The sidecar records `proxy_bytes`, `proxy_budget_bytes` and
+`proxy_budget_exceeded` (`null` / `null` / `false` when no local proxy was
+used). A proxied run that writes no products records them in
+`<out>.latest_attempt.meta.json` beside the untouched data. A proxied run
+ends with one line such as `Proxy traffic: 3.2 MB of 10 MB budget`.
+
+The count is an estimate of the tunnelled payload — request line, headers
+and body sent, response headers and body as received — not the proxy's
+invoice: TLS records, the `CONNECT` exchange, WebSocket frames and the
+browser's own background requests are not visible to it, so the bill is a
+few per cent higher. 1 MB = 1,000,000 bytes.
+
+| Engine | Counted from |
+|---|---|
+| Playwright | `request.sizes()` of every finished or failed request |
+| pyppeteer | CDP `Network.loadingFinished.encodedDataLength` of every `http(s)` response that came over the network, plus each request's line, headers and body (`Network.requestWillBeSent`) |
+| Selenium | the same CDP events, read from Chrome's performance log |
+| pre-check | the `HEAD` itself (request + response headers) |
+
 Measured 2026-10-20 against a local stub page holding a png, a jpg, a woff2,
 an mp4 and a script. Without blocking, every engine fetched all of them. With
 blocking, Playwright and Selenium fetched only the page and the script, and
diff --git a/output_writer.py b/output_writer.py
index 9b61c7e..d48d3b1 100644
--- a/output_writer.py
+++ b/output_writer.py
@@ -12,6 +12,7 @@
 import uuid
 
 import notify
+import proxy_meter
 from dataclasses import dataclass, asdict, field
 from datetime import datetime, timezone
 from typing import Optional, List, Set
@@ -397,12 +398,15 @@ def write_failed_attempt(out_prefix: str, *, stop_reason: str,
                     products=0, mode=mode)
     meta.update(blocked=False, transport_error=transport_error,
                 data_updated=False)
+    # proxy_bytes / proxy_budget_bytes / proxy_budget_exceeded
+    meta.update(proxy_meter.sidecar())
     path = f"{out_prefix}{ATTEMPT_META_SUFFIX}"
     with _atomic(path) as f:
         json.dump(meta, f, ensure_ascii=False, indent=2)
     print(f"[+] Wrote attempt metadata -> {path} (status=failed, "
           f"stop_reason={stop_reason}); the data and {out_prefix}.meta.json "
           f"were left as they were.")
+    proxy_meter.report()
     return path
 
 
@@ -596,6 +600,11 @@ def finish_run(products: List[Product], out_prefix: str, fmt: str,
     # fix: 28 of 32 repos behaved this way. Same shape as the exit-code
     # unification this file already carries — a rule keyed on a list of
     # names has a hole for every name nobody added to it.
+    meter = proxy_meter.current()
+    if meter.exceeded:
+        # The run stopped itself at --proxy-budget-mb. That outranks whatever
+        # the pages it did fetch said: rows kept -> 6, none -> 5 (never 3/4).
+        stop_reason, blocked = proxy_meter.STOP_REASON, False
     complete = stop_reason in COMPLETE_STOP_REASONS and not pages_failed
     write_stats: dict = {}
     rc = save(products, out_prefix, fmt, allow_empty=allow_empty,
@@ -620,8 +629,17 @@ def finish_run(products: List[Product], out_prefix: str, fmt: str,
     # not overwrite a key the run itself set.
     for k, v in write_stats.items():
         meta.setdefault(k, v)
+    for k, v in proxy_meter.sidecar().items():
+        meta.setdefault(k, v)
     if wrote_output:
         write_run_meta(out_prefix, meta)
+    elif meter.active:
+        # Nothing written, but what the proxy carried is still recorded
+        # beside the untouched data: <out>.latest_attempt.meta.json.
+        attempt = dict(meta, data_updated=False)
+        with _atomic(f"{out_prefix}{ATTEMPT_META_SUFFIX}") as f:
+            json.dump(attempt, f, ensure_ascii=False, indent=2)
+    proxy_meter.report()
 
     rc = _finish_code(products, rc, blocked, stop_reason, complete,
                       out_prefix, pages_completed, pages_requested)
diff --git a/playwright_scraper.py b/playwright_scraper.py
index 865f221..d89eca7 100644
--- a/playwright_scraper.py
+++ b/playwright_scraper.py
@@ -76,6 +76,7 @@
                         ProxyPrecheckError, PROXY_PRECHECK_STOP_REASON,
                         should_block_media, playwright_media_route)
 import env_config
+import proxy_meter
 from arg_types import positive_int, nonneg_int, nonneg_float
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
@@ -286,6 +287,10 @@ def _launch_local(pw, args, pool):
         context.route("**/*", playwright_media_route)
         logger.info("Blocking images/media/fonts "
                     "(--load-media keeps them).")
+    if proxy:
+        # Counted for --proxy-budget-mb. Registered after the media route, so
+        # it is asked first and aborts every request once the budget is spent.
+        proxy_meter.current().attach_playwright(context)
     return browser, context, context.new_page()
 
 
@@ -587,6 +592,10 @@ def _fetch_one_page(session, args, pool, page_num: int, url: str) -> PageOutcome
     budget = SolveBudget()
 
     for block_attempt in range(block_retries + 1):
+        if proxy_meter.exceeded():
+            # --proxy-budget-mb is spent: no new navigation through the proxy.
+            outcome.load_failed = True
+            return outcome
         logger.info("Fetching page %d/%d: %s", page_num, args.pages, url)
         # Retry a navigation timeout rather than ending the run on it.
         # One network flap on page 12 of 50 used to break the loop, and
@@ -827,7 +836,7 @@ def worker(index: int):
                 session = _BrowserSession(pw, args, _worker_pool(pool, index)).open()
                 try:
                     first = True
-                    while not exhausted.is_set():
+                    while not exhausted.is_set() and not proxy_meter.exceeded():
                         try:
                             page_num, url = work.get_nowait()
                         except queue.Empty:
@@ -912,6 +921,8 @@ def _crawl_details(session, args, listing_rows, checkpoint):
     for i, url in enumerate(urls, start=1):
         if checkpoint.has(i):
             continue
+        if proxy_meter.exceeded():
+            break  # --proxy-budget-mb: no further product page through the proxy
         if i > 1:
             time.sleep(args.delay)
         try:
@@ -993,6 +1004,9 @@ def scrape(args) -> int:
                        "remote browser has its own exit, and layering a second "
                        "proxy on top would contradict it.")
         pool = None
+    # Meters what goes through a LOCAL proxy (none over --cdp-endpoint or
+    # without a proxy) and stops the run at --proxy-budget-mb.
+    proxy_meter.start(getattr(args, "proxy_budget_mb", None), active=bool(pool))
 
     if pool:
         # One HEAD for robots.txt through the proxy BEFORE any browser
@@ -1128,6 +1142,8 @@ def scrape(args) -> int:
                     url = (planned[0] if planned else
                            _next_url_from_page(session.page, args, 1))
                     for page_num in range(2, args.pages + 1):
+                        if proxy_meter.exceeded():
+                            break  # --proxy-budget-mb; finish_run names the stop
                         # A new exit per page is what actually spreads a run's
                         # volume, and it costs a browser relaunch: see
                         # _launch_local for why carrying the session across
@@ -1207,7 +1223,13 @@ def scrape(args) -> int:
                  if ok_pages else args.url)
 
     rows, row_type = all_products, None
-    if args.mode == "detail":
+    if args.mode == "detail" and proxy_meter.exceeded():
+        # The budget ran out on the listing: no product page is opened, and
+        # listing rows are not written as if they were size rows.
+        logger.warning("The proxy budget was spent on the listing pages — no "
+                       "product page opened (--proxy-budget-mb).")
+        rows, row_type = [], ProductVariant
+    elif args.mode == "detail":
         # A fresh session for the crawl. The listing session is closed by now
         # — and under --concurrency it was closed before the workers even
         # started — so reopening is the consistent choice rather than
@@ -1291,6 +1313,7 @@ def parse_args():
                         "(no parser reads them, and a proxy bills per byte); "
                         "never blocked with --cdp-endpoint or with no proxy.")
     p.add_argument("--force-block-media", action="store_true", help=argparse.SUPPRESS)
+    proxy_meter.add_budget_argument(p)
     p.add_argument("--proxy-file", default=None,
                    help="File with one proxy URL per line (# comments and blank "
                         "lines skipped) to rotate across. Wins over --proxy.")
diff --git a/proxy_meter.py b/proxy_meter.py
new file mode 100644
index 0000000..eb08537
--- /dev/null
+++ b/proxy_meter.py
@@ -0,0 +1,457 @@
+"""proxy_meter.py — count what a run sends through a local proxy, and stop
+at a budget.
+
+Only a LOCAL proxy is metered (`--proxy` / `--proxy-file` / the `*_PROXY`
+variable): a residential proxy bills per byte, and a run that loops or a page
+that grew can spend a lot before anyone looks. Nothing is counted or limited
+over `--cdp-endpoint` (the remote browser is billed on its own side), over the
+Scraper API, or with no proxy.
+
+What is counted, per engine — an ESTIMATE of the tunnelled payload:
+
+* Playwright: `request.sizes()` of every finished or failed request —
+  request headers + body, response headers + (encoded) body.
+* pyppeteer: CDP `Network.loadingFinished.encodedDataLength` (response
+  headers + encoded body as Chromium received them) of every response that
+  came over the network, plus each request's line, headers and post body
+  from `Network.requestWillBeSent`. Chrome's own error page and cache hits
+  are not counted.
+* Selenium: the same CDP events, read from Chrome's performance log.
+* HTTP clients (requests / curl): the request line, headers and body sent,
+  plus the response status line, headers and body as received on the wire.
+
+TLS records, the CONNECT exchange and TCP/IP framing are not visible to a
+browser or an HTTP client, so the proxy's own bill is a few per cent higher.
+WebSocket frames, and the browser's own background requests that are not
+part of any page, are not counted either. 1 MB here is 1,000,000 bytes.
+
+The budget (`--proxy-budget-mb`, default 10, 0 = unlimited): once the
+running total goes past it, the engine stops fetching new pages, any request
+the browser still tries is aborted, the rows already gathered are kept, and
+the run ends with stop_reason `proxy_budget_exceeded` — exit 6 when it has
+rows, 5 when it has none. It never silently goes on.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import atexit
+import json
+import sys
+import threading
+from typing import Any, Iterable, Optional
+from urllib.parse import urlsplit
+
+MB = 1_000_000
+DEFAULT_BUDGET_MB = 10.0
+STOP_REASON = "proxy_budget_exceeded"
+SIDECAR_KEYS = ("proxy_bytes", "proxy_budget_bytes", "proxy_budget_exceeded")
+
+
+class ProxyBudgetExceeded(RuntimeError):
+    """Raised by `ProxyMeter.check()` once the budget is spent."""
+
+
+def budget_mb(value: str) -> float:
+    """argparse type for --proxy-budget-mb: a number >= 0 (0 = unlimited)."""
+    try:
+        number = float(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError("--proxy-budget-mb must be a number of megabytes, got %r"
+                                         % value) from None
+    if number != number or number < 0 or number == float("inf"):
+        raise argparse.ArgumentTypeError("--proxy-budget-mb must be >= 0 (0 = unlimited), got %r" % value)
+    return number
+
+
+def add_budget_argument(parser, *, env_var: Optional[str] = None) -> None:
+    """--proxy-budget-mb. `env_var` only names the variable in the help."""
+    parser.add_argument(
+        "--proxy-budget-mb", type=budget_mb, default=DEFAULT_BUDGET_MB, metavar="MB",
+        help="Stop once this many megabytes have gone through --proxy / --proxy-file (default %g; 0 = "
+             "unlimited%s). The rows gathered so far are kept and the run exits 6 (5 with no rows), "
+             "stop_reason proxy_budget_exceeded. The count is an estimate of the tunnelled payload "
+             "(TLS/CONNECT overhead not included). Not applied with --cdp-endpoint or with no proxy"
+             % (DEFAULT_BUDGET_MB, "; or set %s" % env_var if env_var else ""))
+
+
+def _header_bytes(headers: Any) -> int:
+    """`Name: value\\r\\n` per header (a dict, a list of pairs, or CDP's
+    `[{name, value}]`)."""
+    if not headers:
+        return 0
+    pairs: Iterable[Any]
+    if isinstance(headers, dict):
+        pairs = headers.items()
+    else:
+        pairs = ((h.get("name", ""), h.get("value", "")) if isinstance(h, dict) else h for h in headers)
+    return sum(len(str(k)) + len(str(v)) + 4 for k, v in pairs)
+
+
+def _request_line_bytes(method: Optional[str], url: str) -> int:
+    parts = urlsplit(url or "")
+    target = (parts.path or "/") + ("?" + parts.query if parts.query else "")
+    return len("%s %s HTTP/1.1\r\n" % (method or "GET", target)) + 2  # + the blank line
+
+
+class ProxyMeter:
+    """A thread-safe running byte count with a budget."""
+
+    def __init__(self, budget_mb: float = DEFAULT_BUDGET_MB, active: bool = False):
+        self.active = bool(active)
+        self.budget_bytes = int(round(budget_mb * MB)) if budget_mb else 0
+        self.bytes = 0
+        self.exceeded = False
+        self._lock = threading.Lock()
+        self._reported = False
+        self._on_exceeded: list = []
+        self._partial: dict = {}
+
+    # ------------------------------------------------------------- counting
+    def add(self, n: Any) -> bool:
+        """Add `n` bytes. Returns True when the budget is (now) exceeded."""
+        if not self.active:
+            return False
+        try:
+            n = int(n or 0)
+        except (TypeError, ValueError):
+            return self.exceeded
+        if n <= 0:
+            return self.exceeded
+        fire = False
+        with self._lock:
+            self.bytes += n
+            if self.budget_bytes and self.bytes > self.budget_bytes and not self.exceeded:
+                self.exceeded = fire = True
+        if fire:
+            print("[!] Proxy budget of %s exceeded (%.1f MB so far): no new requests; the rows gathered so far "
+                  "are kept (stop_reason %s). Raise it with --proxy-budget-mb (0 = unlimited)."
+                  % (self.budget_text(), self.bytes / MB, STOP_REASON), file=sys.stderr, flush=True)
+            for callback in list(self._on_exceeded):
+                try:
+                    callback()
+                except Exception:  # noqa: BLE001 - blocking is best effort; the loop check is the rule
+                    pass
+        return self.exceeded
+
+    def on_exceeded(self, callback) -> None:
+        """Run `callback()` once when the budget is first exceeded (e.g. to
+        block every further request in a live browser)."""
+        if self.exceeded:
+            try:
+                callback()
+            except Exception:  # noqa: BLE001
+                pass
+        else:
+            self._on_exceeded.append(callback)
+
+    def check(self) -> None:
+        """Raise ProxyBudgetExceeded once the budget is spent."""
+        if self.exceeded:
+            raise ProxyBudgetExceeded(STOP_REASON)
+
+    # ------------------------------------------------------------- reporting
+    def budget_text(self) -> str:
+        return "%g MB" % (self.budget_bytes / MB) if self.budget_bytes else "no (unlimited)"
+
+    def summary_line(self) -> str:
+        return "Proxy traffic: %.1f MB of %s budget" % (self.bytes / MB, self.budget_text())
+
+    def sidecar(self) -> dict:
+        """The three sidecar keys. null when no local proxy was metered."""
+        return {
+            "proxy_bytes": self.bytes if self.active else None,
+            "proxy_budget_bytes": (self.budget_bytes or None) if self.active else None,
+            "proxy_budget_exceeded": bool(self.exceeded),
+        }
+
+    def report(self) -> None:
+        """Print the summary line once (only when a local proxy was metered)."""
+        if self.active and not self._reported:
+            self._reported = True
+            print(self.summary_line(), file=sys.stderr, flush=True)
+
+    # ------------------------------------------------------------- Playwright
+    @staticmethod
+    def _sizes_total(sizes: dict) -> int:
+        return sum(int(sizes.get(k) or 0) for k in (
+            "requestHeadersSize", "requestBodySize", "responseHeadersSize", "responseBodySize"))
+
+    def _playwright_fallback(self, request) -> int:
+        try:
+            body = request.post_data_buffer or b""
+        except Exception:  # noqa: BLE001
+            body = b""
+        try:
+            headers = request.headers
+        except Exception:  # noqa: BLE001
+            headers = {}
+        return _request_line_bytes(request.method, request.url) + _header_bytes(headers) + len(body)
+
+    def playwright_request_done(self, request) -> None:
+        """`requestfinished` / `requestfailed` handler (sync API)."""
+        try:
+            total = self._sizes_total(request.sizes())
+        except Exception:  # noqa: BLE001 - a failed request may have no sizes: count what was sent
+            total = self._playwright_fallback(request)
+        self.add(total)
+
+    async def playwright_request_done_async(self, request) -> None:
+        """`requestfinished` / `requestfailed` handler (async API)."""
+        try:
+            total = self._sizes_total(await request.sizes())
+        except Exception:  # noqa: BLE001
+            total = self._playwright_fallback(request)
+        self.add(total)
+
+    def attach_playwright(self, target) -> None:
+        """Count every request of a sync-API BrowserContext (or Page)."""
+        if not self.active:
+            return
+        target.on("requestfinished", self.playwright_request_done)
+        target.on("requestfailed", self.playwright_request_done)
+        if self.budget_bytes:
+            target.route("**/*", self.playwright_budget_route)
+
+    async def attach_playwright_async(self, target) -> None:
+        """Count every request of an async-API BrowserContext (or Page)."""
+        if not self.active:
+            return
+        target.on("requestfinished", self.playwright_request_done_async)
+        target.on("requestfailed", self.playwright_request_done_async)
+        if self.budget_bytes:
+            await target.route("**/*", self.playwright_budget_route_async)
+
+    def playwright_budget_route(self, route) -> None:
+        """Abort everything once the budget is spent; otherwise hand the
+        request to the next route (media blocking) or the network."""
+        if self.exceeded:
+            route.abort("blockedbyclient")
+        else:
+            route.fallback()
+
+    async def playwright_budget_route_async(self, route) -> None:
+        if self.exceeded:
+            await route.abort("blockedbyclient")
+        else:
+            await route.fallback()
+
+    # ------------------------------------------------------------- CDP events
+    def feed_cdp(self, method: str, params: dict) -> None:
+        """One CDP Network event (pyppeteer session or Selenium's log)."""
+        if not self.active or not isinstance(params, dict):
+            return
+        if method == "Network.requestWillBeSent":
+            redirect = params.get("redirectResponse") or {}
+            if redirect:
+                # The 3xx that led here was received and has no loadingFinished.
+                self.add(redirect.get("encodedDataLength") or 0)
+            request = params.get("request") or {}
+            url = request.get("url") or ""
+            if not url.startswith(("http://", "https://")):
+                return  # data:, blob:, chrome:// (the new-tab page), ... never cross the proxy
+            body = request.get("postData") or ""
+            self.add(_request_line_bytes(request.get("method"), url) + _header_bytes(request.get("headers"))
+                     + len(body.encode("utf-8", "replace") if isinstance(body, str) else body))
+        elif method == "Network.responseReceived":
+            # Only a response that came over the network is counted at
+            # loadingFinished: Chrome's own error page ("This site can't be
+            # reached", ~250 KB) is reported as a finished load of the failed
+            # URL with no responseReceived, the new-tab page's chrome://
+            # scripts (~2.8 MB, seen with Selenium) never leave the browser,
+            # and a cache hit crossed nothing.
+            response = params.get("response") or {}
+            local = str(response.get("url") or "").split(":", 1)[0] not in ("", "http", "https")
+            if not local and not (response.get("fromDiskCache") or response.get("fromServiceWorker")
+                                  or response.get("fromPrefetchCache")):
+                with self._lock:
+                    self._partial.setdefault(params.get("requestId"), 0)
+        elif method == "Network.dataReceived":
+            # Only kept for a load that then fails: loadingFinished reports
+            # the whole length itself.
+            request_id = params.get("requestId")
+            with self._lock:
+                if request_id in self._partial:
+                    self._partial[request_id] += int(params.get("encodedDataLength") or 0)
+        elif method == "Network.loadingFinished":
+            with self._lock:
+                responded = self._partial.pop(params.get("requestId"), None) is not None
+            if responded:
+                self.add(params.get("encodedDataLength") or 0)
+        elif method == "Network.loadingFailed":
+            with self._lock:
+                partial = self._partial.pop(params.get("requestId"), 0)
+            self.add(partial)
+
+    def feed_selenium_log(self, entries: Iterable[dict]) -> None:
+        """Entries of `driver.get_log("performance")`."""
+        for entry in entries or ():
+            try:
+                message = json.loads(entry["message"])["message"]
+            except Exception:  # noqa: BLE001
+                continue
+            self.feed_cdp(message.get("method", ""), message.get("params") or {})
+
+    async def attach_pyppeteer(self, page) -> None:
+        """Count a pyppeteer page's traffic over its own CDP session; once
+        the budget is spent, block every further URL in that page."""
+        if not self.active:
+            return
+        client = await page.target.createCDPSession()
+        client.on("Network.requestWillBeSent", lambda params: self.feed_cdp("Network.requestWillBeSent", params))
+        for event in ("Network.responseReceived", "Network.dataReceived", "Network.loadingFinished",
+                      "Network.loadingFailed"):
+            client.on(event, lambda params, _event=event: self.feed_cdp(_event, params))
+        await client.send("Network.enable", {})
+        if self.budget_bytes:
+            # The budget can trip on another thread (a concurrent worker) that
+            # has no event loop: hand the block to this page's own loop.
+            loop = asyncio.get_event_loop()
+
+            async def _send_block():
+                try:
+                    await client.send("Network.setBlockedURLs", {"urls": ["*"]})
+                except Exception:  # noqa: BLE001 - the loop check still stops the run
+                    pass
+
+            def _block():
+                loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_send_block()))
+            self.on_exceeded(_block)
+
+    def attach_selenium(self, driver, *, keep_entries: bool = True) -> None:
+        """Tee a Chrome WebDriver's performance log into the meter.
+
+        The driver must have been created with `selenium_logging_prefs`
+        applied. `driver.get_log("performance")` is wrapped so the engine's
+        own readers still get every entry; `poll_selenium` reads the log for
+        the meter alone and, with `keep_entries`, hands what it read to the
+        engine's next `get_log` call."""
+        if not self.active or getattr(driver, "_proxy_meter_attached", False):
+            return
+        original = driver.get_log
+        held: list = []
+
+        def get_log(kind):
+            entries = original(kind)
+            if kind != "performance":
+                return entries
+            self.feed_selenium_log(entries)
+            out = held + list(entries)
+            held.clear()
+            self._selenium_block_if_needed(driver)
+            return out
+
+        def poll():
+            try:
+                entries = original("performance")
+            except Exception:  # noqa: BLE001 - e.g. a driver without the log enabled
+                return
+            self.feed_selenium_log(entries)
+            if keep_entries:
+                held.extend(entries)
+            self._selenium_block_if_needed(driver)
+
+        driver.get_log = get_log
+        driver._proxy_meter_poll = poll
+        driver._proxy_meter_attached = True
+
+    def poll_selenium(self, driver) -> None:
+        """Bring the count up to date (call after each navigation/scroll)."""
+        poll = getattr(driver, "_proxy_meter_poll", None)
+        if poll is not None:
+            poll()
+
+    def _selenium_block_if_needed(self, driver) -> None:
+        if self.exceeded and self.budget_bytes and not getattr(driver, "_proxy_meter_blocked", False):
+            driver._proxy_meter_blocked = True
+            try:
+                driver.execute_cdp_cmd("Network.enable", {})
+                driver.execute_cdp_cmd("Network.setBlockedURLs", {"urls": ["*"]})
+            except Exception:  # noqa: BLE001 - the loop check still stops the run
+                pass
+
+    # ------------------------------------------------------------- HTTP clients
+    def count_http(self, *, method: str, url: str, request_headers: Any = None, request_body: Any = None,
+                   status_line: str = "", response_headers: Any = None, response_body_bytes: int = 0) -> None:
+        body = request_body or b""
+        if isinstance(body, str):
+            body = body.encode("utf-8", "replace")
+        sent = _request_line_bytes(method, url) + _header_bytes(request_headers) + len(body)
+        received = (len(status_line) + 2 if status_line else len("HTTP/1.1 200 OK\r\n")) \
+            + _header_bytes(response_headers) + 2 + int(response_body_bytes or 0)
+        self.add(sent + received)
+
+    def count_requests_response(self, response) -> None:
+        """A `requests.Response` (its redirect history too). Reads the body
+        if it was not read yet; the size is what came off the wire (before
+        decompression) when urllib3 can tell, else the decoded length."""
+        if not self.active:
+            return
+        for r in list(getattr(response, "history", None) or []) + [response]:
+            try:
+                content = r.content
+            except Exception:  # noqa: BLE001
+                content = b""
+            wire = 0
+            try:
+                wire = int(r.raw.tell()) if r.raw is not None else 0
+            except Exception:  # noqa: BLE001
+                wire = 0
+            prepared = r.request
+            self.count_http(
+                method=getattr(prepared, "method", "GET"), url=getattr(prepared, "url", "") or r.url,
+                request_headers=dict(getattr(prepared, "headers", {}) or {}),
+                request_body=getattr(prepared, "body", None),
+                status_line="HTTP/1.1 %s %s" % (r.status_code, r.reason or ""),
+                response_headers=dict(r.headers or {}), response_body_bytes=wire or len(content or b""))
+
+    def requests_hook(self, response, *args, **kwargs):
+        """`hooks={"response": meter.requests_hook}` for requests."""
+        self.count_requests_response(response)
+        return response
+
+
+# --------------------------------------------------------------------------- #
+# The run's meter
+# --------------------------------------------------------------------------- #
+_current = ProxyMeter(active=False)
+
+
+def start(budget_mb: Optional[float], *, active: bool) -> ProxyMeter:
+    """Install the run's meter. `active` = traffic goes through a LOCAL proxy."""
+    global _current
+    _current = ProxyMeter(DEFAULT_BUDGET_MB if budget_mb is None else budget_mb, active=active)
+    return _current
+
+
+def current() -> ProxyMeter:
+    return _current
+
+
+def sidecar() -> dict:
+    return _current.sidecar()
+
+
+def report() -> None:
+    _current.report()
+
+
+def exceeded() -> bool:
+    return _current.exceeded
+
+
+def selenium_logging_prefs(options) -> None:
+    """Turn Chrome's performance log on in a Selenium `Options`, keeping any
+    other logging preference already set."""
+    try:
+        prefs = dict(options.to_capabilities().get("goog:loggingPrefs") or {})
+    except Exception:  # noqa: BLE001
+        prefs = {}
+    prefs["performance"] = "ALL"
+    options.set_capability("goog:loggingPrefs", prefs)
+
+
+@atexit.register
+def _report_at_exit() -> None:  # a crash path still prints the line once
+    _current.report()
diff --git a/proxy_pool.py b/proxy_pool.py
index c2422d7..764be61 100644
--- a/proxy_pool.py
+++ b/proxy_pool.py
@@ -437,8 +437,10 @@ def precheck_proxy(proxy_url: Optional[str], probe_url: Optional[str],
     except ImportError:
         return None
     try:
+        import proxy_meter  # the HEAD goes through the proxy too: it counts toward --proxy-budget-mb
         response = requests.head(probe_url, proxies={"http": proxy_url, "https": proxy_url},
-                                 timeout=timeout, allow_redirects=False)
+                                 timeout=timeout, allow_redirects=False,
+                                 hooks={"response": proxy_meter.current().requests_hook})
     except requests.exceptions.ProxyError as exc:
         # An https target: the gateway answers the CONNECT, and urllib3
         # reports its status line as "Tunnel connection failed: 407 ...".
diff --git a/puppeteer_scraper.py b/puppeteer_scraper.py
index 7331039..8452a88 100644
--- a/puppeteer_scraper.py
+++ b/puppeteer_scraper.py
@@ -46,6 +46,7 @@
                         ProxyPrecheckError, PROXY_PRECHECK_STOP_REASON,
                         should_block_media, pyppeteer_block_media)
 import env_config
+import proxy_meter
 from arg_types import positive_int, nonneg_float
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
@@ -214,6 +215,8 @@ async def scrape(args) -> int:
     pages_completed = 0
     final_url = args.url
 
+    # Meters what goes through a LOCAL proxy and stops at --proxy-budget-mb.
+    proxy_meter.start(getattr(args, "proxy_budget_mb", None), active=bool(args.proxy) and not args.cdp_endpoint)
     if args.proxy and not args.cdp_endpoint:
         # One HEAD for robots.txt through the proxy BEFORE the browser
         # starts — see playwright_scraper.py. Only a gateway 407/402 stops.
@@ -264,6 +267,8 @@ async def scrape(args) -> int:
             await page.authenticate({"username": creds.username or "",
                                      "password": creds.password or ""})
             logger.info("Proxy credentials sent over CDP, not via the command line.")
+    if args.proxy and not args.cdp_endpoint:
+        await proxy_meter.current().attach_pyppeteer(page)  # --proxy-budget-mb
     if should_block_media(args, proxy_in_use=bool(args.proxy),
                           over_cdp=bool(args.cdp_endpoint)):
         # CDP Fetch with resource-type patterns: only images, media and fonts
@@ -300,11 +305,16 @@ async def scrape(args) -> int:
     try:
         url = args.url
         for page_num in range(1, args.pages + 1):
+            if proxy_meter.exceeded():
+                break  # --proxy-budget-mb is spent; finish_run names the stop
             logger.info("Fetching page %d/%d: %s", page_num, args.pages, url)
             # Retry a navigation failure rather than ending the run on it —
             # see playwright_scraper.py for why.
             http_status = None
             for attempt in range(1, args.retries + 1):
+                if proxy_meter.exceeded():
+                    load_failed = True
+                    break
                 try:
                     # Keep the document's Response: its status is the one
                     # signal a refusing edge always gives, and without it an
@@ -475,6 +485,7 @@ def parse_args():
                         "(no parser reads them, and a proxy bills per byte); "
                         "never blocked with --cdp-endpoint or with no proxy.")
     p.add_argument("--force-block-media", action="store_true", help=argparse.SUPPRESS)
+    proxy_meter.add_budget_argument(p)
     p.add_argument("--webhook", default=None, metavar="URL",
                    help="POST the run summary (the same fields as the "
                         ".meta.json sidecar, plus the exit code) to this "
diff --git a/pyproject.toml b/pyproject.toml
index 1ecfc98..619bd8f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -80,6 +80,7 @@ py-modules = [
     "playwright_scraper",
     "product_detail_parser",
     "product_parser",
+    "proxy_meter",
     "proxy_pool",
     "puppeteer_scraper",
     "run_state",
@@ -155,6 +156,7 @@ files = [
     "product_detail_parser.py",
     "product_parser.py",
     "proxy_pool.py",
+    "proxy_meter.py",
     "env_config.py",
     "notify.py",
     "run_state.py",
diff --git a/selenium_scraper.py b/selenium_scraper.py
index c34c2d6..3eb7209 100644
--- a/selenium_scraper.py
+++ b/selenium_scraper.py
@@ -73,6 +73,7 @@
                         ProxyPrecheckError, PROXY_PRECHECK_STOP_REASON,
                         should_block_media, selenium_block_media)
 import env_config
+import proxy_meter
 from arg_types import positive_int, nonneg_float
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
@@ -586,6 +587,9 @@ def build_driver(args) -> webdriver.Chrome:
         parsed = urlparse(args.proxy)
         options.add_argument(f"--proxy-server={_bare_proxy(args.proxy)}")
         logger.info("Using proxy %s", mask_proxy(args.proxy))
+        # Chrome's performance log carries the CDP Network events the
+        # --proxy-budget-mb meter counts.
+        proxy_meter.selenium_logging_prefs(options)
         if parsed.username or parsed.password:
             logger.warning(
                 "The credentials in --proxy were dropped, not sent: Chromium's "
@@ -649,6 +653,8 @@ def build_driver(args) -> webdriver.Chrome:
     driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
         "source": "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
     })
+    if args.proxy:
+        proxy_meter.current().attach_selenium(driver, keep_entries=False)
     driver.media_blocked = False
     if should_block_media(args, proxy_in_use=bool(args.proxy)):
         # By URL pattern: Selenium has no request events, only
@@ -750,6 +756,8 @@ def scrape(args) -> int:
     stop_reason = "completed"
     pages_completed = 0
     final_url = args.url
+    # Meters what goes through a LOCAL proxy and stops at --proxy-budget-mb.
+    proxy_meter.start(getattr(args, "proxy_budget_mb", None), active=bool(args.proxy) and not args.cdp_endpoint)
     if args.proxy and not args.cdp_endpoint:
         # One HEAD for robots.txt BEFORE chromedriver starts — see
         # playwright_scraper.py. Sent exactly as this browser will use the
@@ -790,12 +798,17 @@ def scrape(args) -> int:
     try:
         url = args.url
         for page_num in range(1, args.pages + 1):
+            if proxy_meter.exceeded():
+                break  # --proxy-budget-mb is spent; finish_run names the stop
             logger.info("Fetching page %d/%d: %s", page_num, args.pages, url)
             # Retry a navigation failure rather than ending the run on it —
             # see playwright_scraper.py for why.
             http_status = None
             for attempt in range(1, args.retries + 1):
                 load_failed = False
+                if proxy_meter.exceeded():
+                    load_failed = True
+                    break
                 try:
                     driver.get(url)
                     wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
@@ -826,6 +839,7 @@ def scrape(args) -> int:
                                 "parsing whatever loaded (may be a bot-check/consent page).")
 
             html = driver.page_source
+            proxy_meter.current().poll_selenium(driver)  # --proxy-budget-mb
 
             vendor = detect_bot_challenge(html)
             if vendor:
@@ -922,6 +936,7 @@ def scrape(args) -> int:
         if args.cdp_endpoint:
             driver.close()  # detach this tab only, leave the remote browser running
         else:
+            proxy_meter.current().poll_selenium(driver)  # what the last page still loaded
             driver.quit()
 
     return finish_run(all_products, args.out, args.format, args.allow_empty,
@@ -955,6 +970,7 @@ def parse_args():
                         "(no parser reads them, and a proxy bills per byte); "
                         "never blocked with --cdp-endpoint or with no proxy.")
     p.add_argument("--force-block-media", action="store_true", help=argparse.SUPPRESS)
+    proxy_meter.add_budget_argument(p)
     p.add_argument("--chrome-binary", default=None,
                    help="Path to the Chrome/Chromium binary to launch locally. Only "
                         "needed when it isn't in one of chromedriver's default "
diff --git a/smoke_test.py b/smoke_test.py
index 40301b3..76aeaa4 100644
--- a/smoke_test.py
+++ b/smoke_test.py
@@ -3695,7 +3695,9 @@ def _verdict(code, allow_block):
         ok &= check("the canary never puts the proxy credential in argv — it "
                     "goes through the environment env_config.py already "
                     "reads",
-                    not any("--proxy" in ln for ln in _cwf_code)
+                    # --proxy / --proxy-file carry the credential;
+                    # --proxy-budget-mb is a number and may appear.
+                    not any(re.search(r"--proxy(?:[ =]|-file)", ln) for ln in _cwf_code)
                     and "FARFETCH_PROXY: ${{ secrets.FARFETCH_PROXY }}" in _cwf)
         # Reversed on 2026-10-15. It used to SKIP green without the secret,
         # and the no-proxy job was refused on every run, so for a month the
@@ -4534,6 +4536,496 @@ def _engine_args(mod, argv):
         os.environ.update(saved_env)
 
 
+def check_proxy_traffic_meter(ok: bool) -> bool:
+    """proxy traffic meter and --proxy-budget-mb (proxy_meter.py; its byte
+    counts against a stub proxy are in tests/test_proxy_meter.py)"""
+    import asyncio as _asyncio
+    import proxy_meter as _pm
+    import proxy_pool as _pp
+    import output_writer as _ow
+
+    ns = argparse.Namespace
+    listing = "https://www.farfetch.com/shopping/kids/girls-clothing-4/items.aspx"
+    proxy_argv = ["--proxy", "http://" + _UP + "127.0.0.1:1"]  # pre-check: refused, no verdict
+
+    def _page_html(n, empty=False):
+        if empty:
+            return "<html><body><div class='grid'></div></body></html>"
+        return (SAMPLE_LISTING_HTML.replace("29998189", "%d1" % (n + 1000))
+                .replace("item-99999", "item-%d2" % (n + 1000)))
+
+    class _Meter:
+        """What each fake navigation costs through the proxy."""
+        cost = 0
+        gotos = []
+        empty = False
+
+        @classmethod
+        def nav(cls, url):
+            cls.gotos.append(url)
+            _pm.current().add(cls.cost)
+            n = 1
+            if "page=" in url:
+                n = int(url.split("page=")[1].split("&")[0])
+            return _page_html(n, cls.empty)
+
+    # ---- fakes for the three engines --------------------------------------
+    class _PWErr(Exception):
+        pass
+
+    class _PWTimeout(_PWErr):
+        pass
+
+    class _PWPage:
+        def __init__(self):
+            self.url, self._html = "about:blank", ""
+
+        def goto(self, url, **k):
+            self._html, self.url = _Meter.nav(url), url
+            return ns(status=200)
+
+        def content(self):
+            return self._html
+
+        def wait_for_function(self, *a, **k):
+            return None
+
+        def wait_for_timeout(self, *a):
+            return None
+
+        def query_selector(self, *a):
+            return None
+
+        def screenshot(self, **k):
+            return None
+
+        def close(self):
+            return None
+
+    class _PWCtx:
+        def __init__(self):
+            self.routes, self.events = [], []
+
+        def route(self, pattern, handler):
+            self.routes.append(handler)
+
+        def on(self, name, fn):
+            self.events.append(name)
+
+        def add_init_script(self, *a):
+            pass
+
+        def new_page(self):
+            return _PWPage()
+
+    class _PWBrowser:
+        version = "124.0.6367.60"
+
+        def __init__(self):
+            self.contexts = []
+
+        def new_context(self, **k):
+            self.contexts.append(_PWCtx())
+            return self.contexts[-1]
+
+        def close(self):
+            pass
+
+    @contextlib.contextmanager
+    def _sync_playwright():
+        yield ns(chromium=ns(launch=lambda **k: _PWBrowser()))
+
+    class _SelErr(Exception):
+        pass
+
+    class _Wait:
+        def __init__(self, *a, **k):
+            pass
+
+        def until(self, fn):
+            return True
+
+    _sel_fakes = {
+        "selenium": {}, "selenium.webdriver": {"Chrome": None},
+        "selenium.webdriver.chrome": {},
+        "selenium.webdriver.chrome.options": {"Options": None},
+        "selenium.webdriver.chrome.service": {"Service": None},
+        "selenium.webdriver.common": {}, "selenium.webdriver.common.by": {"By": ns(CSS_SELECTOR="css")},
+        "selenium.common": {},
+        "selenium.common.exceptions": {"TimeoutException": _SelErr,
+                                       "NoSuchElementException": _SelErr,
+                                       "WebDriverException": _SelErr},
+        "selenium.webdriver.support": {},
+        "selenium.webdriver.support.ui": {"WebDriverWait": _Wait},
+    }
+
+    class _SelDriver:
+        def __init__(self):
+            self.page_source, self.current_url = "", "about:blank"
+
+        def get(self, url):
+            self.page_source, self.current_url = _Meter.nav(url), url
+
+        def execute_script(self, js, *a):
+            return "complete" if "readyState" in js else 200
+
+        def find_element(self, *a):
+            raise _SelErr()
+
+        def save_screenshot(self, *a):
+            return True
+
+        def quit(self):
+            pass
+
+        def close(self):
+            pass
+
+    class _PPage:
+        def __init__(self):
+            self.url, self._html = "about:blank", ""
+            self.client = ns(sent=[], on=lambda *a: None)
+            self.target = ns(createCDPSession=self._session)
+
+        async def _session(self):
+            async def send(cmd, params=None):
+                self.client.sent.append(cmd)
+            self.client.send = send
+            return self.client
+
+        async def authenticate(self, *a):
+            return None
+
+        async def setUserAgent(self, *a):
+            return None
+
+        async def goto(self, url, *a, **k):
+            self._html, self.url = _Meter.nav(url), url
+            return ns(status=200)
+
+        async def waitForFunction(self, *a, **k):
+            return None
+
+        async def content(self):
+            return self._html
+
+        async def evaluate(self, *a, **k):
+            return None
+
+        async def screenshot(self, *a, **k):
+            return None
+
+        async def close(self):
+            return None
+
+    class _PBrowser:
+        def __init__(self):
+            self.page = _PPage()
+
+        async def pages(self):
+            return [self.page]
+
+        async def version(self):
+            return "HeadlessChrome/124.0.0.0"
+
+        async def close(self):
+            return None
+
+        async def disconnect(self):
+            return None
+
+    async def _pyp_launch(**k):
+        return _PBrowser()
+
+    engines = {
+        "playwright": _load_engine_with_fake_driver("playwright_scraper.py", {
+            "playwright": {},
+            "playwright.sync_api": {"sync_playwright": _sync_playwright,
+                                    "Error": _PWErr, "TimeoutError": _PWTimeout}}),
+        "pyppeteer": _load_engine_with_fake_driver("puppeteer_scraper.py", {
+            "pyppeteer": {"launch": _pyp_launch, "connect": _pyp_launch}}),
+        "selenium": _load_engine_with_fake_driver("selenium_scraper.py", _sel_fakes),
+    }
+    for mod in engines.values():
+        mod.handle_captcha_if_present = (lambda *a, **k: None) if mod is not engines["pyppeteer"] \
+            else _async_none
+    engines["selenium"].build_driver_with_timeout = lambda args: _SelDriver()
+
+    def _run(name, argv, *, cost, empty=False):
+        """(rc, text, data files, meta, attempt) for one scrape() through the fakes."""
+        mod = engines[name]
+        _Meter.cost, _Meter.gotos, _Meter.empty = cost, [], empty
+        with tempfile.TemporaryDirectory() as d:
+            out = os.path.join(d, "run")
+            with open(out + ".json", "w", encoding="utf-8") as f:
+                f.write("[]")  # a previous run's data
+            args = _engine_args(mod, ["--url", listing, "--out", out, "--format", "json",
+                                      "--delay", "0", "--retry-delay", "0"] + argv)
+            try:
+                with _captured_output() as buf:
+                    rc = mod.scrape(args)
+                    if name == "pyppeteer":
+                        rc = _asyncio.run(rc)
+            finally:
+                _pm.start(None, active=False)
+
+            def read(suffix):
+                path = out + suffix
+                return json.load(open(path, encoding="utf-8")) if os.path.exists(path) else None
+            return rc, buf.getvalue(), read(".json"), read(".meta.json"), \
+                read(".latest_attempt.meta.json")
+
+    for name in engines:
+        # 1,000-byte budget, 600 bytes per navigation: over it on page 2.
+        rc, text, data, meta, attempt = _run(name, ["--pages", "4", "--proxy-budget-mb", "0.001"]
+                                             + proxy_argv, cost=600)
+        ok &= check(f"proxy budget ({name}): over the budget on page 2 of 4 -> exit 6 (got {rc}), "
+                    f"no page 3 fetched ({len(_Meter.gotos)} navigations)",
+                    rc == EXIT_PARTIAL and len(_Meter.gotos) == 2)
+        ok &= check(f"proxy budget ({name}): ...the rows of pages 1-2 are kept",
+                    data is not None and len(data) == 4)
+        ok &= check(f"proxy budget ({name}): ...sidecar says partial / proxy_budget_exceeded "
+                    f"with the byte count",
+                    meta is not None and meta["status"] == "partial"
+                    and meta["stop_reason"] == "proxy_budget_exceeded"
+                    and (meta["proxy_bytes"], meta["proxy_budget_bytes"],
+                         meta["proxy_budget_exceeded"]) == (1200, 1000, True))
+        ok &= check(f"proxy budget ({name}): ...and one 'Proxy traffic' line is printed",
+                    text.count("Proxy traffic: 0.0 MB of 0.001 MB budget") == 1)
+
+        rc, text, data, meta, attempt = _run(name, ["--proxy-budget-mb", "0.001"] + proxy_argv,
+                                             cost=5000, empty=True)
+        ok &= check(f"proxy budget ({name}): spent with no row read -> exit 5, not 4 (got {rc})",
+                    rc == EXIT_FETCH_FAILED)
+        ok &= check(f"proxy budget ({name}): ...the previous data stays, no .meta.json, and "
+                    f"<out>.latest_attempt.meta.json records the stop and the bytes",
+                    data == [] and meta is None and attempt is not None
+                    and attempt["stop_reason"] == "proxy_budget_exceeded"
+                    and attempt["proxy_bytes"] == 5000 and attempt["data_updated"] is False)
+
+        rc, text, data, meta, attempt = _run(name, ["--proxy-budget-mb", "0.001"], cost=10 ** 9)
+        ok &= check(f"proxy budget ({name}): no proxy -> nothing metered or stopped "
+                    f"(exit {rc}, proxy_bytes null, no 'Proxy traffic' line)",
+                    rc == 0 and meta["proxy_bytes"] is None
+                    and meta["proxy_budget_exceeded"] is False and "Proxy traffic" not in text)
+        rc, text, data, meta, attempt = _run(name, ["--proxy-budget-mb", "0"] + proxy_argv,
+                                             cost=10 ** 9)
+        ok &= check(f"proxy budget ({name}): --proxy-budget-mb 0 is unlimited (exit {rc})",
+                    rc == 0 and meta["proxy_budget_bytes"] is None
+                    and meta["proxy_bytes"] == 10 ** 9 and meta["proxy_budget_exceeded"] is False)
+        rc, text, data, meta, attempt = _run(name, proxy_argv, cost=1000)
+        ok &= check(f"proxy budget ({name}): under the default 10 MB the run is complete and "
+                    f"records its bytes",
+                    rc == 0 and meta["proxy_bytes"] == 1000
+                    and meta["proxy_budget_bytes"] == 10 ** 7
+                    and "Proxy traffic: 0.0 MB of 10 MB budget" in text)
+
+    rc, text, data, meta, attempt = _run("playwright", ["--pages", "6", "--concurrency", "2",
+                                                        "--proxy-budget-mb", "0.001"]
+                                         + proxy_argv, cost=600)
+    ok &= check(f"proxy budget (playwright --concurrency 2): workers take no page once the "
+                f"budget is spent ({len(_Meter.gotos)} of 6 navigations, exit {rc})",
+                rc == EXIT_PARTIAL and len(_Meter.gotos) <= 3
+                and meta["stop_reason"] == "proxy_budget_exceeded")
+    rc, text, data, meta, attempt = _run("playwright", ["--mode", "detail",
+                                                        "--proxy-budget-mb", "0.001"]
+                                         + proxy_argv, cost=5000)
+    ok &= check(f"proxy budget (playwright --mode detail): spent on the listing -> no product "
+                f"page opened, listing rows not written as size rows (exit {rc})",
+                rc == EXIT_FETCH_FAILED and len(_Meter.gotos) == 1
+                and attempt["stop_reason"] == "proxy_budget_exceeded")
+
+    # ---- finish_run: the budget stop outranks every other outcome ---------
+    for rows, blocked, reason, want in (([Product(sku="1")], True, "blocked_akamai", EXIT_PARTIAL),
+                                        ([], True, "blocked_akamai", EXIT_FETCH_FAILED),
+                                        ([Product(sku="1")], False, "completed", EXIT_PARTIAL),
+                                        ([], False, "completed", EXIT_FETCH_FAILED)):
+        _pm.start(0.001, active=True)
+        _pm.current().add(5000)
+        try:
+            with tempfile.TemporaryDirectory() as d, _captured_output():
+                out = os.path.join(d, "o")
+                rc = finish_run(rows, out, "json", False, blocked=blocked, stop_reason=reason,
+                                pages_requested=2, pages_completed=1, start_url=listing,
+                                final_url=listing)
+                path = out + (".meta.json" if rows else ".latest_attempt.meta.json")
+                meta = json.load(open(path, encoding="utf-8"))
+        finally:
+            _pm.start(None, active=False)
+        ok &= check(f"proxy budget: finish_run with {len(rows)} row(s), blocked={blocked}, "
+                    f"{reason} -> exit {want} (got {rc}), stop_reason proxy_budget_exceeded",
+                    rc == want and meta["stop_reason"] == "proxy_budget_exceeded"
+                    and meta["proxy_budget_exceeded"] is True)
+
+    # ---- the flag on every CLI --------------------------------------------
+    for name, mod in engines.items():
+        args = _engine_args(mod, ["--url", listing])
+        ok &= check(f"proxy budget ({name}): --proxy-budget-mb defaults to 10",
+                    args.proxy_budget_mb == 10)
+        for bad in ("-1", "lots"):
+            with _captured_output() as buf:
+                try:
+                    _engine_args(mod, ["--url", listing, "--proxy-budget-mb", bad])
+                    code = 0
+                except SystemExit as e:
+                    code = e.code
+            ok &= check(f"proxy budget ({name}): --proxy-budget-mb {bad} is a usage error "
+                        f"naming the flag", code == 2 and "--proxy-budget-mb" in buf.getvalue())
+
+    # ---- wiring: only a LOCAL proxy is metered ----------------------------
+    pwm = engines["playwright"]
+    for argv, active, want_events, want_routes in ((proxy_argv, True, True, 2),
+                                                   (proxy_argv + ["--load-media"], True, True, 1),
+                                                   (proxy_argv + ["--proxy-budget-mb", "0"], True, True, 1),
+                                                   ([], False, False, 0)):
+        _args = _engine_args(pwm, ["--url", listing] + argv)
+        _pm.start(_args.proxy_budget_mb, active=active)
+        browser = _PWBrowser()
+        try:
+            with _captured_output():
+                pwm._launch_local(ns(chromium=ns(launch=lambda _b=browser, **k: _b)), _args,
+                                  _pp.from_args(_args))
+        finally:
+            meter = _pm.current()
+            _pm.start(None, active=False)
+        ctx = browser.contexts[0]
+        ok &= check(f"proxy meter (playwright {_label(argv)}): "
+                    f"{'counts requestfinished/requestfailed' if want_events else 'not attached'}, "
+                    f"{want_routes} route(s)",
+                    (sorted(set(ctx.events)) == ["requestfailed", "requestfinished"]) is want_events
+                    and len(ctx.routes) == want_routes
+                    and (want_routes != 2 or ctx.routes[-1] == meter.playwright_budget_route))
+
+    class _Opts:
+        def __init__(self):
+            self.arguments, self.caps = [], {}
+            self.binary_location = None
+
+        def add_argument(self, a):
+            self.arguments.append(a)
+
+        def set_capability(self, k, v):
+            self.caps[k] = v
+
+        def to_capabilities(self):
+            return dict(self.caps)
+
+    class _Drv:
+        def __init__(self, **k):
+            self.capabilities, self.options = {}, k.get("options")
+
+        def execute_cdp_cmd(self, *a):
+            pass
+
+        def get_log(self, kind):
+            return []
+
+    import shutil as _shutil
+    selm = _load_engine_with_fake_driver("selenium_scraper.py", _sel_fakes)
+    selm.Options, selm.Service = _Opts, (lambda **k: ns(**k))
+    selm.webdriver = ns(Chrome=_Drv)
+    for argv, want in ((proxy_argv, True), ([], False)):
+        _args = _engine_args(selm, ["--url", listing, "--chromedriver", sys.executable] + argv)
+        _pm.start(10, active=want)
+        try:
+            with _captured_output():
+                drv = selm.build_driver(_args)
+        finally:
+            _pm.start(None, active=False)
+        for a in drv.options.arguments:
+            if a.startswith("--user-data-dir="):
+                _shutil.rmtree(a.split("=", 1)[1], ignore_errors=True)
+        ok &= check(f"proxy meter (selenium {_label(argv)}): "
+                    f"{'performance log on and teed into the meter' if want else 'not attached'}",
+                    (drv.options.caps.get("goog:loggingPrefs", {}).get("performance") == "ALL"
+                     and hasattr(drv, "_proxy_meter_poll")) is want)
+
+    for argv, want in ((proxy_argv, True), ([], False),
+                       (proxy_argv + ["--cdp-endpoint", "ws://127.0.0.1:1"], False)):
+        ppm = engines["pyppeteer"]
+        browser = _PBrowser()
+
+        async def _give(_b=browser, **k):
+            return _b
+        ppm.launch, ppm.connect = _give, _give
+        ppm._connect_remote = lambda args, _b=browser: _give(_b)
+        _Meter.cost, _Meter.gotos, _Meter.empty = 0, [], False
+        with tempfile.TemporaryDirectory() as d, _captured_output():
+            _args = _engine_args(ppm, ["--url", listing, "--out", os.path.join(d, "o"),
+                                       "--delay", "0"] + argv)
+            try:
+                _asyncio.run(ppm.scrape(_args))
+            finally:
+                _pm.start(None, active=False)
+        ok &= check(f"proxy meter (pyppeteer {_label(argv)}): "
+                    f"{'a counting CDP session (Network.enable)' if want else 'no meter session'}",
+                    ("Network.enable" in browser.page.client.sent) is want)
+
+    # ---- the canary --------------------------------------------------------
+    wf = open(os.path.join(_repo_root, ".github", "workflows", "canary.yml"), encoding="utf-8").read()
+    jobs = wf.split("\n  production-like:", 1)
+    proxied = jobs[1] if len(jobs) == 2 else ""
+    ok &= check("canary: the proxied run passes --proxy-budget-mb 10",
+                "--out canary_run --dump-html canary_run.html --proxy-budget-mb 10" in proxied)
+    step = proxied.split("- name: Proxy traffic", 1)
+    body = step[1].split("- name:", 1)[0] if len(step) == 2 else ""
+    ok &= check("canary: an always-run 'Proxy traffic' step follows the proxied run and calls "
+                ".github/proxy_traffic_summary.py for its output",
+                proxied.index("--proxy-budget-mb 10") < proxied.find("- name: Proxy traffic")
+                and "if: always()" in body
+                and "python3 .github/proxy_traffic_summary.py canary_run production-like 10" in body)
+    script = os.path.join(_repo_root, ".github", "proxy_traffic_summary.py")
+    with tempfile.TemporaryDirectory() as d:
+        summary = os.path.join(d, "summary.md")
+        env = dict(os.environ, GITHUB_STEP_SUMMARY=summary)
+
+        def _summarise(meta=None, attempt=None):
+            for suffix, data in ((".meta.json", meta), (".latest_attempt.meta.json", attempt)):
+                path = os.path.join(d, "canary_run" + suffix)
+                if os.path.exists(path):
+                    os.remove(path)
+                if data is not None:
+                    with open(path, "w", encoding="utf-8") as f:
+                        json.dump(data, f)
+            r = subprocess.run([sys.executable, script, "canary_run", "production-like", "10"],
+                               cwd=d, env=env, capture_output=True, text=True, timeout=60)
+            last = open(summary, encoding="utf-8").read().splitlines()[-1] \
+                if os.path.exists(summary) else ""
+            return r.returncode, r.stdout, last
+        rc, out, last = _summarise({"proxy_bytes": 4_321_000, "proxy_budget_exceeded": False,
+                                    "finished_at": "2026-10-21T00:00:02"})
+        ok &= check("canary summary: '**production-like**: proxy traffic 4.3 MB of 10 MB', green",
+                    rc == 0 and last == "**production-like**: proxy traffic 4.3 MB of 10 MB")
+        rc, out, last = _summarise(None, {"proxy_bytes": 10_200_000, "proxy_budget_exceeded": True,
+                                          "finished_at": "2026-10-21T00:00:03"})
+        ok &= check("canary summary: a run stopped at the budget is red with "
+                    "'::error::stopped: proxy budget 10 MB exceeded'",
+                    rc == 1 and "::error::stopped: proxy budget 10 MB exceeded" in out
+                    and last == "**production-like**: proxy traffic 10.2 MB of 10 MB")
+        rc, out, last = _summarise()
+        ok &= check("canary summary: a missing sidecar is reported, not a crash",
+                    rc == 0 and "not recorded" in last)
+        # This repo writes finished_at as ISO-8601: the newer of the two
+        # sidecars must be chosen by time, also beside one with no timestamp.
+        rc, out, last = _summarise({"proxy_bytes": 1, "proxy_budget_exceeded": False,
+                                    "finished_at": "2026-10-21T00:00:01+00:00"},
+                                   {"proxy_bytes": 10_200_000, "proxy_budget_exceeded": True,
+                                    "finished_at": "2026-10-21T00:00:05+00:00"})
+        ok &= check("canary summary: of two sidecars with ISO finished_at the newer one is "
+                    "the run's (got rc %s)" % rc, rc == 1)
+        rc, out, last = _summarise({"proxy_bytes": 10_200_000, "proxy_budget_exceeded": True,
+                                    "finished_at": "2026-10-21T00:00:09+00:00"},
+                                   {"proxy_bytes": 1, "proxy_budget_exceeded": False})
+        ok &= check("canary summary: an ISO timestamp beside a sidecar without one does not "
+                    "crash the step (got rc %s)" % rc,
+                    rc == 1 and "::error::stopped: proxy budget 10 MB exceeded" in out)
+    return ok
+
+
+async def _async_none(*a, **k):
+    return None
+
+
 def check_proxy_traffic_savers(ok: bool) -> bool:
     """proxy pre-check before a browser starts, and media blocking"""
     import asyncio as _asyncio
@@ -4867,6 +5359,12 @@ def __init__(self):
         def add_argument(self, a):
             self.arguments.append(a)
 
+        def set_capability(self, name, value):  # the meter's performance log
+            self.capabilities = {name: value}
+
+        def to_capabilities(self):
+            return dict(getattr(self, "capabilities", {}))
+
     class _SelDriver:
         def __init__(self, **k):
             self.capabilities = {}
@@ -5139,6 +5637,7 @@ def main() -> int:
     ok = check_the_dockerfile_s_copy_list_vs_the_entrypoint_s_i(ok)
     ok = check_audit_2026_10_14_contract_fixes(ok)
     ok = check_proxy_traffic_savers(ok)
+    ok = check_proxy_traffic_meter(ok)
 
     # The report below is only worth something if it reflects what the
     # sections recorded: an engine whose driver library is absent must appear
diff --git a/tests/test_proxy_meter.py b/tests/test_proxy_meter.py
new file mode 100644
index 0000000..6aa4560
--- /dev/null
+++ b/tests/test_proxy_meter.py
@@ -0,0 +1,521 @@
+"""proxy_meter: the byte count against a stub proxy that serves known sizes,
+the budget, and every engine's way of counting.
+
+The stub proxy IS the site: it answers absolute-URI requests for
+http://meter.test/... itself and counts every byte that crosses its sockets,
+which is the ground truth the meter is compared with. Nothing leaves this
+machine.
+
+The real-browser tests run when that engine and a Chrome/Chromium binary are
+present; with PROXY_METER_REQUIRE=<playwright|pyppeteer|selenium> (CI's
+engine jobs) a missing engine or browser is a failure, not a skip.
+"""
+import argparse
+import asyncio
+import json
+import os
+import shutil
+import socket
+import sys
+import threading
+import time
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import proxy_meter  # noqa: E402
+
+REQUIRE = os.environ.get("PROXY_METER_REQUIRE", "")
+SITE = "http://meter.test/"
+PAGES = {
+    "/": ("text/html", b"<html><head><link rel='stylesheet' href='/style.css'><script src='/app.js'></script>"
+          b"</head><body>" + b"<p>row</p>" * 2000
+          + b"<script>fetch('/data.json').then(function (r) { return r.text(); })"
+            b".then(function (t) { document.title = 'done ' + t.length; });</script></body></html>"),
+    "/app.js": ("application/javascript", b"var a = 1;" * 5000),
+    "/style.css": ("text/css", b"p { color: #123456; }" * 2000),
+    "/data.json": ("application/json", b'{"k": "' + b"v" * 30000 + b'"}'),
+}
+PAGE_BYTES = sum(len(body) for _, body in PAGES.values())  # ~152 KB of bodies
+
+
+class StubProxy:
+    """A forward proxy that serves PAGES itself and counts bytes per path."""
+
+    def __init__(self):
+        self.sock = socket.socket()
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.sock.bind(("127.0.0.1", 0))
+        self.sock.listen(64)
+        self.port = self.sock.getsockname()[1]
+        self.url = "http://127.0.0.1:%d" % self.port
+        self.per_path = {}
+        self.lock = threading.Lock()
+        threading.Thread(target=self._serve, daemon=True).start()
+
+    def site_bytes(self):
+        """Bytes in + out for the stub site's own paths (a browser's own
+        background CONNECTs to its vendor are not the page's traffic)."""
+        with self.lock:
+            return sum(v for k, v in self.per_path.items() if k in PAGES or k.startswith("/"))
+
+    def _serve(self):
+        while True:
+            try:
+                conn, _ = self.sock.accept()
+            except OSError:
+                return
+            threading.Thread(target=self._handle, args=(conn,), daemon=True).start()
+
+    def _handle(self, conn):
+        buf = b""
+        try:
+            while True:
+                while b"\r\n\r\n" not in buf:
+                    chunk = conn.recv(65536)
+                    if not chunk:
+                        return
+                    buf += chunk
+                head, buf = buf.split(b"\r\n\r\n", 1)
+                lines = head.split(b"\r\n")
+                method, target = lines[0].decode("latin-1").split(" ")[:2]
+                length = 0
+                for line in lines[1:]:
+                    if line.lower().startswith(b"content-length:"):
+                        length = int(line.split(b":", 1)[1])
+                while len(buf) < length:
+                    chunk = conn.recv(65536)
+                    if not chunk:
+                        return
+                    buf += chunk
+                received = len(head) + 4 + length
+                buf = buf[length:]
+                if method == "CONNECT":
+                    path = target
+                    body, status, ctype = b"", "403 Forbidden", "text/plain"
+                else:
+                    path = ("/" + target.split("://", 1)[1].split("/", 1)[1]) if "://" in target else target
+                    path = path.split("?", 1)[0]
+                    ctype, body = PAGES.get(path, ("text/plain", b"not found"))
+                    status = "200 OK" if path in PAGES else "404 Not Found"
+                response = ("HTTP/1.1 %s\r\nContent-Type: %s\r\nContent-Length: %d\r\nCache-Control: no-store\r\n"
+                            "Connection: %s\r\n\r\n" % (status, ctype, len(body),
+                                                        "close" if method == "CONNECT" else "keep-alive")).encode()
+                if method != "HEAD":
+                    response += body
+                conn.sendall(response)
+                with self.lock:
+                    self.per_path[path] = self.per_path.get(path, 0) + received + len(response)
+                if method == "CONNECT":
+                    return
+        except OSError:
+            return
+        finally:
+            conn.close()
+
+    def close(self):
+        try:
+            self.sock.close()
+        except OSError:
+            pass
+
+
+def chrome_binary():
+    for candidate in (os.environ.get("CHROME_BIN"), "/usr/bin/chromium", "/usr/bin/chromium-browser",
+                      "/usr/bin/google-chrome", "/usr/bin/google-chrome-stable"):
+        if candidate and Path(candidate).exists():
+            return candidate
+    return shutil.which("chromium") or shutil.which("google-chrome")
+
+
+def unavailable(test, engine, why):
+    if REQUIRE == engine:
+        test.fail("PROXY_METER_REQUIRE=%s but %s" % (engine, why))
+    test.skipTest(why)
+
+
+class Budget(unittest.TestCase):
+    def test_flag_default_and_validation(self):
+        parser = argparse.ArgumentParser()
+        proxy_meter.add_budget_argument(parser)
+        self.assertEqual(parser.parse_args([]).proxy_budget_mb, 10)
+        self.assertEqual(parser.parse_args(["--proxy-budget-mb", "0"]).proxy_budget_mb, 0)
+        self.assertEqual(parser.parse_args(["--proxy-budget-mb", "2.5"]).proxy_budget_mb, 2.5)
+        for bad in ("-1", "x", "nan", "inf"):
+            with self.assertRaises(SystemExit):
+                parser.parse_args(["--proxy-budget-mb", bad])
+
+    def test_running_total_budget_and_callback(self):
+        meter = proxy_meter.ProxyMeter(1, active=True)
+        fired = []
+        meter.on_exceeded(lambda: fired.append(1))
+        self.assertFalse(meter.add(600_000))
+        meter.check()
+        self.assertFalse(meter.add(400_000), "exactly the budget is not over it")
+        self.assertTrue(meter.add(1))
+        meter.add(5)
+        self.assertEqual(fired, [1], "the callback fires once")
+        with self.assertRaises(proxy_meter.ProxyBudgetExceeded):
+            meter.check()
+        self.assertEqual(meter.sidecar(), {"proxy_bytes": 1_000_006, "proxy_budget_bytes": 1_000_000,
+                                           "proxy_budget_exceeded": True})
+        self.assertEqual(meter.summary_line(), "Proxy traffic: 1.0 MB of 1 MB budget")
+
+    def test_zero_is_unlimited(self):
+        meter = proxy_meter.ProxyMeter(0, active=True)
+        meter.add(10 ** 12)
+        self.assertFalse(meter.exceeded)
+        self.assertEqual(meter.sidecar()["proxy_budget_bytes"], None)
+
+    def test_inactive_meter_counts_nothing(self):
+        meter = proxy_meter.ProxyMeter(0.001, active=False)
+        meter.add(10 ** 9)
+        self.assertEqual(meter.sidecar(), {"proxy_bytes": None, "proxy_budget_bytes": None,
+                                           "proxy_budget_exceeded": False})
+        self.assertFalse(meter.exceeded)
+
+    def test_start_replaces_the_run_meter(self):
+        meter = proxy_meter.start(3, active=True)
+        self.assertIs(proxy_meter.current(), meter)
+        self.assertEqual(meter.budget_bytes, 3_000_000)
+        proxy_meter.start(None, active=False)
+        self.assertEqual(proxy_meter.current().budget_bytes, 10_000_000)
+
+
+class HttpClient(unittest.TestCase):
+    def test_requests_through_the_stub_proxy_matches_its_count(self):
+        import requests
+        proxy = StubProxy()
+        try:
+            meter = proxy_meter.ProxyMeter(0, active=True)
+            session = requests.Session()
+            session.hooks["response"].append(meter.requests_hook)
+            for path in PAGES:
+                response = session.get(SITE + path.lstrip("/"), proxies={"http": proxy.url}, timeout=10)
+                self.assertEqual(response.status_code, 200)
+            session.post(SITE + "data.json", data=b"x" * 5000, proxies={"http": proxy.url}, timeout=10)
+            time.sleep(0.2)
+            truth = proxy.site_bytes()
+            self.assertGreater(truth, PAGE_BYTES)
+            self.assertLess(abs(meter.bytes - truth) / truth, 0.02, (meter.bytes, truth))
+        finally:
+            proxy.close()
+
+
+class CdpEvents(unittest.TestCase):
+    def test_loading_finished_and_request_sizes(self):
+        meter = proxy_meter.ProxyMeter(0, active=True)
+        meter.feed_cdp("Network.requestWillBeSent", {"requestId": "1", "request": {
+            "method": "POST", "url": "http://x.test/a?b=1", "headers": {"A": "bc"}, "postData": "12345"}})
+        sent = len("POST /a?b=1 HTTP/1.1\r\n") + 2 + len("A") + len("bc") + 4 + 5
+        self.assertEqual(meter.bytes, sent)
+        meter.feed_cdp("Network.responseReceived", {"requestId": "1", "response": {"status": 200}})
+        meter.feed_cdp("Network.dataReceived", {"requestId": "1", "encodedDataLength": 400})
+        meter.feed_cdp("Network.loadingFinished", {"requestId": "1", "encodedDataLength": 1000})
+        self.assertEqual(meter.bytes, sent + 1000, "loadingFinished carries the whole length")
+        meter.feed_cdp("Network.responseReceived", {"requestId": "2", "response": {"status": 200}})
+        meter.feed_cdp("Network.dataReceived", {"requestId": "2", "encodedDataLength": 300})
+        meter.feed_cdp("Network.loadingFailed", {"requestId": "2"})
+        self.assertEqual(meter.bytes, sent + 1300, "a failed load counts what arrived")
+        # Chrome's error page for a failed navigation: a finished "load" of
+        # ~250 KB with no response, generated locally (seen through a refusing proxy).
+        meter.feed_cdp("Network.loadingFailed", {"requestId": "e", "errorText": "net::ERR_TUNNEL_CONNECTION_FAILED"})
+        meter.feed_cdp("Network.dataReceived", {"requestId": "e", "encodedDataLength": 0})
+        meter.feed_cdp("Network.loadingFinished", {"requestId": "e", "encodedDataLength": 250609})
+        meter.feed_cdp("Network.responseReceived", {"requestId": "c", "response": {"fromDiskCache": True}})
+        meter.feed_cdp("Network.loadingFinished", {"requestId": "c", "encodedDataLength": 900})
+        meter.feed_cdp("Network.requestWillBeSent", {"requestId": "n", "request": {
+            "method": "GET", "url": "chrome://new-tab-page/x.js", "headers": {"A": "b"}}})
+        meter.feed_cdp("Network.responseReceived", {"requestId": "n",
+                                                    "response": {"url": "chrome://new-tab-page/x.js"}})
+        meter.feed_cdp("Network.loadingFinished", {"requestId": "n", "encodedDataLength": 6575})
+        self.assertEqual(meter.bytes, sent + 1300,
+                         "an error page, a cache hit or a chrome:// page crossed no proxy")
+        before = meter.bytes
+        meter.feed_cdp("Network.requestWillBeSent", {"requestId": "3", "request": {"method": "GET", "url": "data:x"}})
+        self.assertEqual(meter.bytes, before, "a data: URL never crosses the proxy")
+        meter.feed_cdp("Network.requestWillBeSent", {"requestId": "4", "redirectResponse": {"encodedDataLength": 77},
+                                                     "request": {"method": "GET", "url": "http://x.test/"}})
+        self.assertEqual(meter.bytes, before + 77 + len("GET / HTTP/1.1\r\n") + 2)
+
+    def test_selenium_log_is_teed_not_stolen(self):
+        def entry(method, **params):
+            return {"message": json.dumps({"message": {"method": method, "params": params}})}
+
+        class Driver:
+            def __init__(self):
+                self.log = [entry("Network.responseReceived", requestId="a", response={}),
+                            entry("Network.loadingFinished", requestId="a", encodedDataLength=600)]
+                self.sent = []
+
+            def get_log(self, kind):
+                out, self.log = self.log, []
+                return out
+
+            def execute_cdp_cmd(self, cmd, params):
+                self.sent.append((cmd, params))
+
+        meter = proxy_meter.ProxyMeter(0.001, active=True)  # 1,000 bytes
+        driver = Driver()
+        meter.attach_selenium(driver, keep_entries=True)
+        meter.poll_selenium(driver)
+        self.assertEqual(meter.bytes, 600)
+        driver.log += [entry("Network.responseReceived", requestId="b", response={}),
+                       entry("Network.loadingFinished", requestId="b", encodedDataLength=500)]
+        got = driver.get_log("performance")
+        self.assertEqual(len(got), 4, "the engine still reads what the meter polled")
+        self.assertEqual(meter.bytes, 1100)
+        self.assertTrue(meter.exceeded)
+        self.assertIn(("Network.setBlockedURLs", {"urls": ["*"]}), driver.sent)
+        self.assertEqual(driver.get_log("performance"), [])
+        inactive = proxy_meter.ProxyMeter(1, active=False)
+        other = Driver()
+        inactive.attach_selenium(other)
+        self.assertFalse(hasattr(other, "_proxy_meter_poll"), "no proxy: the driver is left alone")
+
+    def test_pyppeteer_session_and_block(self):
+        class Client:
+            def __init__(self):
+                self.handlers, self.sent = {}, []
+
+            def on(self, name, fn):
+                self.handlers[name] = fn
+
+            async def send(self, cmd, params=None):
+                self.sent.append((cmd, params))
+
+        client = Client()
+
+        async def create():
+            return client
+
+        page = argparse.Namespace(target=argparse.Namespace(createCDPSession=create))
+        meter = proxy_meter.ProxyMeter(0.001, active=True)
+
+        async def go():
+            await meter.attach_pyppeteer(page)
+            client.handlers["Network.responseReceived"]({"requestId": "a", "response": {}})
+            client.handlers["Network.loadingFinished"]({"requestId": "a", "encodedDataLength": 2000})
+            await asyncio.sleep(0)
+            await asyncio.sleep(0)
+
+        asyncio.run(go())
+        self.assertIn(("Network.enable", {}), client.sent)
+        self.assertEqual(meter.bytes, 2000)
+        self.assertIn(("Network.setBlockedURLs", {"urls": ["*"]}), client.sent)
+
+    def test_pyppeteer_block_from_a_worker_thread(self):
+        """The budget can trip on a thread with no event loop (a concurrent
+        worker): the block is still sent, on the page's own loop."""
+        sent = []
+
+        class Client:
+            def on(self, name, fn):
+                pass
+
+            async def send(self, cmd, params=None):
+                sent.append(cmd)
+
+        async def create():
+            return Client()
+
+        page = argparse.Namespace(target=argparse.Namespace(createCDPSession=create))
+        meter = proxy_meter.ProxyMeter(0.001, active=True)
+
+        async def go():
+            await meter.attach_pyppeteer(page)
+            worker = threading.Thread(target=meter.add, args=(5000,))
+            worker.start()
+            worker.join()
+            for _ in range(5):
+                await asyncio.sleep(0)
+
+        asyncio.run(go())
+        self.assertTrue(meter.exceeded)
+        self.assertIn("Network.setBlockedURLs", sent)
+
+
+class PlaywrightHooks(unittest.TestCase):
+    def test_sizes_and_budget_route(self):
+        class Request:
+            method, url, headers, post_data_buffer = "GET", "http://x.test/", {"a": "b"}, None
+
+            def sizes(self):
+                return {"requestHeadersSize": 100, "requestBodySize": 0,
+                        "responseHeadersSize": 200, "responseBodySize": 5000}
+
+        class Failed(Request):
+            def sizes(self):
+                raise RuntimeError("no sizes for a failed request")
+
+        class Route:
+            def __init__(self):
+                self.done = []
+
+            def abort(self, reason=None):
+                self.done.append("abort")
+
+            def fallback(self):
+                self.done.append("fallback")
+
+        meter = proxy_meter.ProxyMeter(0.006, active=True)
+        route = Route()
+        meter.playwright_budget_route(route)
+        meter.playwright_request_done(Request())
+        self.assertEqual(meter.bytes, 5300)
+        meter.playwright_request_done(Failed())
+        self.assertEqual(meter.bytes, 5300 + len("GET / HTTP/1.1\r\n") + 2 + 1 + 1 + 4)
+        meter.playwright_request_done(Request())
+        self.assertTrue(meter.exceeded)
+        meter.playwright_budget_route(route)
+        self.assertEqual(route.done, ["fallback", "abort"])
+
+
+def _assert_close(test, meter, proxy, tolerance=0.10):
+    time.sleep(0.3)
+    truth = proxy.site_bytes()
+    test.assertGreater(truth, PAGE_BYTES, proxy.per_path)
+    test.assertLess(abs(meter.bytes - truth) / truth, tolerance, (meter.bytes, truth, proxy.per_path))
+
+
+class RealBrowsers(unittest.TestCase):
+    """The engines' own byte reports against the stub proxy's count."""
+
+    def test_playwright(self):
+        try:
+            from playwright.async_api import async_playwright
+        except ImportError:
+            return unavailable(self, "playwright", "playwright is not installed")
+        proxy = StubProxy()
+        meter = proxy_meter.ProxyMeter(0, active=True)
+
+        async def go():
+            async with async_playwright() as pw:
+                try:
+                    browser = await pw.chromium.launch(headless=True)
+                except Exception as exc:  # noqa: BLE001
+                    return str(exc).splitlines()[0]
+                context = await browser.new_context(proxy={"server": proxy.url})
+                await meter.attach_playwright_async(context)
+                page = await context.new_page()
+                await page.goto(SITE)
+                await page.wait_for_function("document.title.indexOf('done') === 0", timeout=10000)
+                await page.wait_for_timeout(500)
+                await context.close()
+                await browser.close()
+            return None
+
+        try:
+            failed = asyncio.run(go())
+        finally:
+            proxy.close()
+        if failed:
+            return unavailable(self, "playwright", "Chromium did not launch: %s" % failed)
+        _assert_close(self, meter, proxy)
+
+    def test_playwright_budget_aborts_further_requests(self):
+        try:
+            from playwright.async_api import async_playwright
+        except ImportError:
+            return unavailable(self, "playwright", "playwright is not installed")
+        proxy = StubProxy()
+        meter = proxy_meter.ProxyMeter(0.05, active=True)  # 50 KB: spent by the first page
+        errors = []
+
+        async def go():
+            async with async_playwright() as pw:
+                try:
+                    browser = await pw.chromium.launch(headless=True)
+                except Exception as exc:  # noqa: BLE001
+                    return str(exc).splitlines()[0]
+                context = await browser.new_context(proxy={"server": proxy.url})
+                await meter.attach_playwright_async(context)
+                page = await context.new_page()
+                await page.goto(SITE)
+                await page.wait_for_timeout(1500)
+                try:
+                    await page.goto(SITE + "?second")
+                except Exception as exc:  # noqa: BLE001
+                    errors.append(str(exc))
+                await context.close()
+                await browser.close()
+            return None
+
+        try:
+            failed = asyncio.run(go())
+        finally:
+            proxy.close()
+        if failed:
+            return unavailable(self, "playwright", "Chromium did not launch: %s" % failed)
+        self.assertTrue(meter.exceeded)
+        self.assertTrue(errors and "ERR_BLOCKED_BY_CLIENT" in errors[0], errors)
+        self.assertEqual(sum(1 for k in proxy.per_path if k == "/"), 1)
+
+    def test_pyppeteer(self):
+        try:
+            from pyppeteer import launch
+        except ImportError:
+            return unavailable(self, "pyppeteer", "pyppeteer is not installed")
+        binary = chrome_binary()
+        if not binary:
+            return unavailable(self, "pyppeteer", "no Chrome/Chromium binary (set CHROME_BIN)")
+        proxy = StubProxy()
+        meter = proxy_meter.ProxyMeter(0, active=True)
+
+        async def go():
+            browser = await launch(headless=True, executablePath=binary, handleSIGINT=False, handleSIGTERM=False,
+                                   handleSIGHUP=False, args=["--no-sandbox", "--proxy-server=" + proxy.url])
+            try:
+                page = await browser.newPage()
+                await meter.attach_pyppeteer(page)
+                await page.goto(SITE)
+                await page.waitForFunction("document.title.indexOf('done') === 0", {"timeout": 10000})
+                await asyncio.sleep(0.5)
+            finally:
+                await browser.close()
+
+        try:
+            asyncio.run(go())
+        finally:
+            proxy.close()
+        _assert_close(self, meter, proxy)
+
+    def test_selenium(self):
+        try:
+            from selenium import webdriver
+            from selenium.webdriver.chrome.options import Options
+            from selenium.webdriver.chrome.service import Service
+        except ImportError:
+            return unavailable(self, "selenium", "selenium is not installed")
+        binary = chrome_binary()
+        if not binary:
+            return unavailable(self, "selenium", "no Chrome/Chromium binary (set CHROME_BIN)")
+        proxy = StubProxy()
+        meter = proxy_meter.ProxyMeter(0, active=True)
+        options = Options()
+        for arg in ("--headless=new", "--no-sandbox", "--disable-dev-shm-usage", "--proxy-server=" + proxy.url):
+            options.add_argument(arg)
+        options.binary_location = binary
+        proxy_meter.selenium_logging_prefs(options)
+        driver_path = os.environ.get("CHROMEDRIVER") or shutil.which("chromedriver")
+        try:
+            driver = webdriver.Chrome(service=Service(driver_path) if driver_path else Service(), options=options)
+        except Exception as exc:  # noqa: BLE001
+            proxy.close()
+            return unavailable(self, "selenium", "chromedriver did not start: %s" % str(exc).splitlines()[0])
+        try:
+            meter.attach_selenium(driver, keep_entries=False)
+            driver.get(SITE)
+            deadline = time.time() + 10
+            while not driver.title.startswith("done") and time.time() < deadline:
+                time.sleep(0.2)
+            time.sleep(0.5)
+            meter.poll_selenium(driver)
+        finally:
+            driver.quit()
+            proxy.close()
+        _assert_close(self, meter, proxy)
+
+
+if __name__ == "__main__":
+    unittest.main()