#!/usr/bin/env python3
"""NO_PROXY matching lab: which clients send which hosts through the proxy.

A loopback HTTP proxy records every absolute-form request and CONNECT and
answers them itself. Test destinations are reserved .test names and
documentation (TEST-NET / 2001:db8::/32) addresses. On macOS every client runs
inside sandbox-exec with outbound network denied except loopback, so a request
that skips the proxy fails immediately and nothing leaves the machine.

Each cell is classified only from the proxy log: "proxy" if the proxy saw at
least one request from that cell, otherwise "not_proxied". Kernel sandbox
denials and the loopback origin log are recorded as separate corroborating
evidence of a direct attempt; they never change the classification.

Commands:
  setup   --work DIR                 download pinned toolchains/packages into DIR
  run     --work DIR --out DIR       run the matrix and write results.json/.csv
  compare OLD.json NEW.json          compare route classifications of two runs
"""

import argparse
import csv
import datetime as dt
import getpass
import hashlib
import json
import os
import platform
import re
import shutil
import socket
import socketserver
import struct
import subprocess
import sys
import tarfile
import threading
import time
import urllib.parse
import urllib.request
from pathlib import Path

HERE = Path(__file__).resolve().parent
NODE_VERSIONS = ["v26.10.0", "v24.21.0"]
NODE_DIST = "https://nodejs.org/dist"
# SHA-256 of the official tarballs, copied from nodejs.org SHASUMS256.txt when
# this lab was first run. setup checks the download against SHASUMS256.txt and,
# when the platform is listed here, against this pin as well.
NODE_PINNED_SHA256 = {
    "node-v26.10.0-darwin-arm64.tar.gz": "751fdf7439f115d87ee2a8f3f18c065b6151852068e3e666ac60ac2996f75ac9",
    "node-v24.21.0-darwin-arm64.tar.gz": "bed7eea5325e1108f32ce5228ddd6a5f0f08a499ee42aa7442aea583702f6057",
}
PY_LIBS = ["urllib", "requests", "httpx", "httpx2", "aiohttp"]
DEFAULT_CURLS = ["/usr/bin/curl", "/opt/homebrew/opt/curl/bin/curl"]
DEFAULT_WGET = "/opt/homebrew/bin/wget"
SANDBOX_PROFILE = HERE / "nonet.sb"
P = "{proxy}"  # placeholder replaced with the lab proxy URL
# Client errors that show the client itself tried to resolve or reach the
# destination (a direct attempt), as opposed to failing before any network I/O.
DIRECT_ERROR = re.compile(r"ENOTFOUND|EAI_NONAME|nodename nor servname|Could not resolve host|unable to resolve "
                          r"host|no such host|EPERM|Operation not permitted|operation not permitted|"
                          r"NameResolutionError|NewConnectionError|Failed to connect to (?!127\.0\.0\.1\b)")
# Errors that mention the lab proxy mean the client tried the proxy, not the destination.
PROXY_FAILURE = re.compile(r"ProxyError|Unable to connect to proxy|127\.0\.0\.1")
HARNESS_FILES = ["no_proxy_lab.py", "nonet.sb", "requirements.txt", "package.json", "package-lock.json",
                 "clients/py_client.py", "clients/node_client.cjs", "clients/go_client.go"]

# ---------------------------------------------------------------------------
# Test cases. Every value is applied to both no_proxy and NO_PROXY unless the
# case is in the "variables" section, which tests the variable names.
# ---------------------------------------------------------------------------
A, S, D = "http://example.test/", "http://sub.example.test/", "http://a.b.example.test/"
L, O = "http://notexample.test/", "http://other.test/"
A8080, S8080 = "http://example.test:8080/", "http://sub.example.test:8080/"
V4, V4B, V4OUT, V4P = "http://192.0.2.10/", "http://192.0.2.11/", "http://198.51.100.10/", "http://192.0.2.10:8080/"
V6, V6OUT, V6P = "http://[2001:db8::10]/", "http://[2001:db8:1::10]/", "http://[2001:db8::10]:8080/"
AT, ST, AU = "http://example.test./", "http://sub.example.test./", "http://EXAMPLE.TEST/"


def both(value):
    return {"no_proxy": value, "NO_PROXY": value}


