From dc8cd6cf7b433cb54bba21a10854badc1f165d45 Mon Sep 17 00:00:00 2001
From: jehrr <79864894+jehrr@users.noreply.github.com>
Date: Wed, 21 Oct 2026 11:23:54 +0000
Subject: [PATCH] Count proxy traffic per run and stop at --proxy-budget-mb
 (default 10 MB)

A local browser run through --proxy / --proxy-file now counts the bytes
it sends through the proxy (Playwright: request.sizes(); Puppeteer and
Selenium: CDP Network.loadingFinished.encodedDataLength plus request
sizes; Selenium turns Chrome's performance log on for this behind a
proxy). Once the total passes --proxy-budget-mb no further URL or post is
fetched, the browser aborts what it still tries, the rows read so far are
kept, and the run exits 6 (5 with no rows) with stop_reason
proxy_budget_exceeded. The sidecar records proxy_bytes,
proxy_budget_bytes and proxy_budget_exceeded, and the run ends with one
"Proxy traffic: X.X MB of Y MB budget" line. Nothing is metered over
--cdp-endpoint, --scraper-api or with no proxy.

The canary passes --proxy-budget-mb 10, writes its traffic to the job
summary and fails when the budget was hit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CKsd7q25N19WzUZjrAdzx8
---
 .github/proxy_traffic_summary.py |  71 +++++
 .github/workflows/canary.yml     |   8 +-
 .github/workflows/tests.yml      |   4 +
 CHANGELOG.md                     |  20 ++
 Dockerfile                       |   2 +-
 README.md                        |  41 ++-
 output_writer.py                 |  47 +++
 page_flow.py                     |  19 +-
 playwright_scraper.py            |   6 +
 proxy_meter.py                   | 442 ++++++++++++++++++++++++++++
 proxy_pool.py                    |   4 +-
 puppeteer_scraper.py             |   4 +
 pyproject.toml                   |   1 +
 selenium_scraper.py              |   9 +
 smoke_test.py                    | 306 +++++++++++++++++++
 tests/test_proxy_meter.py        | 485 +++++++++++++++++++++++++++++++
 16 files changed, 1458 insertions(+), 11 deletions(-)
 create mode 100644 .github/proxy_traffic_summary.py
 create mode 100644 proxy_meter.py
 create mode 100644 tests/test_proxy_meter.py

diff --git a/.github/proxy_traffic_summary.py b/.github/proxy_traffic_summary.py
new file mode 100644
index 0000000..20add55
--- /dev/null
+++ b/.github/proxy_traffic_summary.py
@@ -0,0 +1,71 @@
+#!/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
+
+
+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 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=lambda m: m.get("finished_at") or 0) 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 5112320..2cc8693 100644
--- a/.github/workflows/canary.yml
+++ b/.github/workflows/canary.yml
@@ -49,9 +49,15 @@ jobs:
         run: |
           set +e
           python3 playwright_scraper.py --url "$CANARY_ACCOUNT" --posts 3 \
-            --format json --out canary.json --dump-html
+            --format json --out canary.json --dump-html --proxy-budget-mb 10
           echo "canary_exit=$?" >> "$GITHUB_ENV"
 
+      # What the run sent through INSTAGRAM_PROXY, read from its sidecar; red
+      # when it stopped at the 10 MB budget (stop_reason proxy_budget_exceeded).
+      - name: Proxy traffic
+        if: always()
+        run: python3 .github/proxy_traffic_summary.py canary.json canary 10
+
       - name: Require a complete run with full rows
         run: |
           python3 -c "
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 9f4ca95..2d6de7c 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -123,6 +123,10 @@ jobs:
         run: |
           source .venv-${{ matrix.engine }}/bin/activate
           python3 smoke_test.py
+          # The proxy meter against a stub proxy in a real browser: a missing
+          # browser fails this job instead of skipping.
+          PROXY_METER_REQUIRE=${{ matrix.engine == 'puppeteer' && 'pyppeteer' || matrix.engine }} \
+            python3 -m unittest -v tests.test_proxy_meter
           python3 -c "
           import ${{ matrix.engine }}_scraper as m
           assert getattr(m, {'playwright': 'async_playwright', 'selenium': 'webdriver', 'puppeteer': 'pyppeteer_launch'}['${{ matrix.engine }}']) is not None, \
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b9a01c7..04b7212 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -24,10 +24,30 @@ that changes a default is called out at the top of its entry.
 >   blocks them by URL extension. `--load-media` restores the old behaviour.
 >   Nothing changes with `--cdp-endpoint`, `--scraper-api` or with no proxy.
 >   `<out>.meta.json` records `media_blocked`.
+> * Behind a local proxy, a run now stops once about 10 MB have gone through
+>   it (`--proxy-budget-mb`, `0` = unlimited): the rows read so far are kept
+>   and the run exits 6 (5 with none), `stop_reason: proxy_budget_exceeded`.
+>   A proxied run that collects nothing now writes
+>   `<out>.latest_attempt.meta.json` with its outcome and byte count. Not
+>   applied with `--cdp-endpoint`, `--scraper-api` or without a proxy.
 
 ### Added
 
 - `--load-media` in all three engines.