CASES = [
    # Controls: the proxy variables alone must route each URL shape via the proxy.
    {"id": "b-none", "section": "baseline", "label": "no NO_PROXY set", "vars": {},
     "urls": [A, S, A8080, V4, V4P, V6, V6P, AT, AU]},
    {"id": "b-unrelated", "section": "baseline", "label": "unrelated entry", "vars": both("other.test"), "urls": [A]},
    # Main matrix.
    {"id": "m-exact", "section": "main", "label": "exact host / bare suffix", "vars": both("example.test"), "urls": [A, S, D, L]},
    {"id": "m-leading-dot", "section": "main", "label": "leading dot", "vars": both(".example.test"), "urls": [A, S, D, L]},
    {"id": "m-star-dot", "section": "main", "label": "*. wildcard", "vars": both("*.example.test"), "urls": [A, S, D, L]},
    {"id": "m-star", "section": "main", "label": "bare *", "vars": both("*"), "urls": [A, V4]},
    {"id": "m-star-in-list", "section": "main", "label": "* inside a list", "vars": both("other.test,*"), "urls": [A]},
    {"id": "m-star-spaced", "section": "main", "label": "* with spaces", "vars": both(" * "), "urls": [A]},
    {"id": "m-ipv4", "section": "main", "label": "IPv4 address", "vars": both("192.0.2.10"), "urls": [V4, V4B]},
    {"id": "m-ipv4-cidr", "section": "main", "label": "IPv4 CIDR", "vars": both("192.0.2.0/24"), "urls": [V4, V4OUT]},
    {"id": "m-ipv4-port", "section": "main", "label": "IPv4 address:port", "vars": both("192.0.2.10:8080"), "urls": [V4P, V4]},
    {"id": "m-ipv6-bare", "section": "main", "label": "IPv6 bare", "vars": both("2001:db8::10"), "urls": [V6]},
    {"id": "m-ipv6-bracketed", "section": "main", "label": "IPv6 bracketed", "vars": both("[2001:db8::10]"), "urls": [V6]},
    {"id": "m-ipv6-cidr", "section": "main", "label": "IPv6 CIDR", "vars": both("2001:db8::/48"), "urls": [V6, V6OUT]},
    {"id": "m-ipv6-bracketed-port", "section": "main", "label": "IPv6 [address]:port", "vars": both("[2001:db8::10]:8080"), "urls": [V6P, V6]},
    {"id": "m-host-port", "section": "main", "label": "host:port", "vars": both("example.test:8080"), "urls": [A8080, A]},
    {"id": "m-dot-port", "section": "main", "label": ".domain:port", "vars": both(".example.test:8080"), "urls": [S8080, S]},
    {"id": "m-trailing-dot-entry", "section": "main", "label": "trailing dot in entry", "vars": both("example.test."), "urls": [A, S]},
    {"id": "m-trailing-dot-host", "section": "main", "label": "trailing dot in request host", "vars": both("example.test"), "urls": [AT, ST]},
    {"id": "m-uppercase-entry", "section": "main", "label": "uppercase entry", "vars": both("EXAMPLE.TEST"), "urls": [A, S]},
    {"id": "m-uppercase-host", "section": "main", "label": "uppercase request host", "vars": both("example.test"), "urls": [AU]},
    {"id": "m-spaces-around-comma", "section": "main", "label": "spaces around comma", "vars": both("other.test , example.test"), "urls": [A]},
    {"id": "m-trailing-space", "section": "main", "label": "space after entry", "vars": both("example.test ,other.test"), "urls": [A]},
    {"id": "m-space-separated", "section": "main", "label": "space-separated list", "vars": both("other.test example.test"), "urls": [A, O]},
    {"id": "m-empty-item-middle", "section": "main", "label": "empty item in list", "vars": both("other.test,,example.test"), "urls": [A]},
    {"id": "m-empty-items-trailing", "section": "main", "label": "trailing empty items", "vars": both("other.test,,"), "urls": [A]},
    # The combined spelling suggested by the single-entry rows, tested directly rather than inferred.
    {"id": "m-apex-plus-dot", "section": "main", "label": "apex and leading dot listed together",
     "vars": both("example.test,.example.test"), "urls": [A, S, D, L]},
    # Variable names and precedence.
    {"id": "v-upper-only", "section": "variables", "label": "only NO_PROXY set", "vars": {"NO_PROXY": "example.test"}, "urls": [A]},
    {"id": "v-lower-only", "section": "variables", "label": "only no_proxy set", "vars": {"no_proxy": "example.test"}, "urls": [A]},
    {"id": "v-conflict", "section": "variables", "label": "NO_PROXY and no_proxy differ",
     "vars": {"NO_PROXY": "example.test", "no_proxy": "other.test"}, "urls": [A, O]},
    {"id": "v-empty-lower-no-proxy", "section": "variables", "label": "no_proxy empty, NO_PROXY set",
     "vars": {"NO_PROXY": "example.test", "no_proxy": ""}, "urls": [A]},
    {"id": "v-empty-lower-http-proxy", "section": "variables", "label": "http_proxy empty, HTTP_PROXY set",
     "vars": {}, "proxy_vars": {"HTTP_PROXY": P, "http_proxy": "", "HTTPS_PROXY": P, "https_proxy": P}, "urls": [A]},
    # Loopback special cases: no NO_PROXY at all.
    {"id": "lo-none", "section": "loopback", "label": "loopback destination, no NO_PROXY", "vars": {},
     "urls": ["http://127.0.0.1:{origin}/", "http://localhost:{origin}/", "http://[::1]:{origin}/"]},
    {"id": "lo-listed", "section": "loopback", "label": "loopback listed, bare ::1",
     "vars": both("localhost,127.0.0.1,::1"),
     "urls": ["http://127.0.0.1:{origin}/", "http://localhost:{origin}/", "http://[::1]:{origin}/"]},
    {"id": "lo-listed-bracketed", "section": "loopback", "label": "loopback listed, bracketed [::1]",
     "vars": both("localhost,127.0.0.1,[::1]"),
     "urls": ["http://127.0.0.1:{origin}/", "http://localhost:{origin}/", "http://[::1]:{origin}/"]},
    # HTTPS spot-check: same decision as the HTTP twin?
    {"id": "h-none", "section": "https", "label": "no NO_PROXY set", "vars": {},
     "urls": ["https://example.test/", "https://192.0.2.10/", "https://[2001:db8::10]/"],
     "twins": {"https://example.test/": ["b-none", A], "https://192.0.2.10/": ["b-none", V4],
               "https://[2001:db8::10]/": ["b-none", V6]}},
    {"id": "h-exact", "section": "https", "label": "exact host / bare suffix", "vars": both("example.test"),
     "urls": ["https://sub.example.test/"], "twins": {"https://sub.example.test/": ["m-exact", S]}},
    {"id": "h-leading-dot", "section": "https", "label": "leading dot", "vars": both(".example.test"),
     "urls": ["https://example.test/"], "twins": {"https://example.test/": ["m-leading-dot", A]}},
    {"id": "h-star-dot", "section": "https", "label": "*. wildcard", "vars": both("*.example.test"),
     "urls": ["https://example.test/"], "twins": {"https://example.test/": ["m-star-dot", A]}},
    {"id": "h-ipv4-cidr", "section": "https", "label": "IPv4 CIDR", "vars": both("192.0.2.0/24"),
     "urls": ["https://192.0.2.10/"], "twins": {"https://192.0.2.10/": ["m-ipv4-cidr", V4]}},
    {"id": "h-ipv6-bare", "section": "https", "label": "IPv6 bare", "vars": both("2001:db8::10"),
     "urls": ["https://[2001:db8::10]/"], "twins": {"https://[2001:db8::10]/": ["m-ipv6-bare", V6]}},
]
DEFAULT_PROXY_VARS = {"http_proxy": P, "HTTP_PROXY": P, "https_proxy": P, "HTTPS_PROXY": P}

SAME_RUNTIME_PAIRS = [
    ("node-26.10.0-fetch", "node-26.10.0-http"),
    ("node-24.21.0-fetch", "node-24.21.0-http"),
    ("node-26.10.0-fetch", "undici-8.11.0-EnvHttpProxyAgent"),
    ("node-24.21.0-fetch", "undici-7.29.1-EnvHttpProxyAgent"),
    ("undici-7.29.1-EnvHttpProxyAgent", "undici-8.11.0-EnvHttpProxyAgent"),
    ("node-24.21.0-fetch", "node-26.10.0-fetch"),
    ("curl-8.7.1", "curl-8.22.0"),
    ("python-urllib-3.14.7", "requests-2.34.2"),
    ("python-urllib-3.14.7", "aiohttp-3.14.3"),
    ("httpx-0.28.1", "httpx2-2.13.1"),
]


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def utcnow():
    return dt.datetime.now(dt.timezone.utc)


def iso(t):
    return t.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"


def within(stamp, start, end, slack=1.0):
    """True when an ISO UTC stamp falls inside [start - slack, end + slack]."""
    if stamp is None:
        return True
    parse = lambda v: dt.datetime.strptime(v, "%Y-%m-%dT%H:%M:%S.%fZ")  # noqa: E731
    t = parse(stamp)
    return parse(start) - dt.timedelta(seconds=slack) <= t <= parse(end) + dt.timedelta(seconds=slack)


def sha256_file(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


class Sanitizer:
    """Keep local paths, the account name and the machine name out of outputs."""

    def __init__(self, work):
        self.pairs = [(str(work), "<work>"), (str(HERE), "<lab>"), (str(Path.home()), "~")]
        user = re.escape(getpass.getuser())
        host = socket.gethostname().split(".")[0]
        # Account names appear in home-directory paths; the short machine name can appear anywhere.
        self.patterns = [re.compile(rf"/(Users|home)/{user}\b")]
        if len(host) > 2:
            self.patterns.append(re.compile(rf"\b{re.escape(host)}\b", re.I))

    def __call__(self, value):
        if isinstance(value, str):
            for old, new in self.pairs:
                value = value.replace(old, new)
            for pat in self.patterns:
                value = pat.sub("<redacted>", value)
            return value
        if isinstance(value, dict):
            return {k: self(v) for k, v in value.items()}
        if isinstance(value, list):
            return [self(v) for v in value]
        return value


# ---------------------------------------------------------------------------
# setup
# ---------------------------------------------------------------------------
def node_platform():
    osname = {"darwin": "darwin", "linux": "linux"}.get(sys.platform)
    arch = {"arm64": "arm64", "aarch64": "arm64", "x86_64": "x64", "AMD64": "x64"}.get(platform.machine())
    if not osname or not arch:
        raise SystemExit(f"unsupported platform {sys.platform}/{platform.machine()}")
    return f"{osname}-{arch}"


def download(url, dest):
    with urllib.request.urlopen(url, timeout=120) as r, open(dest, "wb") as f:
        shutil.copyfileobj(r, f)


def setup_node(work, record):
    plat = node_platform()
    dl = work / "downloads"
    dl.mkdir(parents=True, exist_ok=True)
    for v in NODE_VERSIONS:
        name = f"node-{v}-{plat}.tar.gz"
        target = work / "node" / v
        sums_path = dl / f"SHASUMS256-{v}.txt"
        download(f"{NODE_DIST}/{v}/SHASUMS256.txt", sums_path)
        listed = None
        for line in sums_path.read_text().splitlines():
            parts = line.split()
            if len(parts) == 2 and parts[1] == name:
                listed = parts[0]
        if not listed:
            raise SystemExit(f"{name} not listed in SHASUMS256.txt for {v}")
        tar_path = dl / name
        if not tar_path.exists() or sha256_file(tar_path) != listed:
            download(f"{NODE_DIST}/{v}/{name}", tar_path)
        actual = sha256_file(tar_path)
        pinned = NODE_PINNED_SHA256.get(name)
        ok = actual == listed and (pinned is None or pinned == actual)
        record["node"].append({"version": v, "file": name, "sha256": actual, "shasums256_txt": listed,
                               "pinned": pinned, "verified": ok})
        if not ok:
            raise SystemExit(f"checksum mismatch for {name}: {actual} (SHASUMS {listed}, pinned {pinned})")
        if target.exists():
            shutil.rmtree(target)
        target.parent.mkdir(parents=True, exist_ok=True)
        with tarfile.open(tar_path, "r:gz") as tf:
            top = tf.getnames()[0].split("/")[0]
            tf.extractall(target.parent, filter="data")
        (target.parent / top).rename(target)
        print(f"node {v}: sha256 verified ({actual[:16]}...)")


def setup_python(work, python, record):
    venv = work / "venv"
    if venv.exists():
        shutil.rmtree(venv)
    subprocess.run([python, "-m", "venv", str(venv)], check=True)
    pip = [str(venv / "bin" / "python"), "-m", "pip", "--isolated", "--disable-pip-version-check",
           "--no-input", "install", "--cache-dir", str(work / "pip-cache"), "--only-binary", ":all:",
           "--index-url", "https://pypi.org/simple", "-r", str(HERE / "requirements.txt")]
    subprocess.run(pip, check=True)
    freeze = subprocess.run([str(venv / "bin" / "python"), "-m", "pip", "--isolated", "freeze"],
                            capture_output=True, text=True, check=True).stdout
    record["python_freeze"] = freeze.splitlines()
    print("python venv ready")


def node_bin(work, v):
    return work / "node" / v / "bin" / "node"


def setup_npm(work, record):
    npm_dir = work / "npm"
    if npm_dir.exists():
        shutil.rmtree(npm_dir)
    npm_dir.mkdir(parents=True)
    for f in ("package.json", "package-lock.json"):
        shutil.copy2(HERE / f, npm_dir / f)
    empty_rc = work / "empty-npmrc"
    empty_rc.write_text("")
    node = node_bin(work, NODE_VERSIONS[0])
    npm_cli = node.parent.parent / "lib" / "node_modules" / "npm" / "bin" / "npm-cli.js"
    env = {"PATH": f"{node.parent}:/usr/bin:/bin", "HOME": str(work / "home"), "npm_config_update_notifier": "false"}
    (work / "home").mkdir(exist_ok=True)
    subprocess.run([str(node), str(npm_cli), "ci", "--ignore-scripts", "--no-audit", "--no-fund",
                    "--cache", str(work / "npm-cache"), "--userconfig", str(empty_rc),
                    "--registry", "https://registry.npmjs.org/"], cwd=npm_dir, env=env, check=True)
    for alias in ("undici7", "undici8"):
        pkg = json.loads((npm_dir / "node_modules" / alias / "package.json").read_text())
        record["npm"].append({"alias": alias, "name": pkg["name"], "version": pkg["version"]})
    print("npm packages ready")


def setup_go(work, go, record):
    out = work / "bin" / "go_client"
    out.parent.mkdir(parents=True, exist_ok=True)
    env = {"PATH": "/usr/bin:/bin", "HOME": str(work / "home"), "GOCACHE": str(work / "go-cache"),
           "GOPATH": str(work / "gopath"), "GOTOOLCHAIN": "local", "CGO_ENABLED": "0", "GOFLAGS": ""}
    subprocess.run([go, "build", "-trimpath", "-o", str(out), str(HERE / "clients" / "go_client.go")],
                   env=env, check=True)
    record["go"] = subprocess.run([go, "version"], capture_output=True, text=True, env=env).stdout.strip()
    print("go client built")


def cmd_setup(args):
    work = Path(args.work).resolve()
    work.mkdir(parents=True, exist_ok=True)
    (work / "home").mkdir(exist_ok=True)
    record = {"started_utc": iso(utcnow()), "node": [], "npm": []}
    go = args.go or shutil.which("go") or "/opt/homebrew/bin/go"
    setup_node(work, record)
    setup_python(work, args.python, record)
    setup_npm(work, record)
    setup_go(work, go, record)
    record["finished_utc"] = iso(utcnow())
    clean = Sanitizer(work)
    (work / "setup-record.json").write_text(json.dumps(clean(record), indent=2) + "\n")
    print(f"setup complete; record in {work / 'setup-record.json'}")


# ---------------------------------------------------------------------------
# Loopback proxy and origin
# ---------------------------------------------------------------------------
class Lab:
    def __init__(self):
        self.lock = threading.Lock()
        self.current = None
        self.entries = []
        self.seq = 0
        self.t0 = time.monotonic()

    def record(self, server, **fields):
        with self.lock:
            self.seq += 1
            entry = {"seq": self.seq, "t_ms": round((time.monotonic() - self.t0) * 1000, 1),
                     "server": server, "cell": self.current}
            entry.update(fields)
            self.entries.append(entry)
            return entry["cell"]


def read_head(sock, buf=b""):
    while b"\r\n\r\n" not in buf and len(buf) < 65536:
        chunk = sock.recv(4096)
        if not chunk:
            break
        buf += chunk
    head, _, rest = buf.partition(b"\r\n\r\n")
    lines = head.decode("latin-1").split("\r\n")
    parts = lines[0].split(" ") if lines and lines[0] else []
    headers = {}
    for line in lines[1:]:
        k, _, v = line.partition(":")
        headers[k.strip().lower()] = v.strip()
    method, target = (parts[0], parts[1]) if len(parts) >= 2 else ("", "")
    return method, target, headers, rest


def respond(sock, body):
    data = body.encode()
    sock.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: %d\r\n"
                 b"Connection: close\r\n\r\n%s" % (len(data), data))
    wait_for_client_close(sock)