+- Proxy traffic meter (`proxy_meter.py`) in all three browser engines, active
+  only behind a local proxy: Playwright sums `request.sizes()`, Puppeteer
+  and Selenium read CDP `Network.loadingFinished.encodedDataLength` plus the
+  request sizes (Selenium turns Chrome's performance log on for this). 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).
+  The proxy pre-check's own `HEAD` is counted too.
+- CI only: the canary passes `--proxy-budget-mb 10`, writes `**canary**:
+  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 f483150..546bea9 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -19,7 +19,7 @@ RUN pip install --no-cache-dir -r requirements-playwright.txt \
 # installed in this image, so their own try/except ImportError guard is
 # exactly what fires, the same as smoke_test.py's no-engine-installed path
 # in CI.
-COPY env_config.py proxy_pool.py output_writer.py captcha_solver.py \
+COPY env_config.py proxy_pool.py proxy_meter.py output_writer.py captcha_solver.py \
      fingerprint_client.py scraper_api_client.py scraper_api_engine.py diff_runs.py \
      page_parser.py page_flow.py run_state.py playwright_scraper.py puppeteer_scraper.py \
      selenium_scraper.py smoke_test.py ./
diff --git a/README.md b/README.md
index 8845408..fd6e317 100644
--- a/README.md
+++ b/README.md
@@ -210,7 +210,7 @@ or more, or spread a big list over time.
 
 ## Traffic through a paid proxy
 
-A residential proxy bills per byte, so two things happen only while a
+A residential proxy bills per byte, so three things happen only while a
 LOCAL browser goes through `--proxy` / `--proxy-file` (or
 `INSTAGRAM_PROXY`). Neither applies with `--cdp-endpoint` (that browser
 brings its own exit and is billed on its own side) or with `--scraper-api`
@@ -257,6 +257,33 @@ offline: the fixtures (the parser reads embedded JSON only), and each real
 browser driven through a local proxy to a test page. With blocking, only the
 HTML and the script went through the proxy, and the script still ran.
 
+**A traffic budget.** Every byte a local 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 URL (and no further `--posts` post) is fetched, the browser aborts
+any request it still tries, the rows already read are kept, and the run
+ends with `stop_reason: proxy_budget_exceeded`: exit 6 with rows, 5
+without. The URLs not fetched are listed as `not_attempted`, so a `--checkpoint` run can
+`--resume` them. It never goes on silently. At the end of a proxied run one
+line is printed, e.g. `Proxy traffic: 3.2 MB of 10 MB budget`, and 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 rows records them in
+`<out>.latest_attempt.meta.json`, leaving `<out>` and `<out>.meta.json` as
+they were.
+
+The count is an estimate of the tunnelled payload — request line, headers
+and body sent, response headers and body as received (compressed) — not
+the proxy's invoice: TLS records, the `CONNECT` exchange, WebSocket frames
+and Chrome'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 (request headers + body, response headers + body) |
+| Puppeteer | CDP `Network.loadingFinished.encodedDataLength` of every response that came over the network, plus each request's line, headers and body (`Network.requestWillBeSent`) |
+| Selenium | the same CDP events, from Chrome's performance log (turned on behind a local proxy only) |
+
 ## Scraper API mode
 
 With `--scraper-api` no browser is driven: each page is one 2Captcha
@@ -277,17 +304,18 @@ solves spent, and a hash of the output file). With the options below it
 also lists the posts left out (`excluded_urls`), every post seen so far
 (`seen_post_skus`) and whether the run was incremental. A run that
 collects nothing writes neither, so it never replaces your previous good
-file. The one exception is a proxy refused at the pre-check (see
-[Traffic through a paid proxy](#traffic-through-a-paid-proxy)): that run
-records itself in `<out>.latest_attempt.meta.json`, a separate file.
+file. The exceptions are a proxy refused at the pre-check, and any run
+through a local proxy that collects nothing (see
+[Traffic through a paid proxy](#traffic-through-a-paid-proxy)): those runs
+record themselves in `<out>.latest_attempt.meta.json`, a separate file.
 
 | Exit | Meaning |
 |---|---|
 | `0` | complete |
-| `6` | partial: rows were written, but the run did not finish cleanly — `stop_reason` says why (`blocked`, `rate_limited`, `remote_api_error`, `failed_pages`, `parse_error`, `proxy_pool_exhausted`) |
+| `6` | partial: rows were written, but the run did not finish cleanly — `stop_reason` says why (`blocked`, `rate_limited`, `remote_api_error`, `failed_pages`, `parse_error`, `proxy_pool_exhausted`, `proxy_budget_exceeded`) |
 | `3` | blocked or throttled, no rows |
 | `4` | no rows: every input was not found, or every post was older than `--since` or already collected by `--incremental-from` |
-| `5` | nothing could be read due to fetch, parse, or remote-service failure; no rows. Also a proxy that answered `407`/`402` at the pre-check, before any browser was started |
+| `5` | nothing could be read due to fetch, parse, or remote-service failure; no rows. Also a proxy that answered `407`/`402` at the pre-check, before any browser was started, and a proxy budget that ran out before any row was read |
 | `2` | bad usage, including input where every line was skipped |
 | `1` | crash (a bug; please report it) |
 
@@ -410,6 +438,7 @@ the command line.
 | `--proxy` / `--proxy-file` / `--proxy-shuffle` | `INSTAGRAM_PROXY` | one proxy or a rotating pool, for a local browser |
 | `--proxy-block-retries` | 3 | blocked answers before that proxy is dropped from the pool |
 | `--load-media` | off | load images, video and fonts behind a proxy too (they are aborted by default there; see [Traffic through a paid proxy](#traffic-through-a-paid-proxy)) |
+| `--proxy-budget-mb` | 10 | stop once this many MB went through a local proxy; `0` = unlimited (see [Traffic through a paid proxy](#traffic-through-a-paid-proxy)) |
 | `--scraper-api` | off | no browser: fetch each page through 2Captcha's Scraper API (needs `TWOCAPTCHA_KEY`; routed through `INSTAGRAM_CDP_ENDPOINT` when set) — posts only, see [Scraper API mode](#scraper-api-mode) |
 | `--cdp-endpoint` | `INSTAGRAM_CDP_ENDPOINT` | connect to a Scraping Browser API profile instead of launching a browser |
 | `--solve-captcha` | when-blocked | `off` disables the Browser API's own captcha auto-solve; there is no local solver |
diff --git a/output_writer.py b/output_writer.py
index 8705d72..c265bd5 100644
--- a/output_writer.py
+++ b/output_writer.py
@@ -29,6 +29,8 @@
 # caller (CI, a cron job, another program) must be able to tell these apart
 # without parsing stdout.
 # --------------------------------------------------------------------------- #
+import proxy_meter
+
 EXIT_OK = 0
 EXIT_CRASH = 1
 EXIT_BAD_USAGE = 2
@@ -358,6 +360,34 @@ def write_failed_attempt(
         "transport_error": transport_error,
         "data_updated": False,
     }
+    meta.update(proxy_meter.sidecar())  # proxy_bytes / proxy_budget_bytes / proxy_budget_exceeded
+    path = f"{out_path}{ATTEMPT_META_SUFFIX}"
+    with _atomic(path) as f:
+        json.dump(meta, f, ensure_ascii=False, indent=2)
+    proxy_meter.report()
+    return path
+
+
+def _write_attempt(out_path: str, *, status: str, stop_reason: str, engine: str, url: str, pages_requested: int,
+                   pages_completed: int, failed_pages: List[int], blocked: bool, started_at: float,
+                   extra: Optional[dict] = None) -> str:
+    """`<out>.latest_attempt.meta.json` for a proxied run that wrote no data."""
+    meta = {
+        "status": status,
+        "stop_reason": stop_reason,
+        "engine": engine,
+        "url": url,
+        "pages_requested": pages_requested,
+        "pages_completed": pages_completed,
+        "failed_pages": failed_pages or [],
+        "product_count": 0,
+        "started_at": started_at,
+        "finished_at": time.time(),
+        "blocked": bool(blocked),
+        "data_updated": False,
+    }
+    meta.update(proxy_meter.sidecar())
+    meta.update(extra or {})
     path = f"{out_path}{ATTEMPT_META_SUFFIX}"
     with _atomic(path) as f:
         json.dump(meta, f, ensure_ascii=False, indent=2)
@@ -398,6 +428,9 @@ def finish_run(
     NEVER overwrites a previous good output with an empty one unless the
     caller explicitly passed --allow-empty."""
     failed_pages = failed_pages or []
+    meter = proxy_meter.current()
+    if meter.exceeded:
+        incomplete_reason = proxy_meter.STOP_REASON
     partial = bool(failed_pages) or bool(incomplete_reason)
     zero_products = len(products) == 0
 
@@ -435,6 +468,11 @@ def finish_run(
     else:
         status, exit_code = "complete", EXIT_OK
         stop_reason = status
+    if meter.exceeded:
+        # The run stopped itself at --proxy-budget-mb: rows kept -> 6, none -> 5,
+        # whatever else the pages it did fetch said.
+        status, exit_code = ("remote_api_error", EXIT_REMOTE_API_ERROR) if zero_products else ("partial", EXIT_PARTIAL)
+        stop_reason = proxy_meter.STOP_REASON
 
     # The "never overwrite good output with empty" rule: a zero-product
     # outcome (whatever its status above — blocked/remote_api_error/empty
@@ -446,6 +484,13 @@ def finish_run(
     # engine's own logs carry the diagnostic detail — that's what a
     # failed run's output is FOR, not this file.
     if zero_products and not allow_empty:
+        if meter.active:
+            # Nothing is written, but what the proxy carried is still recorded
+            # (`<out>.latest_attempt.meta.json`; data and .meta.json untouched).
+            _write_attempt(out_path, status=status, stop_reason=stop_reason, engine=engine, url=url,
+                           pages_requested=pages_requested, pages_completed=pages_completed,
+                           failed_pages=failed_pages, blocked=blocked, started_at=started_at, extra=extra_meta)
+        proxy_meter.report()
         return exit_code
 
     escaped = write_output(products, out_path, fmt)
@@ -454,6 +499,7 @@ def finish_run(
         # Under the caller's extra, never over it: housekeeping must not
         # silently replace a value the engine chose to record.
         extra["csv_cells_escaped"] = escaped
+    extra.update(proxy_meter.sidecar())
     extra.update(extra_meta or {})
     if rejected_rows:
         extra["rejected_rows"] = rejected_rows
@@ -474,4 +520,5 @@ def finish_run(
         price_confirmed_pct=price_confirmed_pct, started_at=started_at,
         extra=extra or None,
     )
+    proxy_meter.report()
     return exit_code
diff --git a/page_flow.py b/page_flow.py
index da6f9a6..fa51c12 100644
--- a/page_flow.py
+++ b/page_flow.py
@@ -38,6 +38,7 @@
 from typing import List, Optional
 
 import page_parser as pp
+import proxy_meter as _meter
 import run_state
 from run_state import add_arguments as add_run_arguments
 from captcha_solver import detect_from_html
@@ -260,6 +261,9 @@ async def fetch_item(engine: Engine, args, url: str, index: int, proxy_pool, cli
     try:
         last_error, status = None, None
         for attempt in range(args.retries + 1):
+            if _meter.exceeded():
+                last_error = _meter.STOP_REASON
+                break
             try:
                 status = await session.goto(url)
                 last_error = None
@@ -275,6 +279,8 @@ async def fetch_item(engine: Engine, args, url: str, index: int, proxy_pool, cli
                     break  # a refused key or an empty balance: every retry would fail the same way
                 if attempt < args.retries:
                     await engine.sleep(args.retry_delay)
+        if last_error == _meter.STOP_REASON:
+            return Outcome(None, False, False, "none", failure=_meter.STOP_REASON)
         if last_error is not None:
             _log.error("%s permanently failed to load: %s", url, last_error)
             return Outcome(None, False, False, "none",
@@ -388,7 +394,7 @@ async def run(engine: Engine, args, *, urls: List[str], proxy_pool, client, star
         records[url] = record
         run_state.save(args, urls, queue, records)
         consecutive_blocks = consecutive_blocks + 1 if outcome.blocked else 0
-        stop = (outcome.failure == "proxy_pool_exhausted" or getattr(engine, "fatal", None)
+        stop = (outcome.failure == "proxy_pool_exhausted" or getattr(engine, "fatal", None) or _meter.exceeded()
                 or (proxy_pool is None and consecutive_blocks >= STOP_AFTER_CONSECUTIVE_BLOCKS))
         if stop:
             for pending in queue[i:args.max_results]:
@@ -453,7 +459,7 @@ def finish(args, *, products: List[Product], blocked: bool, remote_api_error: bo
         failed_pages=failed_pages or None, blocked=blocked, remote_api_error=remote_api_error,
         allow_empty=args.allow_empty, started_at=started_at, price_confirmed_pct=None,
         max_results=args.max_results, rate_limited=rate_limited,
-        incomplete_reason=next((f["reason"] for f in (failures or []) if f["reason"] in ("parse_error", "proxy_pool_exhausted")), None),
+        incomplete_reason=next((f["reason"] for f in (failures or []) if f["reason"] in ("parse_error", "proxy_pool_exhausted", _meter.STOP_REASON)), None),
         capped=bool(capped) if capped is not None else len(urls) > args.max_results or bool(getattr(args, "posts", 0)),
         extra_meta=extra,
     )
@@ -492,6 +498,14 @@ def validate_common(args, *, urls, skipped, print_err) -> Optional[int]:
 PROXY_PRECHECK_TIMEOUT_S = 20.0
 
 
+def start_proxy_meter(args, proxy_pool) -> "_meter.ProxyMeter":
+    """Meter this run's traffic when it goes through a LOCAL proxy (not
+    over --cdp-endpoint, not the Scraper API, not without a proxy)."""
+    active = (proxy_pool is not None and not getattr(args, "cdp_endpoint", None)
+              and not getattr(args, "scraper_api", False))
+    return _meter.start(getattr(args, "proxy_budget_mb", _meter.DEFAULT_BUDGET_MB), active=active)
+
+
 def proxy_precheck(args, proxy_pool, *, urls: List[str], engine_name: str, started_at: float,
                    print_err=None) -> Optional[int]:
     """Pre-check the exit the first URL will use; return the exit code to
@@ -541,3 +555,4 @@ def add_media_arguments(parser) -> None:
     # Turns blocking on without a proxy, so its effect on parsing can be
     # measured for free. Not a user feature.
     parser.add_argument("--force-block-media", action="store_true", help=argparse.SUPPRESS)
+    _meter.add_budget_argument(parser)
diff --git a/playwright_scraper.py b/playwright_scraper.py
index 4122428..865b483 100644
--- a/playwright_scraper.py
+++ b/playwright_scraper.py
@@ -41,6 +41,7 @@
 
 import env_config
 import page_flow
+import proxy_meter
 import scraper_api_engine
 import scraper_api_client
 import page_parser as pp
@@ -225,6 +226,10 @@ async def open(self, proxy) -> _PlaywrightSession:
         self.media_blocked = self.media_blocked or block
         context = await _new_context(self.browser, proxy, self.user_agent, reuse_default=reuse_default,
                                      block_media=block)
+        if proxy is not None and not reuse_default:
+            # Counted for --proxy-budget-mb; registered after the media route,
+            # so it is asked first and aborts everything once the budget is spent.
+            await proxy_meter.current().attach_playwright_async(context)
         page = await context.new_page()
         if self.autosolve:
             await _enable_scraping_browser_auto_solve(context, page)
@@ -266,6 +271,7 @@ async def run(args: argparse.Namespace) -> int:
     proxy_pool = ProxyPool(proxies, shuffle=args.proxy_shuffle, block_retries=args.proxy_block_retries) if proxies else None
     # Before any browser exists: a proxy that refuses its credential (407) or
     # has no traffic left (402) would refuse every request the browser makes.
+    page_flow.start_proxy_meter(args, proxy_pool)
     stop = page_flow.proxy_precheck(args, proxy_pool, urls=urls, engine_name=ENGINE_NAME, started_at=started_at)
     if stop is not None:
         return stop
diff --git a/proxy_meter.py b/proxy_meter.py
new file mode 100644
index 0000000..0de24bc
--- /dev/null
+++ b/proxy_meter.py
@@ -0,0 +1,442 @@
+"""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)
+    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
+    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: 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 = []
+        self._partial = {}
+
+    # ------------------------------------------------------------- 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 url.startswith(("data:", "blob:", "about:", "chrome")):
+                return
+            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, and a cache hit crossed nothing.
+            response = params.get("response") or {}
+            if 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:
+            def _block():
+                asyncio.ensure_future(client.send("Network.setBlockedURLs", {"urls": ["*"]}))
+            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 1dd5bba..c167832 100644
--- a/proxy_pool.py
+++ b/proxy_pool.py
@@ -324,8 +324,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 2b027ef..23b22b2 100644
--- a/puppeteer_scraper.py
+++ b/puppeteer_scraper.py
@@ -50,6 +50,7 @@
 
 import env_config
 import page_flow
+import proxy_meter
 import scraper_api_engine
 import scraper_api_client
 import page_parser as pp
@@ -265,6 +266,8 @@ async def open(self, proxy) -> _PyppeteerSession:
         self.media_blocked = self.media_blocked or block
         page = await _open_page(browser, proxy=proxy, user_agent=self.user_agent, autosolve=self.autosolve,
                                 block_media=block)
+        if proxy is not None and self.remote_browser is None:
+            await proxy_meter.current().attach_pyppeteer(page)  # --proxy-budget-mb
         return _PyppeteerSession(page, browser, owns_browser=self.remote_browser is None)
 
     async def sleep(self, seconds: float) -> None:
@@ -325,6 +328,7 @@ async def run(args: argparse.Namespace) -> int:
         proxy_pool = None
     # Before any browser exists: a proxy that refuses its credential (407) or
     # has no traffic left (402) would refuse every request the browser makes.
+    page_flow.start_proxy_meter(args, proxy_pool)
     stop = page_flow.proxy_precheck(args, proxy_pool, urls=urls, engine_name=ENGINE_NAME, started_at=started_at)
     if stop is not None:
         return stop
diff --git a/pyproject.toml b/pyproject.toml
index 9ceedaa..4bae31e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,6 +31,7 @@ Repository = "https://github.com/2scraper/instagram-scraper"
 py-modules = [
     "env_config",
     "proxy_pool",
+    "proxy_meter",
     "output_writer",
     "captcha_solver",
     "fingerprint_client",
diff --git a/selenium_scraper.py b/selenium_scraper.py
index be2aca7..abf7db8 100644
--- a/selenium_scraper.py
+++ b/selenium_scraper.py
@@ -52,6 +52,7 @@
 
 import env_config
 import page_flow
+import proxy_meter
 import scraper_api_engine
 import page_parser as pp
 from output_writer import EXIT_BAD_USAGE, EXIT_CRASH
@@ -159,6 +160,9 @@ def _build_driver(*, headless: bool, proxy: Optional[Proxy], cdp_endpoint: Optio
                 proxy.host, proxy.port,
             )
         options.add_argument(f"--proxy-server={proxy.server_only()}")
+        # Chrome's performance log carries the CDP Network events that
+        # proxy_meter counts for --proxy-budget-mb.
+        proxy_meter.selenium_logging_prefs(options)
 
     if Service and _CHROMEDRIVER_PATH:
         service = Service(executable_path=_CHROMEDRIVER_PATH)
@@ -208,6 +212,7 @@ async def goto(self, url: str) -> Optional[int]:
         self.driver.set_page_load_timeout(NAV_TIMEOUT_S)
         self.driver.get(url)
         time.sleep(READINESS_WAIT_S)
+        proxy_meter.current().poll_selenium(self.driver)  # --proxy-budget-mb
         try:
             reported = self.driver.execute_script(_STATUS_JS)
             return int(reported) if reported else None
@@ -221,6 +226,7 @@ async def wait(self, seconds: float) -> None:
         time.sleep(seconds)
 
     async def close(self) -> None:
+        proxy_meter.current().poll_selenium(self.driver)  # what the last page still loaded
         _quit(self.driver)
 
 
@@ -245,6 +251,8 @@ async def open(self, proxy) -> _SeleniumSession:
                 self.media_blocked = True
             except WebDriverException as exc:
                 log.warning("Could not block media over CDP (continuing with it): %s", str(exc).splitlines()[0][:200])
+        if proxy is not None and not self.args.cdp_endpoint:
+            proxy_meter.current().attach_selenium(driver, keep_entries=False)  # --proxy-budget-mb
         return _SeleniumSession(driver)
 
     async def sleep(self, seconds: float) -> None:
@@ -296,6 +304,7 @@ def run(args: argparse.Namespace) -> int:
     # has no traffic left (402) would refuse every request the browser makes.
     # Checked with the proxy as configured, credentials included: the
     # question is whether the proxy account itself is refused or exhausted.
+    page_flow.start_proxy_meter(args, proxy_pool)
     stop = page_flow.proxy_precheck(args, proxy_pool, urls=urls, engine_name=ENGINE_NAME, started_at=started_at)
     if stop is not None:
         return stop
diff --git a/smoke_test.py b/smoke_test.py
index f4ba7b0..1ab77e4 100644
--- a/smoke_test.py
+++ b/smoke_test.py
@@ -1527,6 +1527,8 @@ def _():
 
 @check("media: playwright aborts image/media/font and continues everything else, on the context, behind a proxy only")
 def _():
+    import proxy_meter
+    proxy_meter.start(None, active=False)  # media only here; the meter has its own checks
     import argparse
 
     class Route:
@@ -1627,6 +1629,8 @@ async def create():
 
 @check("media: selenium blocks images, video and fonts by URL extension (not js/css) right after the driver is built, behind a proxy only")
 def _():
+    import proxy_meter
+    proxy_meter.start(None, active=False)  # media only here; the meter has its own checks
     from unittest import mock
 
     class Driver:
@@ -1667,6 +1671,308 @@ def _():
         assert json.loads(Path(td, "o.json.meta.json").read_text())["media_blocked"] is True
 
 
+# --------------------------------------------------------------------------- #
+# Proxy traffic meter and --proxy-budget-mb (proxy_meter.py; its own byte
+# counts against a stub proxy are in tests/test_proxy_meter.py)
+# --------------------------------------------------------------------------- #
+import proxy_meter  # noqa: E402
+
+
+class _MeteredEngine(_FakeEngine):
+    """_FakeEngine whose every navigation costs `cost` proxy bytes."""
+
+    def __init__(self, by_url, cost):
+        super().__init__(by_url)
+        self.cost = cost
+
+    async def open(self, proxy):
+        session = await super().open(proxy)
+        goto = session.goto
+
+        async def metered_goto(url):
+            proxy_meter.current().add(self.cost)
+            return await goto(url)
+        session.goto = metered_goto
+        return session
+
+
+def _metered_run(by_url, argv, *, budget_mb, cost, active=True):
+    import contextlib
+    import io
+    args = playwright_scraper.build_arg_parser().parse_args(
+        [*argv, "--delay-between-pages", "0", "--proxy-budget-mb", str(budget_mb)])
+    args._solve_budget = page_flow.SolveBudget(0)
+    urls, _skipped = page_flow.resolve_urls(args)
+    proxy_meter.start(args.proxy_budget_mb, active=active)
+    engine = _MeteredEngine(by_url, cost)
+    err = io.StringIO()
+    try:
+        with tempfile.TemporaryDirectory() as td:
+            args.out = str(Path(td) / "o.json")
+            Path(args.out).write_text("[]")  # a previous run's data
+            with contextlib.redirect_stderr(err):
+                rc = asyncio.run(page_flow.run(engine, args, urls=urls, proxy_pool=None, client=None, started_at=0.0))
+            read = lambda suffix: json.loads(Path(args.out + suffix).read_text()) if Path(args.out + suffix).exists() else None  # noqa: E731
+            meta, attempt, rows = read(".meta.json"), read(output_writer.ATTEMPT_META_SUFFIX), read("")
+    finally:
+        proxy_meter.start(None, active=False)
+    return rc, meta, attempt, rows, engine, err.getvalue()
+
+
+def _three_urls():
+    return ["--urls-file", str(_urls_file(["natgeo", _IMAGE_URL, _CAROUSEL_URL]))]
+
+
+def _three_pages():
+    return {_PROFILE_URL: {"html": _fx("profile_natgeo")}, _IMAGE_URL: {"html": _fx("post_image")},
+            _CAROUSEL_URL: {"html": _fx("post_carousel")}}
+
+
+@check("proxy budget: once the metered bytes pass --proxy-budget-mb no further URL is fetched — the rows already read are kept, exit 6, stop_reason proxy_budget_exceeded, the sidecar carries proxy_bytes / proxy_budget_bytes / proxy_budget_exceeded, and one 'Proxy traffic' line is printed")
+def _():
+    # 1,000-byte budget, 600 bytes per page: over it on the 2nd of 3 URLs.
+    rc, meta, attempt, rows, eng, err = _metered_run(_three_pages(), _three_urls(), budget_mb=0.001, cost=600)
+    assert rc == output_writer.EXIT_PARTIAL, (rc, err[-500:])
+    assert eng.asked == [_PROFILE_URL, _IMAGE_URL], eng.asked
+    assert [r["category"] for r in rows] == ["profile", "post"], rows
+    assert meta["status"] == "partial" and meta["stop_reason"] == "proxy_budget_exceeded", meta
+    assert (meta["proxy_bytes"], meta["proxy_budget_bytes"], meta["proxy_budget_exceeded"]) == (1200, 1000, True), meta
+    assert meta["failed_urls"] == [{"url": _CAROUSEL_URL, "reason": "not_attempted"}], meta["failed_urls"]
+    assert err.count("Proxy traffic: 0.0 MB of 0.001 MB budget") == 1, err[-500:]
+    assert attempt is None, "a run that wrote its rows has nothing to put in the attempt sidecar"
+
+
+@check("proxy budget: spent before any row was read -> exit 5 (not 4, not 3); the previous data and .meta.json stay, and <out>.latest_attempt.meta.json records stop_reason proxy_budget_exceeded with the byte count; a navigation that failed after spending the budget is not retried")
+def _():
+    rc, meta, attempt, rows, eng, err = _metered_run({"*": {"html": "<html><title>x</title></html>"}}, _three_urls(),
+                                                     budget_mb=0.001, cost=5000)
+    assert rc == output_writer.EXIT_REMOTE_API_ERROR, (rc, err[-400:])
+    assert rows == [] and meta is None, "the previous data stays and no .meta.json is written"
+    assert (attempt["stop_reason"], attempt["proxy_bytes"], attempt["proxy_budget_exceeded"]) == \
+        ("proxy_budget_exceeded", 5000, True), attempt
+    assert attempt["data_updated"] is False and len(eng.asked) == 1, (attempt, eng.asked)
+    assert "Proxy traffic: 0.0 MB of 0.001 MB budget" in err
+    rc, meta, attempt, rows, eng, err = _metered_run({"*": {"goto": RuntimeError("net::ERR_TIMED_OUT")}},
+                                                     ["--url", "natgeo", "--retries", "2", "--retry-delay", "0"],
+                                                     budget_mb=0.001, cost=5000)
+    assert rc == output_writer.EXIT_REMOTE_API_ERROR and len(eng.asked) == 1, (rc, eng.asked)
+    assert attempt["stop_reason"] == "proxy_budget_exceeded", attempt
+
+
+@check("proxy budget: with no local proxy nothing is metered or stopped — proxy_bytes is null, no 'Proxy traffic' line; --proxy-budget-mb 0 is unlimited; a run under its budget is complete (exit 0) and records its bytes")
+def _():
+    rc, meta, _a, rows, _e, err = _metered_run(_three_pages(), _three_urls(), budget_mb=0.001, cost=10 ** 9, active=False)
+    assert rc == output_writer.EXIT_OK and meta["proxy_bytes"] is None and meta["proxy_budget_exceeded"] is False, meta
+    assert "Proxy traffic" not in err
+    rc, meta, _a, rows, _e, err = _metered_run(_three_pages(), _three_urls(), budget_mb=0, cost=10 ** 9)
+    assert rc == output_writer.EXIT_OK and meta["proxy_budget_bytes"] is None and meta["proxy_bytes"] == 3 * 10 ** 9, meta
+    rc, meta, _a, rows, _e, err = _metered_run(_three_pages(), _three_urls(), budget_mb=10, cost=1000)
+    assert rc == output_writer.EXIT_OK and meta["proxy_bytes"] == 3000 and meta["proxy_budget_bytes"] == 10 ** 7, (rc, meta)
+    assert "Proxy traffic: 0.0 MB of 10 MB budget" in err
+
+
+@check("proxy budget: finish_run puts the budget stop above every other outcome — a run that also met a block or a failed page still ends 6 with rows and 5 without, stop_reason proxy_budget_exceeded")
+def _():
+    import dataclasses
+    rows = [output_writer.Product(**{f.name: None for f in dataclasses.fields(output_writer.Product)
+                                     if f.default is dataclasses.MISSING and f.default_factory is dataclasses.MISSING})]
+    rows[0].sku = "instagram-user-1"
+    try:
+        for products, blocked, failed, want in ((rows, True, [], 6), ([], True, [], 5), (rows, False, [2], 6), ([], False, [], 5)):
+            proxy_meter.start(0.001, active=True)
+            proxy_meter.current().add(5000)
+            with tempfile.TemporaryDirectory() as td:
+                out = str(Path(td) / "o.json")
+                rc = output_writer.finish_run(products=products, out_path=out, fmt="json", engine="t", url="u",
+                                              pages_requested=2, pages_completed=1, failed_pages=failed, blocked=blocked,
+                                              remote_api_error=False, allow_empty=False, started_at=0.0)
+                path = Path(out + (".meta.json" if products else output_writer.ATTEMPT_META_SUFFIX))
+                meta = json.loads(path.read_text())
+            assert rc == want and meta["stop_reason"] == "proxy_budget_exceeded", (blocked, failed, rc, meta)
+            assert meta["proxy_budget_exceeded"] is True and meta["proxy_bytes"] == 5000, meta
+    finally:
+        proxy_meter.start(None, active=False)
+
+
+@check("proxy budget: --proxy-budget-mb is on every engine (default 10, 0 = unlimited, a negative or non-number value is a usage error naming the flag); the meter is active only for a LOCAL proxy — never over --cdp-endpoint, --scraper-api or with no proxy")
+def _():
+    import argparse
+    import contextlib
+    import io
+    for mod in (playwright_scraper, selenium_scraper, puppeteer_scraper):
+        parser = mod.build_arg_parser()
+        assert parser.parse_args(["--url", "x"]).proxy_budget_mb == 10, mod.__name__
+        assert "--proxy-budget-mb" in parser.format_help(), mod.__name__
+        for bad in ("-1", "lots"):
+            err = io.StringIO()
+            with contextlib.redirect_stderr(err):
+                try:
+                    parser.parse_args(["--url", "x", "--proxy-budget-mb", bad])
+                    raise AssertionError("accepted %s" % bad)
+                except SystemExit as exc:
+                    assert exc.code == 2
+            assert "--proxy-budget-mb" in err.getvalue(), err.getvalue()
+        source = (ROOT / f"{mod.__name__}.py").read_text(encoding="utf-8")
+        assert source.index("page_flow.start_proxy_meter(") < source.index("page_flow.proxy_precheck("), mod.__name__
+    pool = proxy_pool.ProxyPool([proxy_pool.parse_proxy_line("h:1")])
+    ns = lambda **k: argparse.Namespace(**{"proxy_budget_mb": 10, "cdp_endpoint": None, "scraper_api": False, **k})  # noqa: E731
+    try:
+        assert page_flow.start_proxy_meter(ns(), pool).active is True
+        assert page_flow.start_proxy_meter(ns(), None).active is False
+        assert page_flow.start_proxy_meter(ns(cdp_endpoint="ws://h"), pool).active is False
+        assert page_flow.start_proxy_meter(ns(scraper_api=True), pool).active is False
+    finally:
+        proxy_meter.start(None, active=False)
+
+
+@check("proxy meter wiring: Playwright counts a proxied LOCAL context (requestfinished + requestfailed) and adds the budget route after the media route; Selenium turns Chrome's performance log on and tees it; pyppeteer opens a counting CDP session; none of it over --cdp-endpoint or with no proxy")
+def _():
+    import argparse
+    from unittest import mock
+
+    class Context:
+        def __init__(self):
+            self.routes, self.events = [], []
+
+        def on(self, name, fn):
+            self.events.append(name)
+
+        async def route(self, pattern, handler):
+            self.routes.append(handler)
+
+        async def new_page(self):
+            return type("P", (), {"on": lambda *a: None})()
+
+    class Browser:
+        def __init__(self):
+            self.contexts = [Context()]
+
+        async def new_context(self, **kw):
+            return Context()
+
+    proxy = proxy_pool.parse_proxy_line("h:1:u:p")
+    try:
+        for use_proxy, cdp, argv, want_events, want_routes in (
+                (True, False, [], True, 2), (True, False, ["--proxy-budget-mb", "0"], True, 1),
+                (True, False, ["--load-media"], True, 1), (False, False, [], False, 0), (True, True, [], False, 0)):
+            args = playwright_scraper.build_arg_parser().parse_args(["--url", "x", *argv])
+            args.cdp_endpoint = "ws://h:1" if cdp else None
+            proxy_meter.start(args.proxy_budget_mb, active=use_proxy and not cdp)
+            engine = playwright_scraper._PlaywrightEngine(Browser(), args, autosolve=False, user_agent=None, client=None)
+            session = asyncio.run(engine.open(proxy if use_proxy else None))
+            context = session.context
+            assert (sorted(set(context.events)) == ["requestfailed", "requestfinished"]) is want_events, (argv, context.events)
+            assert len(context.routes) == want_routes, (argv, use_proxy, cdp, context.routes)
+            if want_routes and args.proxy_budget_mb:
+                assert context.routes[-1] == proxy_meter.current().playwright_budget_route_async
+
+        class Driver:
+            def __init__(self):
+                self.sent = []
+
+            def execute_cdp_cmd(self, cmd, params):
+                self.sent.append((cmd, params))
+
+            def get_log(self, kind):
+                return []
+
+        for use_proxy, cdp, want in ((True, False, True), (False, False, False), (True, True, False)):
+            driver = Driver()
+            args = selenium_scraper.build_arg_parser().parse_args(["--url", "x"])
+            args.cdp_endpoint = "ws://127.0.0.1:9222" if cdp else None
+            proxy_meter.start(10, active=use_proxy and not cdp)
+            engine = selenium_scraper._SeleniumEngine(args, user_agent=None, client=None)
+            with mock.patch.object(selenium_scraper, "_build_driver", lambda **k: driver):
+                asyncio.run(engine.open(proxy if use_proxy else None))
+            assert hasattr(driver, "_proxy_meter_poll") is want, (use_proxy, cdp)
+
+        class Options:
+            def __init__(self):
+                self.caps = {"goog:loggingPrefs": {"browser": "ALL"}}
+
+            def to_capabilities(self):
+                return dict(self.caps)
+
+            def set_capability(self, name, value):
+                self.caps[name] = value
+        options = Options()
+        proxy_meter.selenium_logging_prefs(options)
+        assert options.caps["goog:loggingPrefs"] == {"browser": "ALL", "performance": "ALL"}
+        assert "proxy_meter.selenium_logging_prefs(options)" in (ROOT / "selenium_scraper.py").read_text(encoding="utf-8")
+
+        attached = []
+
+        async def fake_attach(self, page):
+            attached.append(page)
+
+        async def fake_launch(**kw):
+            return "browser"
+
+        async def fake_open_page(browser, **kw):
+            return argparse.Namespace(on=lambda *a: None)
+        with mock.patch.object(puppeteer_scraper, "_launch", fake_launch), \
+                mock.patch.object(puppeteer_scraper, "_open_page", fake_open_page), \
+                mock.patch.object(proxy_meter.ProxyMeter, "attach_pyppeteer", fake_attach):
+            for use_proxy, remote, want in ((True, None, 1), (False, None, 0), (True, "remote", 0)):
+                attached.clear()
+                args = puppeteer_scraper.build_arg_parser().parse_args(["--url", "x"])
+                proxy_meter.start(10, active=use_proxy and not remote)
+                engine = puppeteer_scraper._PyppeteerEngine(args, remote_browser=remote, autosolve=False,
+                                                            user_agent=None, client=None)
+                asyncio.run(engine.open(proxy if use_proxy else None))
+                assert len(attached) == want, (use_proxy, remote, attached)
+    finally:
+        proxy_meter.start(None, active=False)
+
+
+@check("canary: the run passes --proxy-budget-mb 10 and is followed by an always-run 'Proxy traffic' step; that step writes '**canary**: proxy traffic X.X MB of 10 MB' to the job summary, is red with '::error::stopped: proxy budget 10 MB exceeded' when the run stopped at the budget, and copes with a missing sidecar")
+def _():
+    import subprocess
+    import sys
+    if not (ROOT / ".github").is_dir():
+        return  # the Docker image ships no workflows; the repository's own CI runs this check
+    text = (ROOT / ".github" / "workflows" / "canary.yml").read_text(encoding="utf-8")
+    runs = [line for line in text.replace("\\\n", " ").splitlines() if "_scraper.py" in line and "--out" in line]
+    assert runs and all("--proxy-budget-mb 10" in line for line in runs), runs
+    step = text.split("- name: Proxy traffic", 1)
+    assert len(step) == 2 and text.index("--out canary.json") < text.index("- name: Proxy traffic"), "no summary step"
+    body = step[1].split("- name:", 1)[0]
+    assert "if: always()" in body, "the summary step must run whatever the run did"
+    assert "python3 .github/proxy_traffic_summary.py canary.json canary 10" in body
+    script = str(ROOT / ".github" / "proxy_traffic_summary.py")
+    with tempfile.TemporaryDirectory() as tmp:
+        summary = Path(tmp) / "summary.md"
+        env = dict(os.environ, GITHUB_STEP_SUMMARY=str(summary))
+
+        def summarise(meta=None, attempt=None):
+            for suffix, data in ((".meta.json", meta), (".latest_attempt.meta.json", attempt)):
+                path = Path(tmp) / ("canary.json" + suffix)
+                path.unlink() if path.exists() else None
+                if data is not None:
+                    path.write_text(json.dumps(data), encoding="utf-8")
+            return subprocess.run([sys.executable, script, "canary.json", "canary", "10"], cwd=tmp, env=env,
+                                  capture_output=True, text=True, timeout=60)
+        proc = summarise({"proxy_bytes": 4_321_000, "proxy_budget_bytes": 10 ** 7, "proxy_budget_exceeded": False,
+                          "finished_at": 2})
+        assert proc.returncode == 0 and "::error::" not in proc.stdout, proc.stdout
+        assert summary.read_text().splitlines()[-1] == "**canary**: proxy traffic 4.3 MB of 10 MB"
+        proc = summarise(None, {"proxy_bytes": 10_200_000, "proxy_budget_bytes": 10 ** 7,
+                                "proxy_budget_exceeded": True, "finished_at": 3})
+        assert proc.returncode == 1 and "::error::stopped: proxy budget 10 MB exceeded" in proc.stdout, proc.stdout
+        assert summary.read_text().splitlines()[-1] == "**canary**: proxy traffic 10.2 MB of 10 MB"
+        proc = summarise({"proxy_bytes": 1, "proxy_budget_exceeded": False, "finished_at": 1},
+                         {"proxy_bytes": 10_200_000, "proxy_budget_exceeded": True, "finished_at": 5})
+        assert proc.returncode == 1, "the newer of the two sidecars is the run's"
+        proc = summarise()
+        assert proc.returncode == 0 and "not recorded" in summary.read_text().splitlines()[-1], proc.stdout
+        proc = summarise({"proxy_bytes": None, "proxy_budget_exceeded": False})
+        assert proc.returncode == 0 and "no local proxy" in summary.read_text().splitlines()[-1]
+        (Path(tmp) / "canary.json.meta.json").write_text("{not json", encoding="utf-8")
+        proc = subprocess.run([sys.executable, script, "canary.json", "canary", "10"], cwd=tmp, env=env,
+                              capture_output=True, text=True, timeout=60)
+        assert proc.returncode == 0 and "unreadable" in proc.stdout, proc.stdout
+
+
 def run() -> int:
     """All @check-decorated functions above already ran at import time
     (that's the point — see the `check()` docstring) and self-registered
diff --git a/tests/test_proxy_meter.py b/tests/test_proxy_meter.py
new file mode 100644
index 0000000..c442db0
--- /dev/null
+++ b/tests/test_proxy_meter.py
@@ -0,0 +1,485 @@
+"""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})
+        self.assertEqual(meter.bytes, sent + 1300, "an error page or a cache hit 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)
+
+
+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()