def wait_for_client_close(sock, timeout=3):
    """Let the client close first so TIME_WAIT stays on the client side.

    If the lab servers closed first, their TIME_WAIT sockets could collide with a
    later client's random ephemeral port and stall that connect for about a second.
    """
    sock.settimeout(timeout)
    try:
        while sock.recv(4096):
            pass
    except OSError:
        pass


def reset(sock):
    """Close with RST (SO_LINGER 0): no TIME_WAIT on the lab side."""
    try:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
    except OSError:
        pass


def cell_from_path(target):
    m = re.search(r"/c08/(\d+)", target)
    return int(m.group(1)) if m else None


def make_handler(lab, role):
    class Handler(socketserver.BaseRequestHandler):
        def handle(self):
            sock = self.request
            sock.settimeout(5)
            try:
                method, target, headers, _ = read_head(sock)
                if not method:
                    lab.record(role, kind="empty")
                    reset(sock)
                    return
                host = headers.get("host")
                if role == "origin":
                    cell = lab.record(role, kind="origin", method=method, target=target, host=host,
                                      path_cell=cell_from_path(target))
                    respond(sock, f"C08-ORIGIN cell={cell}\n")
                    return
                if method == "CONNECT":
                    lab.record(role, kind="connect", method=method, target=target, host=host)
                    sock.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n")
                    try:
                        first = sock.recv(1, socket.MSG_PEEK)
                    except socket.timeout:
                        first = b""
                    if first == b"\x16":
                        lab.record(role, kind="tls-in-tunnel", target=target)
                        reset(sock)
                        return
                    if not first:
                        lab.record(role, kind="tunnel-closed", target=target)
                        reset(sock)
                        return
                    m2, t2, h2, _ = read_head(sock)
                    cell = lab.record(role, kind="tunneled", method=m2, target=t2, host=h2.get("host"),
                                      path_cell=cell_from_path(t2))
                    respond(sock, f"C08-PROXY cell={cell}\n")
                    return
                kind = "forward" if re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://", target) else "origin-form"
                cell = lab.record(role, kind=kind, method=method, target=target, host=host,
                                  path_cell=cell_from_path(target))
                respond(sock, f"C08-PROXY cell={cell}\n")
            except (OSError, ValueError) as exc:
                lab.record(role, kind="handler-error", error=type(exc).__name__)

    return Handler


class Server4(socketserver.ThreadingTCPServer):
    daemon_threads = True
    allow_reuse_address = True


class Server6(Server4):
    address_family = socket.AF_INET6


def start_servers(lab):
    proxy = Server4(("127.0.0.1", 0), make_handler(lab, "proxy"))
    for _ in range(50):
        o4 = Server4(("127.0.0.1", 0), make_handler(lab, "origin"))
        port = o4.server_address[1]
        try:
            o6 = Server6(("::1", port), make_handler(lab, "origin"))
            break
        except OSError:
            o4.server_close()
    else:
        raise SystemExit("could not bind the loopback origin on 127.0.0.1 and ::1")
    for srv in (proxy, o4, o6):
        threading.Thread(target=srv.serve_forever, daemon=True).start()
    return proxy, o4, o6


# ---------------------------------------------------------------------------
# Clients
# ---------------------------------------------------------------------------
def first_line(argv):
    try:
        return subprocess.run(argv, capture_output=True, text=True, timeout=20).stdout.splitlines()[0]
    except (OSError, IndexError, subprocess.SubprocessError):
        return ""


def discover_clients(work, args):
    clients = []
    for path in args.curl or DEFAULT_CURLS:
        if not os.path.exists(path):
            print(f"skip curl {path}: not found")
            continue
        line = first_line([path, "--version"])
        ver = line.split()[1]
        clients.append({"id": f"curl-{ver}", "family": "curl", "detail": line,
                        "argv": [path, "-q", "-sS", "-g", "--max-time", "5", "--connect-timeout", "3", "-o", "-"],
                        "json": False, "env": {}})
    wget = args.wget or (DEFAULT_WGET if os.path.exists(DEFAULT_WGET) else shutil.which("wget"))
    if wget and os.path.exists(wget):
        line = first_line([wget, "--version"])
        ver = line.split()[2]
        clients.append({"id": f"wget-{ver}", "family": "wget", "detail": line,
                        "argv": [wget, "--no-config", "--no-hsts", "--tries=1", "--timeout=5", "-nv", "-O", "-"],
                        "json": False, "env": {}})
    go_bin = work / "bin" / "go_client"
    info = json.loads(first_line([str(go_bin), "--version"]))
    clients.append({"id": f"go-{info['go'].removeprefix('go')}-net-http", "family": "go",
                    "detail": f"{info['go']} net/http DefaultTransport (ProxyFromEnvironment)",
                    "argv": [str(go_bin)], "json": True, "env": {}})
    py = str(work / "venv" / "bin" / "python")
    for lib in PY_LIBS:
        info = json.loads(first_line([py, str(HERE / "clients" / "py_client.py"), lib, "--version"]))
        cid = f"python-urllib-{info['version']}" if lib == "urllib" else f"{lib}-{info['version']}"
        detail = f"{lib} {info['version']} on Python {info['python']}" + (" (trust_env=True)" if lib == "aiohttp" else "")
        clients.append({"id": cid, "family": "python", "detail": detail,
                        "argv": [py, str(HERE / "clients" / "py_client.py"), lib], "json": True, "env": {}})
    script = str(HERE / "clients" / "node_client.cjs")
    for v in NODE_VERSIONS:
        node = str(node_bin(work, v))
        for mode in ("fetch", "http"):
            info = json.loads(first_line([node, script, mode, "--version"]))
            base = f"node-{info['node']}-{mode}"
            what = "built-in fetch()" if mode == "fetch" else "http.get()/https.get()"
            clients.append({"id": base, "family": "node", "detail":
                            f"Node {info['node']} {what}, NODE_USE_ENV_PROXY=1, bundled undici {info['bundled_undici']}",
                            "argv": [node, script, mode], "json": True, "env": {"NODE_USE_ENV_PROXY": "1"}})
            clients.append({"id": base + "-noflag", "family": "node-control", "detail":
                            f"Node {info['node']} {what} without NODE_USE_ENV_PROXY (control)",
                            "argv": [node, script, mode], "json": True, "env": {}, "only": ["b-none"]})
    node26 = str(node_bin(work, NODE_VERSIONS[0]))
    for alias in ("undici7", "undici8"):
        path = str(work / "npm" / "node_modules" / alias)
        info = json.loads(first_line_env([node26, script, "undici", "--version"], {"C08_UNDICI_PATH": path}))
        clients.append({"id": f"undici-{info['npm_undici']}-EnvHttpProxyAgent", "family": "undici",
                        "detail": f"npm undici {info['npm_undici']} fetch() with new EnvHttpProxyAgent() on Node {info['node']}",
                        "argv": [node26, script, "undici"], "json": True, "env": {"C08_UNDICI_PATH": path}})
    return clients


def first_line_env(argv, extra):
    env = dict(os.environ)
    env.update(extra)
    return subprocess.run(argv, capture_output=True, text=True, timeout=20, env=env).stdout.splitlines()[0]


def body_marker(text):
    if "C08-PROXY" in text:
        return "proxy"
    if "C08-ORIGIN" in text:
        return "origin"
    return None


# ---------------------------------------------------------------------------
# run
# ---------------------------------------------------------------------------
def url_shape(url):
    u = urllib.parse.urlsplit(url)
    host = u.hostname or ""
    netloc_host = u.netloc.rsplit("@", 1)[-1]
    if netloc_host.startswith("["):
        kind = "ipv6"
    elif re.fullmatch(r"[0-9.]+", host):
        kind = "ipv4"
    else:
        kind = "name"
    port = ":port" if re.search(r":\d+$", netloc_host) else ""
    extra = ""
    if host.endswith("."):
        extra = "+trailing-dot"
    elif re.search(r"[A-Z]", netloc_host):
        extra = "+uppercase"
    return f"{u.scheme}:{kind}{port}{extra}"


def base_env(work):
    return {"PATH": "/usr/bin:/bin:/usr/sbin:/sbin", "HOME": str(work / "home"), "TMPDIR": str(work / "tmp") + "/",
            "LANG": "C", "LC_ALL": "C", "PYTHONNOUSERSITE": "1", "PYTHONDONTWRITEBYTECODE": "1"}


def sandbox_selftest(prefix, work, proxy_port):
    code = ("import socket,sys\n"
            "s=socket.socket(); s.settimeout(3)\n"
            "try:\n s.connect(('192.0.2.1',9)); print('EGRESS-ALLOWED'); sys.exit(3)\n"
            "except PermissionError: print('egress-denied')\n"
            "except OSError as e: print('egress-error', e.errno)\n"
            f"c=socket.create_connection(('127.0.0.1',{proxy_port}),timeout=3); c.close(); print('loopback-ok')\n")
    py = str(work / "venv" / "bin" / "python")
    r = subprocess.run(prefix + [py, "-c", code], capture_output=True, text=True, env=base_env(work))
    return r.stdout.split()


def collect_denials(start, end):
    fmt = "%Y-%m-%d %H:%M:%S"
    argv = ["/usr/bin/log", "show", "--style", "ndjson", "--start", time.strftime(fmt, time.localtime(start - 2)),
            "--end", time.strftime(fmt, time.localtime(end + 3)), "--predicate",
            'eventMessage CONTAINS "deny(1) network-outbound"']
    try:
        out = subprocess.run(argv, capture_output=True, text=True, timeout=300).stdout
    except (OSError, subprocess.SubprocessError) as exc:
        return None, f"log show failed: {exc}"
    denials = []
    for line in out.splitlines():
        try:
            obj = json.loads(line)
        except ValueError:
            continue
        msg = obj.get("eventMessage", "")
        m = re.search(r"Sandbox: (.+?)\((\d+)\) deny\(\d+\) network-outbound (.+)$", msg)
        if not m:
            continue
        try:
            ts = dt.datetime.strptime(obj["timestamp"], "%Y-%m-%d %H:%M:%S.%f%z").astimezone(dt.timezone.utc)
        except (KeyError, ValueError):
            ts = None
        target = m.group(3).strip()
        target = "dns-resolver-socket" if target.endswith("/mDNSResponder") else target
        denials.append({"process": m.group(1), "pid": int(m.group(2)), "target": target,
                        "utc": iso(ts) if ts else None})
    return denials, None


def cmd_run(args):
    work = Path(args.work).resolve()
    out = Path(args.out).resolve()
    if out.exists() and any(out.iterdir()):
        raise SystemExit(f"{out} is not empty; choose a new --out directory")
    out.mkdir(parents=True, exist_ok=True)
    (work / "tmp").mkdir(exist_ok=True)
    (work / "home").mkdir(exist_ok=True)
    clean = Sanitizer(work)

    if args.isolation == "sandbox-exec":
        if not os.path.exists("/usr/bin/sandbox-exec"):
            raise SystemExit("sandbox-exec not found; this isolation mode needs macOS")
        prefix = ["/usr/bin/sandbox-exec", "-f", str(SANDBOX_PROFILE)]
    else:
        prefix = []
        print("WARNING: --isolation none: direct attempts will leave this machine (DNS lookups for .test "
              "names, TCP SYNs to documentation addresses). Use only inside your own network isolation.")

    clients = discover_clients(work, args)
    lab = Lab()
    proxy, o4, _ = start_servers(lab)
    proxy_url = f"http://127.0.0.1:{proxy.server_address[1]}"
    origin_port = o4.server_address[1]

    lab.current = "selftest"
    selftest = sandbox_selftest(prefix, work, proxy.server_address[1]) if prefix else ["not-run"]
    time.sleep(0.1)
    lab.current = None
    if prefix and selftest != ["egress-denied", "loopback-ok"]:
        raise SystemExit(f"sandbox self-test failed: {selftest}")

    cells = []
    n = 0
    run_start = time.time()
    started = utcnow()
    for case in CASES:
        if args.case and case["id"] not in args.case:
            continue
        for client in clients:
            if "only" in client and case["id"] not in client["only"]:
                continue
            if args.client and client["id"] not in args.client:
                continue
            for template in case["urls"]:
                n += 1
                url = template.format(origin=origin_port) + f"c08/{n}"
                env = base_env(work)
                env.update(client["env"])
                pv = case.get("proxy_vars", DEFAULT_PROXY_VARS)
                env.update({k: v.replace(P, proxy_url) for k, v in pv.items()})
                env.update(case["vars"])
                argv = prefix + client["argv"] + [url]
                lab.current = n
                t_start = utcnow()
                t0 = time.monotonic()
                try:
                    proc = subprocess.Popen(argv, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                            cwd=str(work / "home"))
                    pid = proc.pid
                    try:
                        stdout, stderr = proc.communicate(timeout=25)
                        timed_out = False
                    except subprocess.TimeoutExpired:
                        proc.kill()
                        stdout, stderr = proc.communicate()
                        timed_out = True
                    rc = proc.returncode
                except OSError as exc:
                    pid, stdout, stderr, rc, timed_out = None, b"", str(exc).encode(), None, False
                duration = round((time.monotonic() - t0) * 1000, 1)
                time.sleep(0.03)
                lab.current = None
                t_end = utcnow()
                stdout_s = stdout.decode("utf-8", "replace")
                stderr_s = stderr.decode("utf-8", "replace")
                if client["json"]:
                    try:
                        outcome = json.loads(stdout_s.strip().splitlines()[-1])
                    except (ValueError, IndexError):
                        outcome = {"ok": False, "status": None, "marker": None, "error": "no JSON output"}
                else:
                    outcome = {"ok": rc == 0, "status": None, "marker": body_marker(stdout_s),
                               "error": stderr_s.strip()[:300] or None}
                cells.append({
                    "n": n, "client": client["id"], "family": client["family"], "case": case["id"],
                    "section": case["section"], "label": case["label"], "url": template,
                    "shape": url_shape(template.format(origin=origin_port)),
                    "vars": case["vars"],
                    "proxy_vars": {k: ("<proxy>" if v == P else v) for k, v in pv.items()},
                    "pid": pid, "exit": rc, "timed_out": timed_out, "duration_ms": duration,
                    "started_utc": iso(t_start), "ended_utc": iso(t_end),
                    "outcome": outcome, "stderr_tail": stderr_s.strip()[-300:],
                })
                if n % 100 == 0:
                    print(f"{n} cells...", flush=True)
    run_end = time.time()
    finished = utcnow()
    for srv in (proxy, o4):
        srv.shutdown()

    if prefix:
        time.sleep(3)  # let the unified log flush kernel sandbox reports
    denials, denial_error = (collect_denials(run_start, run_end) if prefix else (None, "not collected"))
    by_pid = {}
    for d in denials or []:
        by_pid.setdefault(d["pid"], []).append(d)

    by_cell = {}
    strays = []
    for e in lab.entries:
        if e["cell"] == "selftest":
            continue
        if e["cell"] is None:
            strays.append(e)
        else:
            by_cell.setdefault(e["cell"], []).append(e)

    for c in cells:
        entries = by_cell.get(c["n"], [])
        prox = [e for e in entries if e["server"] == "proxy"]
        orig = [e for e in entries if e["server"] == "origin"]
        mism = [e for e in entries if e.get("path_cell") not in (None, c["n"])]
        dn = [d for d in by_pid.get(c["pid"], []) if within(d["utc"], c["started_utc"], c["ended_utc"])]
        c["route"] = "proxy" if prox else "not_proxied"
        c["proxy_requests"] = [{"kind": e["kind"], "method": e.get("method"), "target": e.get("target"),
                                "host_header": e.get("host")} for e in prox]
        c["origin_requests"] = len(orig)
        c["sandbox_denials"] = sorted({d["target"] for d in dn})
        text = (c["outcome"].get("error") or "") + " " + c["stderr_tail"]
        sources = []
        if orig:
            sources.append("origin-log")
        if c["outcome"].get("marker") == "origin":
            sources.append("origin-marker")
        if dn:
            sources.append("sandbox-denial")
        if DIRECT_ERROR.search(text) and not PROXY_FAILURE.search(text):
            sources.append("client-error")
        c["direct_evidence_sources"] = sources if c["route"] == "not_proxied" else []
        if c["route"] == "proxy":
            c["direct_evidence"] = None
        else:
            c["direct_evidence"] = sources[0] if sources else "none"
        c["attribution_mismatch"] = len(mism)
        # Display class. The proxy log alone decides proxy vs not proxied; the evidence
        # only separates a direct attempt from a client that raised before any request.
        c["observed"] = ("proxy" if c["route"] == "proxy"
                         else "direct" if c["direct_evidence"] != "none" else "no_request")

    results = build_results(cells, clients, strays, denials, denial_error, selftest, started, finished, work, args)
    results = clean(results)
    (out / "results.json").write_text(json.dumps(results, indent=1, ensure_ascii=False) + "\n")
    write_csv(out / "results.csv", results)
    raw = out / "raw"
    raw.mkdir()
    with open(raw / "lab-log.jsonl", "w") as f:
        for e in lab.entries:
            f.write(json.dumps(clean(e)) + "\n")
    with open(raw / "sandbox-denials.jsonl", "w") as f:
        for d in denials or []:
            f.write(json.dumps(clean(d)) + "\n")
    s = results["summary"]["counts"]
    print(json.dumps(s, indent=1))
    print(f"wrote {out / 'results.json'} and results.csv")


# ---------------------------------------------------------------------------
# Analysis
# ---------------------------------------------------------------------------
def build_results(cells, clients, strays, denials, denial_error, selftest, started, finished, work, args):
    main_ids = [c["id"] for c in clients if c["family"] != "node-control"]
    idx = {(c["client"], c["case"], c["url"]): c for c in cells}

    # Baseline shapes that each client really sent through the proxy.
    base_ok = {}
    for c in cells:
        if c["section"] == "baseline" and c["case"] == "b-none":
            base_ok[(c["client"], c["shape"])] = c["route"] == "proxy"
        if c["section"] == "https" and c["case"] == "h-none":
            base_ok[(c["client"], c["shape"])] = c["route"] == "proxy"
    for c in cells:
        if c["section"] in ("main", "variables", "https") and c["case"] != "h-none":
            c["baseline_proxied"] = base_ok.get((c["client"], c["shape"]))
        else:
            c["baseline_proxied"] = None

    rows = []
    for case in CASES:
        if case["section"] not in ("main", "variables", "loopback", "https", "baseline"):
            continue
        for url in case["urls"]:
            routes = {cid: idx[(cid, case["id"], url)]["observed"] for cid in main_ids if (cid, case["id"], url) in idx}
            valid = {cid: r for cid, r in routes.items()
                     if idx[(cid, case["id"], url)].get("baseline_proxied") is not False}
            vals = set(valid.values())
            rows.append({
                "case": case["id"], "section": case["section"], "label": case["label"],
                "vars": case["vars"], "url": url,
                "routes": routes,
                "excluded_baseline_not_proxied": sorted(set(routes) - set(valid)),
                "agreement": len(vals) == 1 and len(valid) == len(main_ids),
                "proxy": sorted(k for k, v in valid.items() if v == "proxy"),
                "direct": sorted(k for k, v in valid.items() if v == "direct"),
                "no_request": sorted(k for k, v in valid.items() if v == "no_request"),
            })

    main_rows = [r for r in rows if r["section"] == "main"]
    portable = [{"case": r["case"], "entry": r["vars"].get("NO_PROXY"), "url": r["url"],
                 "observed": next(iter(set(r["routes"].values())))} for r in main_rows if r["agreement"]]
    divergent = [{"case": r["case"], "entry": r["vars"].get("NO_PROXY"), "url": r["url"],
                  "proxy": r["proxy"], "direct": r["direct"], "no_request": r["no_request"],
                  "excluded": r["excluded_baseline_not_proxied"]} for r in main_rows if not r["agreement"]]
    case_level = []
    for case in CASES:
        if case["section"] != "main":
            continue
        rs = [r for r in main_rows if r["case"] == case["id"]]
        case_level.append({"case": case["id"], "entry": case["vars"].get("NO_PROXY"),
                           "all_urls_agree": all(r["agreement"] for r in rs),
                           "urls": {r["url"]: next(iter(set(r["routes"].values()))) if r["agreement"] else "DIVERGES"
                                    for r in rs}})

    pairs = []
    for a, b in SAME_RUNTIME_PAIRS:
        if a not in main_ids or b not in main_ids:
            pairs.append({"a": a, "b": b, "status": "client missing"})
            continue
        diffs = []
        for r in rows:
            if r["section"] == "baseline":
                continue
            ra, rb = r["routes"].get(a), r["routes"].get(b)
            if ra and rb and ra != rb:
                diffs.append({"case": r["case"], "vars": r["vars"], "url": r["url"], a: ra, b: rb})
        pairs.append({"a": a, "b": b, "differences": diffs})

    https_check = []
    for case in CASES:
        if case["section"] != "https":
            continue
        for url in case["urls"]:
            tcase, turl = case["twins"][url]
            for cid in main_ids:
                c, t = idx.get((cid, case["id"], url)), idx.get((cid, tcase, turl))
                if c and t:
                    https_check.append({"client": cid, "case": case["id"], "url": url, "observed": c["observed"],
                                        "http_twin": f"{tcase} {turl}", "http_observed": t["observed"],
                                        "same": c["observed"] == t["observed"],
                                        "proxy_kinds": sorted({p["kind"] for p in c["proxy_requests"]})})

    controls = [{"client": c["client"], "url": c["url"], "observed": c["observed"]} for c in cells
                if c["family"] == "node-control"]

    http_proxy_modes = {}
    for c in cells:
        if c["route"] == "proxy" and c["url"].startswith("http:") and c["family"] != "node-control":
            kinds = tuple(sorted({p["kind"] for p in c["proxy_requests"]}))
            http_proxy_modes.setdefault(c["client"], set()).add("+".join(kinds))
    http_proxy_modes = {k: sorted(v) for k, v in http_proxy_modes.items()}

    anomalies = {
        "strays": strays,
        "not_proxied_without_direct_evidence (client raised before sending)": [
            {"n": c["n"], "client": c["client"], "case": c["case"], "url": c["url"], "error": c["outcome"].get("error")}
            for c in cells if c["route"] == "not_proxied" and c["direct_evidence"] == "none"],
        "proxied_http_without_proxy_marker": [
            {"n": c["n"], "client": c["client"], "case": c["case"], "url": c["url"], "error": c["outcome"].get("error")}
            for c in cells if c["route"] == "proxy" and c["url"].startswith("http:") and c["outcome"].get("marker") != "proxy"],
        "proxied_with_sandbox_denial": [
            {"n": c["n"], "client": c["client"], "case": c["case"], "url": c["url"], "denials": c["sandbox_denials"]}
            for c in cells if c["route"] == "proxy" and c["sandbox_denials"]],
        "attribution_mismatch": [c["n"] for c in cells if c["attribution_mismatch"]],
        "timeouts": [c["n"] for c in cells if c["timed_out"]],
        "baseline_not_proxied": sorted({f"{k[0]} {k[1]}" for k, v in base_ok.items()
                                        if not v and k[0] in main_ids}),
    }
    counts = {
        "cells": len(cells),
        "clients": len(main_ids),
        "control_clients": len([c for c in clients if c["family"] == "node-control"]),
        "proxy": sum(c["route"] == "proxy" for c in cells),
        "not_proxied": sum(c["route"] == "not_proxied" for c in cells),
        "observed": {k: sum(c["observed"] == k for c in cells) for k in ("proxy", "direct", "no_request")},
        "not_proxied_by_primary_evidence": {
            k: sum(1 for c in cells if c["route"] == "not_proxied" and c["direct_evidence"] == k)
            for k in ("origin-log", "origin-marker", "sandbox-denial", "client-error", "none")},
        "not_proxied_evidence_source_counts": {
            k: sum(1 for c in cells if k in c["direct_evidence_sources"])
            for k in ("origin-log", "origin-marker", "sandbox-denial", "client-error")},
        "main_rows": len(main_rows),
        "main_rows_all_agree": len(portable),
        "main_rows_diverge": len(divergent),
        "https_spot_checks": len(https_check),
        "https_same_as_http_twin": sum(h["same"] for h in https_check),
        "anomalies": {k: len(v) for k, v in anomalies.items()},
    }
    env = {
        "os": f"macOS {platform.mac_ver()[0]}" if sys.platform == "darwin" else platform.platform(),
        "arch": platform.machine(),
        "harness_python": platform.python_version(),
        "isolation": args.isolation,
        "sandbox_profile_sha256": sha256_file(SANDBOX_PROFILE) if SANDBOX_PROFILE.exists() else None,
        "sandbox_selftest": selftest,
        "sandbox_denial_log": denial_error or f"{len(denials or [])} kernel sandbox denial lines collected",
        "files_sha256": {name: sha256_file(HERE / name) for name in HARNESS_FILES if (HERE / name).exists()},
    }
    setup_record = work / "setup-record.json"
    if setup_record.exists():
        env["setup_record"] = json.loads(setup_record.read_text())
    return {
        "title": "NO_PROXY matching matrix (loopback lab, synthetic)",
        "scope": ("Synthetic loopback observations of how each client routes .test names and documentation "
                  "addresses when NO_PROXY is set. A cell is 'proxy' when the lab proxy logged a request from it, "
                  "otherwise 'not_proxied'. No real proxy, provider or third-party host was contacted."),
        "run": {"started_utc": iso(started), "finished_utc": iso(finished)},
        "environment": env,
        "clients": [{"id": c["id"], "family": c["family"], "detail": c["detail"]} for c in clients],
        "cases": CASES,
        "summary": {
            "counts": counts,
            "portable_rows": portable,
            "divergent_rows": divergent,
            "case_level": case_level,
            "same_runtime_pairs": pairs,
            "https_vs_http": https_check,
            "node_opt_in_controls": controls,
            "http_target_proxy_mode": http_proxy_modes,
            "anomalies": anomalies,
        },
        "rows": rows,
        "cells": cells,
    }


def write_csv(path, results):
    cols = ["n", "client", "section", "case", "label", "no_proxy_vars", "proxy_vars", "url", "shape", "route",
            "observed", "direct_evidence_sources", "proxy_requests", "baseline_proxied", "sandbox_denials", "exit", "marker", "error",
            "duration_ms"]
    with open(path, "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(cols)
        for c in results["cells"]:
            w.writerow([
                c["n"], c["client"], c["section"], c["case"], c["label"],
                json.dumps(c["vars"], ensure_ascii=False),
                "" if c["proxy_vars"] == {k: "<proxy>" for k in DEFAULT_PROXY_VARS} else json.dumps(c["proxy_vars"]),
                c["url"], c["shape"], c["route"], c["observed"],
                "+".join(c["direct_evidence_sources"]) or (c["direct_evidence"] or ""),
                "; ".join(f"{p['kind']} {p['target']}" for p in c["proxy_requests"]),
                "" if c["baseline_proxied"] is None else c["baseline_proxied"],
                "; ".join(c["sandbox_denials"]), c["exit"], c["outcome"].get("marker") or "",
                (c["outcome"].get("error") or "")[:200], c["duration_ms"],
            ])


# ---------------------------------------------------------------------------
# compare
# ---------------------------------------------------------------------------
def cmd_compare(args):
    """Compare two results.json files cell by cell (timing fields are ignored)."""
    a = json.loads(Path(args.old).read_text())
    b = json.loads(Path(args.new).read_text())
    key = lambda c: (c["client"], c["case"], c["url"])  # noqa: E731
    ca = {key(c): c for c in a["cells"]}
    cb = {key(c): c for c in b["cells"]}
    common = sorted(set(ca) & set(cb))
    route = [list(k) + [ca[k]["route"], cb[k]["route"]] for k in common if ca[k]["route"] != cb[k]["route"]]
    both_observed = [k for k in common if "observed" in ca[k] and "observed" in cb[k]]
    observed = [list(k) + [ca[k]["observed"], cb[k]["observed"]] for k in both_observed
                if ca[k]["observed"] != cb[k]["observed"]]
    evidence = [list(k) + [ca[k].get("direct_evidence"), cb[k].get("direct_evidence")] for k in common
                if ca[k].get("direct_evidence") != cb[k].get("direct_evidence")]
    report = {
        "compared_cells": len(common),
        "route_differences": route,
        "observed_compared": len(both_observed),
        "observed_differences": observed,
        "only_in_old": [list(k) for k in sorted(set(ca) - set(cb))],
        "only_in_new": [list(k) for k in sorted(set(cb) - set(ca))],
        "primary_evidence_differences (informational)": evidence,
    }
    print(json.dumps(report, indent=1))
    same = not route and not observed and not report["only_in_old"] and not report["only_in_new"]
    print(f"route: {len(common) - len(route)}/{len(common)} identical; "
          f"observed: {len(both_observed) - len(observed)}/{len(both_observed)} identical; "
          f"cells only in old/new: {len(report['only_in_old'])}/{len(report['only_in_new'])}")
    print("IDENTICAL classifications" if same else "CLASSIFICATIONS DIFFER")
    return 0 if same else 1


# ---------------------------------------------------------------------------
# table
# ---------------------------------------------------------------------------
SHORT = [("curl-8.7.1", "curl 8.7.1"), ("curl-8.22.0", "curl 8.22"), ("wget-", "wget"), ("go-", "Go"),
         ("python-urllib-", "urllib"), ("requests-", "requests"), ("httpx-", "httpx"), ("httpx2-", "httpx2"),
         ("aiohttp-", "aiohttp"), ("node-26", "N26 fetch"), ("node-24", "N24 fetch"),
         ("undici-7", "undici 7"), ("undici-8", "undici 8.11")]


def short_name(cid):
    for prefix, name in SHORT:
        if cid.startswith(prefix):
            if cid.startswith("node-") and cid.endswith("-http"):
                return name.replace("fetch", "http")
            return name
    return cid


def cmd_table(args):
    """Print the observed matrix as Markdown: one row per (NO_PROXY value, request URL)."""
    r = json.loads(Path(args.results).read_text())
    clients = [c["id"] for c in r["clients"] if c["family"] != "node-control"]
    mark = {"proxy": "proxy", "direct": "**direct**", "no_request": "ERROR"}
    head = "| Section | NO_PROXY / no_proxy | Request URL | " + " | ".join(short_name(c) for c in clients) + " |"
    print(f"Observed {r['run']['started_utc'][:10]} ({r['environment']['os']}). proxy = the lab proxy logged the "
          "request; **direct** = not proxied and a direct attempt was recorded; ERROR = the client raised before "
          "sending anything.\n")
    print(head)
    print("|" + "---|" * (3 + len(clients)))
    for row in r["rows"]:
        if args.section and row["section"] not in args.section:
            continue
        v = row["vars"]
        if v.get("NO_PROXY") == v.get("no_proxy") and set(v) == {"NO_PROXY", "no_proxy"}:
            entry = f"`{v['NO_PROXY']}`" if v["NO_PROXY"].strip() == v["NO_PROXY"] else f"`\"{v['NO_PROXY']}\"`"
        elif not v:
            case = next(c for c in r["cases"] if c["id"] == row["case"])
            entry = "(unset)" if "proxy_vars" not in case else "`http_proxy=\"\"`, `HTTP_PROXY` set"
        else:
            entry = ", ".join(f"`{k}={val!r}`" for k, val in v.items())
        cells = [mark[row["routes"].get(c, "")] if row["routes"].get(c) else "-" for c in clients]
        note = "" if row["agreement"] else " (differs)"
        print(f"| {row['section']} | {entry} | `{row['url']}`{note} | " + " | ".join(cells) + " |")
    return 0


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)
    s = sub.add_parser("setup")
    s.add_argument("--work", required=True)
    s.add_argument("--python", default=sys.executable, help="Python used for the venv (default: this interpreter)")
    s.add_argument("--go", default=None)
    r = sub.add_parser("run")
    r.add_argument("--work", required=True)
    r.add_argument("--out", required=True)
    r.add_argument("--curl", action="append", help="curl binary (repeatable)")
    r.add_argument("--wget", default=None)
    r.add_argument("--isolation", choices=["sandbox-exec", "none"], default="sandbox-exec")
    r.add_argument("--case", action="append", help="run only this case id (repeatable; for debugging)")
    r.add_argument("--client", action="append", help="run only this client id (repeatable; for debugging)")
    c = sub.add_parser("compare")
    c.add_argument("old")
    c.add_argument("new")
    t = sub.add_parser("table")
    t.add_argument("results")
    t.add_argument("--section", action="append", help="baseline, main, variables, loopback or https (repeatable)")
    args = ap.parse_args()
    if args.cmd == "table":
        return cmd_table(args)
    if args.cmd == "setup":
        return cmd_setup(args)
    if args.cmd == "run":
        return cmd_run(args)
    return cmd_compare(args)


if __name__ == "__main__":
    sys.exit(main())