diff --git a/benchmarks/README.md b/benchmarks/README.md
index c5788477..71dc35ea 100644
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -170,6 +170,12 @@ result is not a native-engine speedup. Metal and vision benchmarks follow CUDA.
 python -m unittest discover -s tests/benchmarks -p 'test_*.py' -v
 ```
 
+## Frozen serving matrix
+
+See [the matrix client guide](serving_matrix.md) and [example plan](serving_matrix.example.json)
+for exact response oracles, workload variants, concurrency rounds, pinned runtime
+metadata and environment-based bearer authentication.
+
 ## Laya on Apple Silicon
 
 [`laya_mps/`](laya_mps/README.md) holds the scripts behind the numbers of the
diff --git a/benchmarks/serving_matrix.example.json b/benchmarks/serving_matrix.example.json
new file mode 100644
index 00000000..9859dc78
--- /dev/null
+++ b/benchmarks/serving_matrix.example.json
@@ -0,0 +1,43 @@
+{
+  "endpoint": "http://127.0.0.1:8000/v1/systemone",
+  "health_endpoint": "http://127.0.0.1:8000/health",
+  "cases": [
+    {
+      "name": "two-decisions",
+      "request": {
+        "model": "test"
+      },
+      "expected_response": {
+        "answers": {
+          "a": {
+            "choice": "yes"
+          },
+          "b": {
+            "noul": 0.75
+          }
+        }
+      }
+    }
+  ],
+  "concurrency": [
+    1,
+    2
+  ],
+  "requests_per_case": 4,
+  "repetitions": 2,
+  "warmup_per_case": 2,
+  "timeout_seconds": 2,
+  "metadata": {
+    "gpu": "N/A: synthetic local CPU server",
+    "gpu_ids": "N/A",
+    "driver": "N/A",
+    "cuda": "N/A",
+    "precision": "N/A",
+    "model_revision": "synthetic-test-v1",
+    "runtime_revision": "local-test-server-v1",
+    "cache_policy": "none",
+    "cuda_evidence": "N/A: no CUDA execution",
+    "reservation": "N/A: local CPU fixture",
+    "host": "127.0.0.1: synthetic CPU fixture"
+  }
+}
diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
new file mode 100644
index 00000000..4c64f88e
--- /dev/null
+++ b/benchmarks/serving_matrix.md
@@ -0,0 +1,166 @@
+# Frozen HTTP serving matrix
+
+`serving_matrix.py` is a Python 3.11+ standard-library client for an already
+running worker or frontend. It measures closed-loop HTTP concurrency and checks
+every response against a frozen exact oracle. It does not launch models or reserve
+a GPU. The [example plan](serving_matrix.example.json) uses synthetic CPU-server
+responses and explicitly records N/A for GPU fields. Replace its requests,
+oracles, endpoints and identity with your deployment before measuring.
+
+```sh
+python3 benchmarks/serving_matrix.py /path/to/frozen-plan.json \
+  --output /tmp/matrix-new-run
+```
+
+The output directory must not exist. Validation precedes output creation and
+network requests. Exit 0 means all phases passed; exit 1 means an HTTP,
+correctness, timeout or transport failure was recorded.
+
+## Plan schema
+
+- `endpoint`: HTTP(S) inference URL, without URL credentials or fragments.
+- `health_endpoint`: optional GET URL; defaults to `/health` on the inference
+  origin. Health requires HTTP 200; its body is retained for inspection.
+- `cases`: nonempty list with unique nonempty `name` values. Each case contains a
+  JSON-object `request` and exactly one of `expected_response` (type-strict JSON)
+  or `expected_response_text` (exact UTF-8 bytes, including whitespace).
+- Alternatively a case has a nonempty `variants` list instead of request/oracle
+  fields. Every variant has a unique `name`, request and oracle. Waves cycle
+  variants by dispatch index, preserving planned order across configurations.
+- `requests_per_case`: positive integer, applied per measured round.
+- `concurrency`: unique positive integers no greater than `requests_per_case`.
+- `repetitions` and `warmup_per_case`: integers >= 2, both default to 2 when
+  omitted. The frozen plan predeclares the budget; do not extend it after seeing
+  results. Effective defaults are recorded in `config.json`.
+- `timeout_seconds`: finite positive deadline for a complete exchange.
+- `metadata`: requires nonempty `gpu`, `gpu_ids`, `driver`, `cuda`, `precision`,
+  `model_revision`, `runtime_revision`, `cache_policy`, `cuda_evidence` and
+  `reservation` and `host` (server node). Record pinned revisions, the actual
+  reserved device and node, not just a shared service URL. CPU
+  fixtures must explicitly record N/A with a reason. The client checks presence;
+  it does not independently verify these declarations or prove CUDA execution.
+
+JSON numbers must be finite. Type-strict matching distinguishes `true` from `1`
+and integers from floats, preserves array order and ignores object-key order.
+Use exact-byte matching when formatting matters. The runner never alters model
+aliases, injects token counts, truncates prompts or invents oracles. Capture a
+correct response before freezing the plan.
+
+## Responses with dynamic timing fields
+
+The native Open-Jev response includes per-request `metadata.inference_seconds`;
+JEMM includes `usage.latency_ms`. A captured whole-body oracle cannot match these
+values on later requests. Declare `ignore_paths` at the plan's top level to
+exclude only those volatile leaf values from JSON comparison:
+
+```json
+{"ignore_paths": ["/metadata/inference_seconds"]}
+```
+
+For JEMM use `/usage/latency_ms` instead. Paths are RFC 6901 JSON Pointers; `/`
+inside a key is encoded `~1`, `~` is `~0`, and arrays use canonical zero-based
+indices. The list must be unique. Every path must resolve in every variant's
+oracle to an existing primitive leaf. Root, object/array exclusions, malformed
+escapes and invalid indices are rejected before output/network access.
+
+Excluded fields must still exist in the actual response with the same JSON type.
+All other fields remain type-strict and exact, including model identity, answers,
+token usage, key sets and array order. Declare only volatile diagnostics; do not
+exclude decisions or token counts to conceal disagreement. No exclusions are
+inferred automatically. Exclusions are incompatible with exact-byte oracles.
+The list is bound by `plan_sha256`; raw response bytes and the oracle snapshot
+remain unmodified, so the original values can still be audited.
+
+## Authentication
+
+Set `OMNI_JEV_TEST_TOKEN` in the environment to send `Authorization: Bearer …`
+on health and every inference phase, matching `bench.py`. It must be printable
+ASCII and is read once before dispatch. Keep credentials out of the plan and
+metadata; the client does not save request headers or the token. An empty/unset
+variable sends no Authorization header. Redirects, environment proxies, pooling
+and retries are disabled. Use HTTPS for authenticated remote endpoints.
+
+## Phases and metrics
+
+One health request precedes one readiness request per variant, the predeclared
+excluded warmups per variant, then one excluded feasibility wave at each concurrency
+and the predeclared measured rounds (default two). Each worker sends its next request after the previous
+exchange completes. There is no arrival-rate or queue-delay simulation. A failure
+stops new dispatch while already admitted work drains.
+
+Client wall time includes DNS/TCP/TLS connection establishment, request serialization,
+complete response read, oracle validation and connection close. Each request opens
+a fresh connection; `connect_seconds` reports the connect call separately (null
+if it fails). TLS handshakes are included. This policy differs from `bench.py`'s
+keep-alive pool, so their latency/throughput figures are not directly comparable.
+Evidence encoding and sink writes affect client dispatch and complete-wave elapsed
+time, but are outside the recorded exchange latency.
+
+The complete-exchange budget is checked after connection and again after validation
+and close; an over-budget exchange is classified as a timeout even if the watchdog
+fires late. Raw status and body remain available, and timed-out decisions do not
+count as successes. This is not a hard wall-clock return bound: system DNS lookup,
+multiple-address connection attempts, JSON validation and close are not preempted
+by the body-read watchdog. They can finish after the budget and are then rejected.
+
+Round requests/s and decisions/s use actual successful counts
+and whole-round elapsed time; planned, attempted and successful denominators are
+separate. Decisions count elements of `answers`. Successful p50/p95 use nearest
+rank; **p95 is the maximum when fewer than 20 samples are present**, including the
+four-sample example. Failure latencies and kinds are retained separately. First-use, capture and
+loading costs must be evaluated separately from warmed results.
+
+Whole-wave rates are explicitly labelled `complete_wave_including_ramp_and_drain`.
+For a complete measured round with n > 2c, `steady_state` additionally reports an
+interior completion window: sort retained monotonic completion timestamps, exclude
+the first and last c completions, and measure from the c-th completion boundary
+to the last retained completion. Count only the n−2c interior requests/decisions.
+Short (n <= 2c), failed or zero-duration windows have `available: false` and a reason.
+In the four-request example, concurrency 2 has no interior-window estimate;
+concurrency 1 can report a two-request window, which is too short to establish stationarity.
+
+Prefer at least 16 requests per worker for measurements (for example 2048 requests
+at concurrency 64); short plans remain useful smoke checks. The interior estimator
+removes boundary completions, not every source of ramp effects or variation. It
+does not prove stationary traffic or isolate server-only throughput.
+
+`round_aggregates` groups each case/concurrency and reports min/median/max and
+relative spread `(max-min)/median` for complete-round rates, p50/p95 and available
+interior rates. Failed rounds are counted explicitly and excluded from aggregates;
+absent metrics are null. When all values are zero, relative spread is zero.
+These are descriptive summaries, not confidence intervals. Stable back-to-back
+rounds do not establish stability across invocations, hosts or days.
+
+Output includes the byte-exact `plan.json`, plan/runner SHA-256 in `config.json`,
+UTC `started_at`/`finished_at`, client hostname/platform/Python version/CPU count,
+connection policy and effective budget in `config.json`, health response,
+`responses.jsonl` with phase/variant, monotonic completion time and raw base64/text/status/
+error data, and `summary.json` with completeness and each round's metrics.
+Completion means the frozen workload passed. It does not establish model quality
+or a universal speedup. Preserve failures and repeated rounds; do not silently
+relax an oracle or extend the run budget after failure.
+
+## Reproduce the example without a model
+
+Run from the repository root with Python 3.11+:
+
+```sh
+python3 - <<'PY'
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from benchmarks.serving_matrix import run
+from tests.benchmarks.test_serving_matrix import server
+with server() as (origin, _), TemporaryDirectory() as scratch:
+    plan = json.loads(Path("benchmarks/serving_matrix.example.json").read_text())
+    plan.update(endpoint=origin + "/v1/systemone", health_endpoint=origin + "/health")
+    path = Path(scratch) / "plan.json"
+    path.write_text(json.dumps(plan))
+    assert run(path, Path(scratch) / "output")["complete"]
+PY
+python3 -m unittest discover -s tests/benchmarks -p 'test_*.py' -v
+```
+
+Tests cover exact oracles, mixed variants, bounded concurrency, stop/drain,
+deadlines, provenance validation and authenticated health/inference. They make
+no GPU or production-throughput claim.
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
new file mode 100644
index 00000000..37c04085
--- /dev/null
+++ b/benchmarks/serving_matrix.py
@@ -0,0 +1,454 @@
+"""Frozen-plan HTTP matrix client; Python 3.11+, standard library only."""
+
+import argparse
+import base64
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor
+import copy
+from datetime import datetime, timezone
+import hashlib
+import http.client
+import json
+import math
+import os
+from pathlib import Path
+import platform
+import socket
+import statistics
+import threading
+import time
+from urllib.parse import urlsplit, urlunsplit
+
+
+def decode_json(data):
+    def reject_constant(value):
+        raise ValueError(f"non-JSON numeric constant: {value}")
+    return json.loads(data, parse_constant=reject_constant)
+
+
+def case_variants(case):
+    return case["variants"] if "variants" in case else [case]
+
+
+def pointer_parent(value, pointer):
+    """Resolve a non-root RFC 6901 pointer; array indices must be canonical."""
+    if not isinstance(pointer, str) or not pointer.startswith("/"):
+        raise ValueError("ignore_paths entries must be non-root JSON Pointers")
+    tokens = pointer[1:].split("/")
+    for token in tokens:
+        if any(i + 1 == len(token) or token[i + 1] not in "01"
+               for i, char in enumerate(token) if char == "~"):
+            raise ValueError("ignore_paths has an invalid JSON Pointer escape")
+    tokens = [token.replace("~1", "/").replace("~0", "~") for token in tokens]
+    parent = value
+    for index, token in enumerate(tokens):
+        if isinstance(parent, dict) and token in parent:
+            key = token
+        elif isinstance(parent, list) and token.isascii() and token.isdigit():
+            # Bound the lexeme before integer conversion, including pathological inputs.
+            if len(token) > len(str(len(parent))):
+                raise ValueError("ignore_paths array index is out of range")
+            key = int(token)
+            if str(key) != token or key >= len(parent):
+                raise ValueError("ignore_paths array index is invalid")
+        else:
+            raise ValueError("ignore_paths does not resolve to an existing field")
+        if index == len(tokens) - 1:
+            return parent, key
+        parent = parent[key]
+
+
+def validate_plan(plan):
+    try:
+        json.dumps(plan, allow_nan=False)
+    except (ValueError, RecursionError) as exc:
+        raise ValueError("plan must contain only finite JSON values within serialization limits") from exc
+    for key in ("endpoint", "health_endpoint"):
+        if key not in plan and key == "health_endpoint":
+            continue
+        parts = urlsplit(plan[key])
+        if parts.scheme not in ("http", "https") or not parts.hostname or parts.username or parts.password or parts.fragment:
+            raise ValueError(f"{key} must be an HTTP(S) URL without credentials or fragment")
+        _ = parts.port
+    cases = plan["cases"]
+    if not isinstance(cases, list) or not cases:
+        raise ValueError("cases must be a nonempty list")
+    names = set()
+    ignore_paths = plan.get("ignore_paths", [])
+    if (not isinstance(ignore_paths, list)
+            or any(not isinstance(path, str) for path in ignore_paths)
+            or len(set(ignore_paths)) != len(ignore_paths)):
+        raise ValueError("ignore_paths must be a list of unique JSON Pointers")
+    for case in cases:
+        name = case["name"]
+        if not isinstance(name, str) or not name or name in names:
+            raise ValueError("case names must be nonempty and unique")
+        names.add(name)
+        variants = case_variants(case)
+        if "variants" in case and any(key in case for key in ("request", "expected_response", "expected_response_text")):
+            raise ValueError("variants replaces the case's request and expectation")
+        if not isinstance(variants, list) or not variants:
+            raise ValueError("variants must be a nonempty list")
+        variant_names = set()
+        for variant in variants:
+            variant_name = variant["name"]
+            if not isinstance(variant_name, str) or not variant_name or variant_name in variant_names:
+                raise ValueError("variant names must be nonempty and unique within a case")
+            variant_names.add(variant_name)
+            if not isinstance(variant["request"], dict):
+                raise ValueError("request must be a JSON object")
+            if ("expected_response" in variant) == ("expected_response_text" in variant):
+                raise ValueError("each variant needs exactly one expected response")
+            for pointer in ignore_paths:
+                if "expected_response_text" in variant:
+                    raise ValueError("ignore_paths cannot be combined with an exact-byte oracle")
+                parent, key = pointer_parent(variant["expected_response"], pointer)
+                if isinstance(parent[key], (dict, list)):
+                    raise ValueError("ignore_paths may exclude only primitive leaf values")
+            if "expected_response_text" in variant and not isinstance(variant["expected_response_text"], str):
+                raise ValueError("expected_response_text must be a string")
+            if "expected_response_text" in variant:
+                try:
+                    variant["expected_response_text"].encode("utf-8")
+                except UnicodeEncodeError as exc:
+                    raise ValueError("expected_response_text must be valid UTF-8 text") from exc
+    count = plan["requests_per_case"]
+    concurrency = plan["concurrency"]
+    if type(count) is not int or count < 1 or not isinstance(concurrency, list) or not concurrency:
+        raise ValueError("requests_per_case and concurrency must be positive")
+    if any(type(c) is not int or c < 1 or c > count for c in concurrency) or len(set(concurrency)) != len(concurrency):
+        raise ValueError("concurrency values must be unique positive integers <= requests_per_case")
+    for name in ("repetitions", "warmup_per_case"):
+        value = plan.get(name, 2)
+        if type(value) is not int or value < 2:
+            raise ValueError(f"{name} must be an integer >= 2")
+    timeout = plan["timeout_seconds"]
+    if type(timeout) not in (int, float) or not math.isfinite(timeout) or timeout <= 0:
+        raise ValueError("timeout_seconds must be finite and positive")
+    if not isinstance(plan["metadata"], dict):
+        raise ValueError("metadata must be a pinned JSON object")
+    for name in (
+        "gpu", "gpu_ids", "driver", "cuda", "precision", "model_revision",
+        "runtime_revision", "cache_policy", "cuda_evidence", "reservation", "host",
+    ):
+        value = plan["metadata"].get(name)
+        if not value or (isinstance(value, str) and not value.strip()):
+            raise ValueError(f"metadata requires {name}")
+
+
+def exact_json(left, right):
+    if type(left) is not type(right):
+        return False
+    if isinstance(left, dict):
+        return left.keys() == right.keys() and all(exact_json(left[key], right[key]) for key in left)
+    if isinstance(left, list):
+        return len(left) == len(right) and all(exact_json(a, b) for a, b in zip(left, right))
+    return left == right
+
+
+def matches_json(actual, expected, ignore_paths):
+    if not ignore_paths:
+        return exact_json(actual, expected)
+    actual, expected = copy.deepcopy(actual), copy.deepcopy(expected)
+    try:
+        for pointer in ignore_paths:
+            actual_parent, actual_key = pointer_parent(actual, pointer)
+            expected_parent, expected_key = pointer_parent(expected, pointer)
+            if type(actual_parent[actual_key]) is not type(expected_parent[expected_key]):
+                return False
+            if isinstance(actual_parent[actual_key], float) and not math.isfinite(actual_parent[actual_key]):
+                return False
+            actual_parent[actual_key] = expected_parent[expected_key] = None
+    except ValueError:
+        return False
+    return exact_json(actual, expected)
+
+
+def exchange(endpoint, timeout, case=None, request_headers=None, ignore_paths=()):
+    """One connection, no proxy, redirect, pooling or retry; retain raw bytes."""
+    started = time.perf_counter()
+    parts = urlsplit(endpoint)
+    path = urlunsplit(("", "", parts.path or "/", parts.query, ""))
+    connection_type = http.client.HTTPSConnection if parts.scheme == "https" else http.client.HTTPConnection
+    connection = connection_type(parts.hostname, parts.port, timeout=timeout)
+    expired = threading.Event()
+    timer = None
+    payload = bytearray()
+    status = None
+    headers = {}
+    error = None
+    decisions = 0
+    connect_seconds = None
+    try:
+        connect_started = time.perf_counter()
+        connection.connect()
+        connect_seconds = time.perf_counter() - connect_started
+        remaining = timeout - (time.perf_counter() - started)
+        if remaining <= 0:
+            raise TimeoutError("connection exceeded deadline")
+        transport = connection.sock
+        transport.settimeout(remaining)
+
+        def expire():
+            expired.set()
+            try:
+                transport.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+
+        timer = threading.Timer(remaining, expire)
+        timer.daemon = True
+        timer.start()
+        body = json.dumps(case["request"], allow_nan=False).encode() if case else None
+        connection.request("POST" if case else "GET", path, body=body,
+                           headers={**(request_headers or {}), **({"Content-Type": "application/json"} if case else {})})
+        response = connection.getresponse()
+        status = response.status
+        headers = dict(response.getheaders())
+        while chunk := response.read1(65536):
+            payload.extend(chunk)
+        if expired.is_set():
+            raise TimeoutError("response exceeded deadline")
+        if response.length not in (None, 0):
+            raise http.client.IncompleteRead(bytes(payload), response.length)
+        if status != 200:
+            error = {"kind": "http", "message": f"HTTP {status}"}
+        elif case:
+            try:
+                decoded = decode_json(bytes(payload))
+            except (ValueError, UnicodeError, RecursionError):
+                decoded = None
+            if "expected_response_text" in case:
+                matches = bytes(payload) == case["expected_response_text"].encode("utf-8")
+            else:
+                matches = matches_json(decoded, case["expected_response"], ignore_paths)
+                # Invalid JSON must fail even when the oracle is JSON null.
+                if decoded is None:
+                    try:
+                        decode_json(bytes(payload))
+                    except (ValueError, UnicodeError, RecursionError):
+                        matches = False
+            if not matches:
+                error = {"kind": "correctness", "message": "response differs from frozen expectation"}
+            elif isinstance(decoded, dict) and isinstance(decoded.get("answers"), (dict, list)):
+                decisions = len(decoded["answers"])
+    except RecursionError:
+        error = {"kind": "correctness", "message": "response validation exceeds recursion limit"}
+    except (OSError, http.client.HTTPException) as exc:
+        kind = "timeout" if expired.is_set() or isinstance(exc, TimeoutError) else "transport"
+        error = {"kind": kind, "message": f"{type(exc).__name__}: {exc}"}
+    finally:
+        if timer:
+            timer.cancel()
+        connection.close()
+    completed_at = time.perf_counter()
+    # Validation/close and a delayed watchdog can also exhaust the exchange budget.
+    if expired.is_set() or completed_at - started >= timeout:
+        error = {"kind": "timeout", "message": "exchange exceeded deadline"}
+        decisions = 0
+    return {
+        "endpoint": endpoint, "status": status, "headers": headers,
+        "raw_response_base64": base64.b64encode(payload).decode("ascii"),
+        "raw_response_text": bytes(payload).decode("utf-8", errors="replace"),
+        "latency_seconds": completed_at - started, "connect_seconds": connect_seconds,
+        "completed_at_monotonic_seconds": completed_at,
+        "successful_decisions": decisions, "error": error,
+    }
+
+
+def round_summary(records, elapsed, case, concurrency, repetition, planned):
+    successful = [r for r in records if r["error"] is None]
+    latencies = sorted(r["latency_seconds"] for r in successful)
+    quantiles = {f"p{p}": latencies[math.ceil(p / 100 * len(latencies)) - 1] if latencies else None for p in (50, 95)}
+    decisions = sum(r["successful_decisions"] for r in successful)
+    complete = len(records) == planned and len(successful) == planned
+    steady = {"available": False, "excluded_each_end": concurrency,
+              "reason": "round incomplete" if not complete else "requires more than 2 * concurrency completions"}
+    if complete and len(records) > 2 * concurrency:
+        ordered = sorted(records, key=lambda r: r["completed_at_monotonic_seconds"])
+        interior = ordered[concurrency:-concurrency]
+        start = ordered[concurrency - 1]["completed_at_monotonic_seconds"]
+        end = interior[-1]["completed_at_monotonic_seconds"]
+        duration = end - start
+        if duration > 0:
+            count = sum(r["successful_decisions"] for r in interior)
+            steady = {"available": True, "excluded_each_end": concurrency,
+                      "requests": len(interior), "decisions": count,
+                      "started_at_monotonic_seconds": start, "finished_at_monotonic_seconds": end,
+                      "elapsed_seconds": duration, "requests_per_second": len(interior) / duration,
+                      "decisions_per_second": count / duration}
+        else:
+            steady["reason"] = "completion window has no positive elapsed time"
+    return {
+        "case": case, "concurrency": concurrency, "repetition": repetition,
+        "planned_requests": planned, "attempted_requests": len(records),
+        "elapsed_seconds": elapsed, "successful_requests": len(successful),
+        "successful_decisions": decisions,
+        "successful_requests_per_second": len(successful) / elapsed,
+        "successful_decisions_per_second": decisions / elapsed,
+        "successful_latency_seconds": quantiles,
+        "failures": dict(Counter(r["error"]["kind"] for r in records if r["error"])),
+        "failure_latency_seconds": [r["latency_seconds"] for r in records if r["error"]],
+        "complete": complete, "steady_state": steady,
+        "throughput_scope": "complete_wave_including_ramp_and_drain",
+    }
+
+
+def aggregate_rounds(rounds, repetitions):
+    groups = {}
+    for result in rounds:
+        groups.setdefault((result["case"], result["concurrency"]), []).append(result)
+
+    def spread(values):
+        if not values:
+            return None
+        low, median, high = min(values), statistics.median(values), max(values)
+        return {"rounds": len(values), "min": low, "median": median, "max": high,
+                "relative_spread": (high - low) / median if median else (0.0 if high == low else None)}
+
+    aggregated = []
+    for (case, concurrency), results in groups.items():
+        complete = [r for r in results if r["complete"]]
+        windows = [r["steady_state"] for r in complete if r["steady_state"]["available"]]
+        metrics = {key: spread([r[key] for r in complete]) for key in (
+            "successful_requests_per_second", "successful_decisions_per_second")}
+        for p in ("p50", "p95"):
+            metrics[p + "_seconds"] = spread([r["successful_latency_seconds"][p] for r in complete])
+        for kind in ("requests", "decisions"):
+            metrics["steady_state_" + kind + "_per_second"] = spread([w[kind + "_per_second"] for w in windows])
+        aggregated.append({"case": case, "concurrency": concurrency,
+                           "planned_rounds": repetitions, "attempted_rounds": len(results),
+                           "complete_rounds": len(complete), "incomplete_rounds": len(results) - len(complete),
+                           "metrics": metrics})
+    return aggregated
+
+
+def run(plan_path, output):
+    snapshot = plan_path.read_bytes()
+    plan = decode_json(snapshot)
+    validate_plan(plan)
+    request_headers = {}
+    if token := os.environ.get("OMNI_JEV_TEST_TOKEN"):
+        if any(ord(c) < 32 or ord(c) > 126 for c in token):
+            raise ValueError("invalid OMNI_JEV_TEST_TOKEN: expected printable ASCII")
+        request_headers["Authorization"] = f"Bearer {token}"
+    output.mkdir(parents=True, exist_ok=False)
+
+    def save(name, value):
+        (output / name).write_text(json.dumps(value, indent=2, allow_nan=False) + "\n")
+
+    (output / "plan.json").write_bytes(snapshot)
+    repetitions, warmup = plan.get("repetitions", 2), plan.get("warmup_per_case", 2)
+    config = {
+        "plan_sha256": hashlib.sha256(snapshot).hexdigest(),
+        "runner_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "plan": plan, "timing": "client wall time including DNS/TCP/TLS, request serialization and response validation",
+        "retries": 0, "redirects": False, "environment_proxy": False,
+        "effective_budget": {"repetitions": repetitions, "warmup_per_case": warmup},
+        "started_at": datetime.now(timezone.utc).isoformat(), "finished_at": None,
+        "client": {"hostname": socket.gethostname(), "platform": platform.platform(),
+                   "python_version": platform.python_version(), "cpu_count": os.cpu_count()},
+        "connection_policy": "new_connection_per_request",
+    }
+    save("config.json", config)
+    parts = urlsplit(plan["endpoint"])
+    health_endpoint = plan.get("health_endpoint", urlunsplit((parts.scheme, parts.netloc, "/health", "", "")))
+    health = exchange(health_endpoint, plan["timeout_seconds"], request_headers=request_headers)
+    save("health.json", health)
+    all_records = []
+    rounds = []
+    stopped = threading.Event()
+    if health["error"]:
+        stopped.set()
+    lock = threading.Lock()
+    with (output / "responses.jsonl").open("w") as sink:
+        def request(case, phase, index, concurrency=1, repetition=None, variant_index=None):
+            variants = case_variants(case)
+            if variant_index is None:
+                variant_index = index % len(variants)
+            variant = variants[variant_index]
+            record = exchange(plan["endpoint"], plan["timeout_seconds"], variant, request_headers,
+                              plan.get("ignore_paths", []))
+            record.update(case=case["name"], phase=phase, index=index,
+                          concurrency=concurrency, repetition=repetition,
+                          variant=variant["name"], variant_index=variant_index)
+            with lock:
+                if record["error"]:
+                    stopped.set()
+                all_records.append(record)
+                sink.write(json.dumps(record, allow_nan=False) + "\n")
+                sink.flush()
+            return record
+
+        for case in plan["cases"]:
+            if stopped.is_set():
+                break
+            for variant_index in range(len(case_variants(case))):
+                request(case, "readiness", 0, variant_index=variant_index)
+                if stopped.is_set():
+                    break
+        for case in plan["cases"]:
+            for variant_index in range(len(case_variants(case))):
+                for index in range(warmup):
+                    if stopped.is_set():
+                        break
+                    request(case, "warmup", index, variant_index=variant_index)
+
+        def wave(case, phase, concurrency, count, repetition=None):
+            next_index = 0
+            completed = []
+
+            def worker():
+                nonlocal next_index
+                while True:
+                    with lock:
+                        if stopped.is_set() or next_index == count:
+                            return
+                        index = next_index
+                        next_index += 1
+                    record = request(case, phase, index, concurrency, repetition)
+                    with lock:
+                        completed.append(record)
+
+            start = time.perf_counter()
+            with ThreadPoolExecutor(max_workers=concurrency) as pool:
+                futures = [pool.submit(worker) for _ in range(concurrency)]
+                for future in futures:
+                    future.result()
+            return completed, time.perf_counter() - start
+
+        for case in plan["cases"]:
+            for concurrency in plan["concurrency"]:
+                if stopped.is_set():
+                    break
+                wave(case, "feasibility", concurrency, min(concurrency, plan["requests_per_case"]))
+                for repetition in range(1, repetitions + 1):
+                    if stopped.is_set():
+                        break
+                    records, elapsed = wave(case, "measured", concurrency, plan["requests_per_case"], repetition)
+                    rounds.append(round_summary(records, elapsed, case["name"], concurrency, repetition, plan["requests_per_case"]))
+    failures = Counter(r["error"]["kind"] for r in [health, *all_records] if r["error"])
+    summary = {
+        "complete": not bool(failures), "failures": dict(failures),
+        "inference_requests": len(all_records), "health_requests": 1,
+        "rounds": rounds,
+        "round_aggregates": aggregate_rounds(rounds, repetitions),
+        "excluded_phases": ["health", "readiness", "warmup", "feasibility"],
+    }
+    save("summary.json", summary)
+    config["finished_at"] = datetime.now(timezone.utc).isoformat()
+    save("config.json", config)
+    return summary
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("plan", type=Path)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args(argv)
+    summary = run(args.plan, args.output)
+    return 0 if summary["complete"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
new file mode 100644
index 00000000..5f813a60
--- /dev/null
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -0,0 +1,631 @@
+import hashlib
+import base64
+from datetime import datetime
+import platform
+import socket
+import statistics
+import json
+import os
+import tempfile
+import threading
+import time
+import unittest
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from benchmarks import serving_matrix
+
+
+EXPECTED = {"answers": {"a": {"choice": "yes"}, "b": {"noul": 0.75}}}
+
+
+@contextmanager
+def server(actions=None, health_status=200, barrier_indices=(), responses_by_tag=None, authorization=None):
+    state = {"gets": 0, "posts": 0, "active": 0, "peak": 0, "redirect_hits": 0, "authorization": []}
+    lock = threading.Lock()
+    barrier = threading.Barrier(len(barrier_indices)) if barrier_indices else None
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def do_GET(self):
+            state["gets"] += 1
+            state["authorization"].append(self.headers.get("Authorization"))
+            if authorization and self.headers.get("Authorization") != authorization:
+                self.send_response(401)
+                self.end_headers()
+                return
+            if self.path == "/redirected":
+                state["redirect_hits"] += 1
+            self.send_response(health_status)
+            self.end_headers()
+            self.wfile.write(b'{"ready":true}')
+
+        def do_POST(self):
+            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            state["authorization"].append(self.headers.get("Authorization"))
+            if authorization and self.headers.get("Authorization") != authorization:
+                self.send_response(401)
+                self.end_headers()
+                return
+            with lock:
+                state["posts"] += 1
+                index = state["posts"]
+                state["active"] += 1
+                state["peak"] = max(state["peak"], state["active"])
+            status, payload, delay = (actions or {}).get(
+                index, (200, json.dumps(EXPECTED).encode(), 0.015)
+            )
+            if responses_by_tag is not None:
+                payload = json.dumps(responses_by_tag[body["tag"]]).encode()
+            if callable(payload):
+                payload = payload(index)
+            try:
+                if index in barrier_indices:
+                    barrier.wait(timeout=5)
+                time.sleep(delay)
+                self.send_response(status)
+                if status == 302:
+                    self.send_header("Location", "/redirected")
+                self.end_headers()
+                if isinstance(payload, list):
+                    for chunk in payload:
+                        self.wfile.write(chunk)
+                        self.wfile.flush()
+                        time.sleep(delay)
+                else:
+                    self.wfile.write(payload)
+            except (BrokenPipeError, ConnectionResetError):
+                pass
+            finally:
+                with lock:
+                    state["active"] -= 1
+
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{httpd.server_port}", state
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+        thread.join()
+
+
+class MatrixTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+
+    def plan(self, origin, **overrides):
+        plan = {
+            "endpoint": origin + "/inference",
+            "cases": [{"name": "two-decisions", "request": {"model": "test"},
+                       "expected_response": EXPECTED}],
+            "concurrency": [1, 2], "requests_per_case": 4,
+            "repetitions": 2, "warmup_per_case": 2,
+            "timeout_seconds": 2, "metadata": {
+                "gpu": "N/A: local CPU test server", "gpu_ids": "N/A",
+                "driver": "N/A", "cuda": "N/A", "precision": "N/A",
+                "model_revision": "synthetic-test-v1", "runtime_revision": "test-server-v1",
+                "cache_policy": "none", "cuda_evidence": "N/A: no CUDA execution",
+                "reservation": "N/A: local CPU test", "host": "synthetic-loopback",
+            },
+        }
+        plan.update(overrides)
+        path = self.root / "input.json"
+        path.write_text(json.dumps(plan, indent=2) + "\n")
+        return path
+
+    def execute(self, path):
+        output = self.root / "results"
+        code = serving_matrix.main([str(path), "--output", str(output)])
+        self.assertTrue((output / "summary.json").is_file(), "runner must write a summary")
+        summary = json.loads((output / "summary.json").read_text())
+        records = [json.loads(line) for line in (output / "responses.jsonl").read_text().splitlines()]
+        return code, summary, records, output
+
+    def test_metadata_requires_each_pinned_identity_before_creating_output(self):
+        path = self.plan("http://127.0.0.1:1")
+        baseline = json.loads(path.read_text())
+        for field in baseline["metadata"]:
+            for index, value in enumerate((None, "", "   ", [], {})):
+                with self.subTest(field=field, value=value):
+                    output = self.root / (field + str(index))
+                    plan = json.loads(json.dumps(baseline))
+                    plan["metadata"][field] = value
+                    path.write_text(json.dumps(plan))
+                    with self.assertRaisesRegex(ValueError, "metadata requires " + field):
+                        serving_matrix.run(path, output)
+                    self.assertFalse(output.exists())
+        baseline["metadata"] = {}
+        path.write_text(json.dumps(baseline))
+        with self.assertRaisesRegex(ValueError, "metadata requires gpu"):
+            serving_matrix.run(path, self.root / "results")
+
+    def test_environment_bearer_auth_covers_health_and_every_phase_without_saving_secret(self):
+        token = "matrix-local-test-secret"
+        with server(authorization="Bearer " + token) as (origin, state), patch.dict(
+            os.environ, {"OMNI_JEV_TEST_TOKEN": token}
+        ):
+            code, summary, records, output = self.execute(self.plan(origin))
+        self.assertEqual(code, 0)
+        self.assertTrue(summary["complete"])
+        self.assertEqual(len(state["authorization"]), 23)
+        self.assertEqual(set(state["authorization"]), {"Bearer " + token})
+        self.assertEqual({r["phase"] for r in records}, {"readiness", "warmup", "feasibility", "measured"})
+        for file in output.iterdir():
+            self.assertNotIn(token, file.read_text())
+
+    def test_invalid_environment_bearer_token_fails_before_network_or_output(self):
+        path = self.plan("http://127.0.0.1:1")
+        for token in ("unsafe\r\nInjected: header", "not-latin-1-☃"):
+            with self.subTest(token=token), patch.dict(os.environ, {"OMNI_JEV_TEST_TOKEN": token}):
+                with self.assertRaisesRegex(ValueError, "invalid OMNI_JEV_TEST_TOKEN"):
+                    serving_matrix.run(path, self.root / "results")
+                self.assertFalse((self.root / "results").exists())
+
+    def test_declared_json_pointers_allow_native_timers_and_preserve_raw_evidence(self):
+        for model, pointer in [("Open-Jev-9B", "/metadata/inference_seconds"), ("JEMM", "/usage/latency_ms")]:
+            with self.subTest(model=model), tempfile.TemporaryDirectory() as scratch:
+                self.root = Path(scratch)
+                expected = {"model": model, "answers": EXPECTED["answers"],
+                            "usage": {"input_tokens": 12, "output_tokens": 0}}
+                parent, key = pointer[1:].split("/")
+                expected.setdefault(parent, {})[key] = 0.0
+                def payload(index):
+                    response = json.loads(json.dumps(expected))
+                    response[parent][key] = float(index)
+                    return json.dumps(response).encode()
+                actions = {i: (200, payload, 0) for i in range(1, 40)}
+                with server(actions) as (origin, _):
+                    path = self.plan(origin, ignore_paths=[pointer], cases=[{
+                        "name": "native", "request": {}, "expected_response": expected}])
+                    snapshot = path.read_bytes()
+                    code, summary, records, output = self.execute(path)
+                self.assertEqual(code, 0)
+                self.assertTrue(summary["complete"])
+                self.assertEqual((output / "plan.json").read_bytes(), snapshot)
+                config = json.loads((output / "config.json").read_text())
+                self.assertEqual(config["plan_sha256"], hashlib.sha256(snapshot).hexdigest())
+                self.assertEqual(config["plan"]["cases"][0]["expected_response"], expected)
+                for record in records:
+                    raw = json.loads(record["raw_response_text"])
+                    self.assertGreater(raw[parent][key], 0)
+                    self.assertEqual(base64.b64decode(record["raw_response_base64"]), record["raw_response_text"].encode())
+
+    def test_timer_exclusion_still_rejects_changed_answers_missing_fields_and_types(self):
+        expected = {"answers": EXPECTED["answers"], "metadata": {"inference_seconds": 0.0}}
+        for mutation in ("answer", "missing", "type", "extra"):
+            with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as scratch:
+                self.root = Path(scratch)
+                actual = json.loads(json.dumps(expected))
+                actual["metadata"]["inference_seconds"] = 1.0
+                if mutation == "answer": actual["answers"]["a"]["choice"] = "no"
+                elif mutation == "missing": del actual["metadata"]["inference_seconds"]
+                elif mutation == "type": actual["metadata"]["inference_seconds"] = "not a number"
+                else: actual["metadata"]["unexpected"] = True
+                with server({1: (200, json.dumps(actual).encode(), 0)}) as (origin, state):
+                    code, summary, records, _ = self.execute(self.plan(origin,
+                        ignore_paths=["/metadata/inference_seconds"], cases=[{
+                            "name": "native", "request": {}, "expected_response": expected}]))
+                self.assertEqual(code, 1)
+                self.assertEqual(summary["failures"], {"correctness": 1})
+                self.assertEqual(state["posts"], 1)
+                self.assertEqual(json.loads(records[0]["raw_response_text"]), actual)
+
+    def test_strict_default_reproduces_native_timing_oracle_failure(self):
+        expected = {"answers": EXPECTED["answers"], "metadata": {"inference_seconds": 0.0}}
+        actual = {"answers": EXPECTED["answers"], "metadata": {"inference_seconds": 1.0}}
+        with server({1: (200, json.dumps(actual).encode(), 0)}) as (origin, state):
+            code, summary, _, _ = self.execute(self.plan(origin, cases=[{
+                "name": "native", "request": {}, "expected_response": expected}]))
+        self.assertEqual(code, 1)
+        self.assertEqual(state["posts"], 1)
+        self.assertEqual(summary["rounds"], [])
+        self.assertEqual(summary["failures"], {"correctness": 1})
+
+    def test_invalid_exclusions_fail_before_network_or_output(self):
+        for index, paths in enumerate([None, "wrong", [""], ["metadata/t"],
+                ["/metadata/~2bad"], ["/missing"], ["/metadata"], ["/metadata/t", "/metadata/t"],
+                [1], ["/items/01"], ["/items/-"]]):
+            with self.subTest(paths=paths):
+                path = self.plan("http://127.0.0.1:1", ignore_paths=paths, cases=[{
+                    "name": "native", "request": {}, "expected_response": {
+                        "metadata": {"t": 0.0}, "items": [0.0, 0.0]}}])
+                output = self.root / str(index)
+                with self.assertRaisesRegex(ValueError, "ignore_paths"):
+                    serving_matrix.run(path, output)
+                self.assertFalse(output.exists())
+        path = self.plan("http://127.0.0.1:1", ignore_paths=["/timer"], cases=[{
+            "name": "bytes", "request": {}, "expected_response_text": "literal"}])
+        with self.assertRaisesRegex(ValueError, "ignore_paths"):
+            serving_matrix.run(path, self.root / "text")
+
+    def test_json_pointer_escaping_and_array_indices_preserve_unexcluded_values(self):
+        expected = {"a/b": {"~timer": 0.0}, "items": [{"timer": 0.0}, {"timer": 0.0}], "answer": "yes"}
+        actual = {"a/b": {"~timer": 3.0}, "items": [{"timer": 7.0}, {"timer": 9.0}], "answer": "yes"}
+        actions = {i: (200, json.dumps(actual).encode(), 0) for i in range(1, 40)}
+        with server(actions) as (origin, _):
+            code, _, _, _ = self.execute(self.plan(origin,
+                ignore_paths=["/a~1b/~0timer", "/items/0/timer", "/items/1/timer"],
+                cases=[{"name": "escaped", "request": {}, "expected_response": expected}]))
+        self.assertEqual(code, 0)
+
+    def test_excluded_float_still_rejects_nonfinite_numeric_lexeme(self):
+        raw = b'{"answers":{},"metadata":{"inference_seconds":1e309}}'
+        expected = {"answers": {}, "metadata": {"inference_seconds": 0.0}}
+        with server({i: (200, raw, 0) for i in range(1, 40)}) as (origin, _):
+            code, summary, records, _ = self.execute(self.plan(origin,
+                ignore_paths=["/metadata/inference_seconds"], cases=[{
+                    "name": "native", "request": {}, "expected_response": expected}]))
+        self.assertEqual(code, 1)
+        self.assertEqual(summary["failures"], {"correctness": 1})
+        self.assertEqual(records[0]["raw_response_text"].encode(), raw)
+
+    def test_partial_round_is_counted_but_excluded_from_spread(self):
+        with server({6: (503, b"unavailable", 0)}) as (origin, _):
+            code, summary, _, output = self.execute(self.plan(origin, concurrency=[1], repetitions=3))
+        self.assertEqual(code, 1)
+        self.assertEqual(len(summary["rounds"]), 1)
+        group = summary["round_aggregates"][0]
+        self.assertEqual(group["planned_rounds"], 3)
+        self.assertEqual(group["attempted_rounds"], 1)
+        self.assertEqual(group["complete_rounds"], 0)
+        self.assertEqual(group["incomplete_rounds"], 1)
+        self.assertTrue(all(value is None for value in group["metrics"].values()))
+        self.assertFalse(summary["rounds"][0]["steady_state"]["available"])
+        self.assertIsNotNone(json.loads((output / "config.json").read_text())["finished_at"])
+
+    def test_predeclared_budget_and_descriptive_spread(self):
+        with server() as (origin, state):
+            code, summary, records, _ = self.execute(self.plan(origin,
+                concurrency=[1], repetitions=3, warmup_per_case=3))
+        self.assertEqual(code, 0)
+        self.assertEqual(state["posts"], 17)
+        self.assertEqual(len([r for r in records if r["phase"] == "warmup"]), 3)
+        self.assertEqual(len(summary["rounds"]), 3)
+        self.assertIn("round_aggregates", summary)
+        group = summary["round_aggregates"][0]
+        self.assertEqual(group["complete_rounds"], 3)
+        rates = [r["successful_requests_per_second"] for r in summary["rounds"]]
+        spread = group["metrics"]["successful_requests_per_second"]
+        self.assertEqual(spread["min"], min(rates))
+        self.assertEqual(spread["median"], statistics.median(rates))
+        self.assertEqual(spread["max"], max(rates))
+        self.assertAlmostEqual(spread["relative_spread"], (max(rates)-min(rates))/statistics.median(rates))
+
+    def test_omitted_budgets_use_recorded_defaults(self):
+        with server() as (origin, _):
+            path = self.plan(origin, concurrency=[1])
+            plan = json.loads(path.read_text());del plan["repetitions"];del plan["warmup_per_case"]
+            path.write_text(json.dumps(plan))
+            code, summary, _, output = self.execute(path)
+        self.assertEqual(code, 0)
+        self.assertEqual(len(summary["rounds"]), 2)
+        config = json.loads((output / "config.json").read_text())
+        self.assertEqual(config["effective_budget"], {"repetitions": 2, "warmup_per_case": 2})
+
+    def test_run_provenance_and_connection_cost_are_recorded(self):
+        with server() as (origin, _):
+            code, _, records, output = self.execute(self.plan(origin, concurrency=[1]))
+        self.assertEqual(code, 0)
+        config = json.loads((output / "config.json").read_text())
+        self.assertIn("started_at", config)
+        started, finished = [datetime.fromisoformat(config[key]) for key in ("started_at", "finished_at")]
+        self.assertEqual(started.utcoffset().total_seconds(), 0)
+        self.assertGreaterEqual(finished, started)
+        self.assertEqual(config["client"], {"hostname": socket.gethostname(), "platform": platform.platform(),
+            "python_version": platform.python_version(), "cpu_count": os.cpu_count()})
+        self.assertEqual(config["connection_policy"], "new_connection_per_request")
+        for record in records:
+            self.assertGreaterEqual(record["connect_seconds"], 0)
+            self.assertLessEqual(record["connect_seconds"], record["latency_seconds"])
+            self.assertGreater(record["completed_at_monotonic_seconds"], 0)
+
+    def test_closed_loop_counts_metrics_and_provenance(self):
+        with server() as (origin, state):
+            path = self.plan(origin)
+            code, summary, records, output = self.execute(path)
+        self.assertEqual(code, 0)
+        self.assertEqual(len(records), 22)
+        self.assertEqual(state["posts"], 22)
+        self.assertEqual(state["peak"], 2)
+        self.assertEqual(summary["failures"], {})
+        self.assertEqual(len(summary["rounds"]), 4)
+        for result in summary["rounds"]:
+            self.assertEqual(result["successful_requests"], 4)
+            self.assertEqual(result["successful_decisions"], 8)
+            self.assertAlmostEqual(result["successful_requests_per_second"], 4 / result["elapsed_seconds"])
+            self.assertAlmostEqual(result["successful_decisions_per_second"], 8 / result["elapsed_seconds"])
+            latencies = sorted(r["latency_seconds"] for r in records if r["phase"] == "measured" and r["concurrency"] == result["concurrency"] and r["repetition"] == result["repetition"])
+            self.assertEqual(result["successful_latency_seconds"]["p50"], latencies[1])
+            self.assertEqual(result["successful_latency_seconds"]["p95"], latencies[3])
+        self.assertEqual((output / "plan.json").read_bytes(), path.read_bytes())
+        config = json.loads((output / "config.json").read_text())
+        self.assertEqual(config["plan_sha256"], hashlib.sha256(path.read_bytes()).hexdigest())
+        self.assertEqual(config["runner_sha256"], hashlib.sha256(Path(serving_matrix.__file__).read_bytes()).hexdigest())
+
+    def test_correctness_failure_stops_dispatch_but_drains_in_flight(self):
+        with server({6: (200, b'{"answers":{}}', 0.005), 7: (200, json.dumps(EXPECTED).encode(), 0.04)}, barrier_indices=(6, 7)) as (origin, state):
+            code, summary, records, _ = self.execute(self.plan(origin, concurrency=[2]))
+        self.assertEqual(code, 1)
+        self.assertEqual(state["posts"], 7)
+        measured = [r for r in records if r["phase"] == "measured"]
+        self.assertEqual(len(measured), 2)
+        self.assertEqual(summary["failures"], {"correctness": 1})
+        result = summary["rounds"][0]
+        self.assertEqual(result["attempted_requests"], 2)
+        self.assertEqual(result["successful_requests"], 1)
+        self.assertEqual(result["successful_decisions"], 2)
+        self.assertAlmostEqual(result["successful_requests_per_second"], 1 / result["elapsed_seconds"])
+        self.assertEqual(len(result["failure_latency_seconds"]), 1)
+
+    def test_http_failure_preserves_body_and_nonzero_status(self):
+        with server({5: (503, b"unavailable", 0.001)}) as (origin, _):
+            code, summary, records, _ = self.execute(self.plan(origin, concurrency=[1]))
+        self.assertEqual(code, 1)
+        self.assertEqual(summary["failures"], {"http": 1})
+        self.assertEqual(records[-1]["status"], 503)
+        self.assertEqual(records[-1]["raw_response_text"], "unavailable")
+        self.assertGreater(records[-1]["latency_seconds"], 0)
+        result = summary["rounds"][0]
+        self.assertEqual(result["successful_requests_per_second"], 0)
+        self.assertEqual(result["successful_latency_seconds"], {"p50": None, "p95": None})
+
+    def test_timeout_has_failure_latency_and_no_retry(self):
+        with server({5: (200, b"late", 0.2)}) as (origin, state):
+            code, summary, records, _ = self.execute(self.plan(origin, concurrency=[1], timeout_seconds=0.05))
+        self.assertEqual(code, 1)
+        self.assertEqual(state["posts"], 5)
+        self.assertEqual(summary["failures"], {"timeout": 1})
+        self.assertGreaterEqual(records[-1]["latency_seconds"], 0.04)
+
+    def test_total_body_deadline_retains_partial_raw_bytes(self):
+        with server({1: (200, [b'{"answers":', b"{}", b"}"], 0.03)}) as (origin, _):
+            code, summary, records, _ = self.execute(self.plan(origin, timeout_seconds=0.05))
+        self.assertEqual(code, 1)
+        self.assertEqual(summary["failures"], {"timeout": 1})
+        self.assertEqual(records[0]["raw_response_text"], '{"answers":')
+        self.assertLess(records[0]["latency_seconds"], 0.15)
+
+    def test_validation_or_close_over_deadline_cannot_succeed(self):
+        raw = json.dumps(EXPECTED).encode()
+        for stage in ("validation", "close"):
+            with self.subTest(stage=stage):
+                clock = {"now": 0.0}
+                response = Mock(status=200, length=0)
+                response.getheaders.return_value = []
+                response.read1.side_effect = [raw, b""]
+                connection = Mock()
+                connection.getresponse.return_value = response
+
+                def match(*args):
+                    if stage == "validation":
+                        clock["now"] = 0.08
+                    return True
+
+                def close():
+                    if stage == "close":
+                        clock["now"] = 0.08
+
+                connection.close.side_effect = close
+                with patch.object(serving_matrix.http.client, "HTTPConnection", return_value=connection), \
+                     patch.object(serving_matrix.threading, "Timer"), \
+                     patch.object(serving_matrix.time, "perf_counter", side_effect=lambda: clock["now"]), \
+                     patch.object(serving_matrix, "matches_json", side_effect=match):
+                    record = serving_matrix.exchange("http://localhost/inference", 0.02,
+                                                    {"request": {}, "expected_response": EXPECTED})
+                self.assertEqual(record["error"]["kind"] if record["error"] else None, "timeout")
+                self.assertEqual(record["successful_decisions"], 0)
+                self.assertEqual(record["status"], 200)
+                self.assertEqual(record["raw_response_text"], raw.decode())
+                self.assertEqual(record["latency_seconds"], 0.08)
+
+    def test_over_deadline_transport_error_is_timeout_without_watchdog_event(self):
+        clock = {"now": 0.0}
+        connection = Mock()
+
+        def fail():
+            clock["now"] = 0.08
+            raise OSError("connection reset after deadline")
+
+        connection.getresponse.side_effect = fail
+        with patch.object(serving_matrix.http.client, "HTTPConnection", return_value=connection), \
+             patch.object(serving_matrix.threading, "Timer"), \
+             patch.object(serving_matrix.time, "perf_counter", side_effect=lambda: clock["now"]):
+            record = serving_matrix.exchange("http://localhost/inference", 0.02,
+                                            {"request": {}, "expected_response": EXPECTED})
+        self.assertEqual(record["error"]["kind"], "timeout")
+        self.assertEqual(record["successful_decisions"], 0)
+        self.assertEqual(record["latency_seconds"], 0.08)
+
+    def test_environment_proxy_is_ignored(self):
+        proxies = {"HTTP_PROXY": "http://127.0.0.1:1", "http_proxy": "http://127.0.0.1:1", "ALL_PROXY": "http://127.0.0.1:1", "NO_PROXY": "", "no_proxy": ""}
+        with server() as (origin, _), patch.dict(os.environ, proxies):
+            code, _, _, _ = self.execute(self.plan(origin, concurrency=[1]))
+        self.assertEqual(code, 0)
+
+    def test_does_not_follow_redirect(self):
+        with server({1: (302, b"redirect", 0)}) as (origin, state):
+            code, summary, records, output = self.execute(self.plan(origin))
+        self.assertEqual(code, 1)
+        self.assertEqual(state["redirect_hits"], 0)
+        self.assertEqual(records[0]["status"], 302)
+        self.assertTrue((output / "health.json").exists())
+        self.assertEqual(summary["failures"], {"http": 1})
+
+    def test_health_failure_stops_before_readiness(self):
+        with server(health_status=503) as (origin, state):
+            code, summary, records, output = self.execute(self.plan(origin))
+        self.assertEqual(code, 1)
+        self.assertEqual(state["posts"], 0)
+        self.assertEqual(records, [])
+        self.assertEqual(summary["rounds"], [])
+        self.assertEqual(json.loads((output / "health.json").read_text())["status"], 503)
+        self.assertEqual(summary["failures"], {"http": 1})
+
+    def test_each_case_has_readiness_and_exact_text_is_supported(self):
+        second = {"name": "text", "request": {}, "expected_response_text": "literal\n"}
+        actions = {2: (200, b"literal\n", 0)}
+        with server(actions) as (origin, _):
+            path = self.plan(origin, cases=[{"name": "first", "request": {}, "expected_response": EXPECTED}, second])
+            code, summary, records, _ = self.execute(path)
+        self.assertEqual([r["case"] for r in records if r["phase"] == "readiness"], ["first", "text"])
+        self.assertEqual(code, 1)  # Subsequent text warmup uses a JSON body and must fail.
+        self.assertEqual(summary["failures"], {"correctness": 1})
+
+    def test_raw_text_expectation_completes_and_preserves_base64(self):
+        text = "literal\n"
+        actions = {index: (200, text.encode(), 0) for index in range(1, 13)}
+        with server(actions) as (origin, _):
+            code, summary, records, _ = self.execute(self.plan(origin, concurrency=[1], cases=[{"name": "text", "request": {}, "expected_response_text": text}]))
+        self.assertEqual(code, 0)
+        self.assertEqual(summary["rounds"][0]["successful_decisions"], 0)
+        self.assertEqual(records[0]["raw_response_base64"], "bGl0ZXJhbAo=")
+
+    def test_json_comparison_rejects_bool_number_equivalence(self):
+        with server({1: (200, b'{"answers":{"a":{"choice":"yes"},"b":{"noul":true}}}', 0)}) as (origin, _):
+            expected = {"answers": {"a": {"choice": "yes"}, "b": {"noul": 1}}}
+            code, summary, _, _ = self.execute(self.plan(origin, cases=[{"name": "typed", "request": {}, "expected_response": expected}]))
+        self.assertEqual(code, 1)
+        self.assertEqual(summary["failures"], {"correctness": 1})
+
+    def test_rejects_invalid_budget_before_creating_output(self):
+        path = self.plan("http://127.0.0.1:1", concurrency=[8])
+        with self.assertRaises(ValueError):
+            self.execute(path)
+        self.assertFalse((self.root / "results").exists())
+        path = self.plan("http://127.0.0.1:1", repetitions=1)
+        with self.assertRaises(ValueError):
+            self.execute(path)
+
+    def test_nonfinite_plan_values_fail_before_http_or_output(self):
+        with server() as (origin, state):
+            for section in ("request", "metadata", "expected_response"):
+                with self.subTest(section=section):
+                    path = self.plan(origin)
+                    plan = json.loads(path.read_text())
+                    if section == "metadata":
+                        plan["metadata"]["nested"] = {"overflow": [float("inf")]}
+                    else:
+                        plan["cases"][0][section]["overflow"] = [float("inf")]
+                    path.write_text(json.dumps(plan).replace("Infinity", "1e309"))
+                    output = self.root / f"results-{section}"
+                    with self.assertRaisesRegex(ValueError, "finite JSON"):
+                        serving_matrix.main([str(path), "--output", str(output)])
+                    self.assertFalse(output.exists())
+                    self.assertEqual(state["gets"], 0)
+                    self.assertEqual(state["posts"], 0)
+
+    def test_surrogate_text_expectation_fails_before_http_or_output(self):
+        case = {"name": "text", "request": {}, "expected_response_text": "\ud800"}
+        with server({1: (200, b"ok", 0)}) as (origin, state):
+            path = self.plan(origin, cases=[case])
+            output = self.root / "results"
+            with self.assertRaisesRegex(ValueError, "UTF-8"):
+                serving_matrix.main([str(path), "--output", str(output)])
+            self.assertFalse(output.exists())
+            self.assertEqual(state["gets"], 0)
+            self.assertEqual(state["posts"], 0)
+
+    def test_deep_json_response_is_retained_as_correctness_failure(self):
+        payload = b"[" * 20000 + b"0" + b"]" * 20000
+        with server({1: (200, payload, 0)}) as (origin, state):
+            try:
+                code, summary, records, _ = self.execute(self.plan(origin))
+            except RecursionError:
+                self.fail("deep response must be retained as a correctness failure")
+        self.assertEqual(code, 1)
+        self.assertEqual(state["posts"], 1)
+        self.assertEqual(summary["failures"], {"correctness": 1})
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["status"], 200)
+        self.assertEqual(records[0]["raw_response_text"].encode(), payload)
+
+    def test_deep_json_comparison_is_retained_as_correctness_failure(self):
+        payload = b"[" * 500 + b"0" + b"]" * 500
+        case = {"name": "deep", "request": {}, "expected_response": json.loads(payload)}
+        with server({1: (200, payload, 0)}) as (origin, state):
+            try:
+                code, summary, records, _ = self.execute(self.plan(origin, cases=[case]))
+            except RecursionError:
+                self.fail("deep comparison must be retained as a correctness failure")
+        self.assertEqual(code, 1)
+        self.assertEqual(state["posts"], 1)
+        self.assertEqual(summary["failures"], {"correctness": 1})
+        self.assertEqual(records[0]["raw_response_text"].encode(), payload)
+
+    def test_refuses_existing_output(self):
+        path = self.plan("http://127.0.0.1:1")
+        output = self.root / "results"
+        output.mkdir()
+        (output / "keep").write_text("original")
+        with self.assertRaises(FileExistsError):
+            self.execute(path)
+        self.assertEqual((output / "keep").read_text(), "original")
+
+    def test_mixed_variants_alternate_and_count_actual_decisions(self):
+        responses = {
+            "short": {"answers": {"a": {"choice": "yes"}}},
+            "long": {"answers": {"a": {"choice": "yes"}, "b": {"noul": 0.25}, "c": {"score": 2}}},
+        }
+        case = {"name": "mixed", "variants": [
+            {"name": name, "request": {"tag": name}, "expected_response": response}
+            for name, response in responses.items()
+        ]}
+        with server(responses_by_tag=responses) as (origin, state):
+            code, summary, records, _ = self.execute(self.plan(origin, cases=[case], concurrency=[2]))
+        self.assertEqual(code, 0)
+        self.assertEqual(state["posts"], 16)
+        self.assertEqual([(r["variant"], r["variant_index"]) for r in records if r["phase"] == "readiness"], [("short", 0), ("long", 1)])
+        self.assertEqual([(r["variant"], r["index"]) for r in records if r["phase"] == "warmup"], [("short", 0), ("short", 1), ("long", 0), ("long", 1)])
+        for record in records:
+            self.assertEqual(record["case"], "mixed")
+            self.assertEqual(json.loads(record["raw_response_text"]), responses[record["variant"]])
+            if record["phase"] in ("feasibility", "measured"):
+                self.assertEqual(record["variant_index"], record["index"] % 2)
+                self.assertEqual(record["variant"], ["short", "long"][record["index"] % 2])
+        for result in summary["rounds"]:
+            self.assertEqual(result["case"], "mixed")
+            self.assertEqual(result["successful_requests"], 4)
+            self.assertEqual(result["successful_decisions"], 8)
+            self.assertAlmostEqual(result["successful_decisions_per_second"], 8 / result["elapsed_seconds"])
+
+
+class MetricTests(unittest.TestCase):
+    def records(self, count):
+        return [{"error": None, "latency_seconds": 0.02,
+                 "successful_decisions": i + 1, "completed_at_monotonic_seconds": float(i+1)}
+                for i in range(count)]
+
+    def test_interior_completion_window_excludes_ramp_and_drain(self):
+        records = self.records(8)
+        records = [records[i] for i in [7, 0, 6, 1, 5, 2, 4, 3]]
+        result = serving_matrix.round_summary(records, 9.0, "case", 2, 1, 8)
+        self.assertIn("steady_state", result)
+        window = result["steady_state"]
+        self.assertTrue(window["available"])
+        self.assertEqual(window["requests"], 4)
+        self.assertEqual(window["decisions"], 18)
+        self.assertEqual(window["elapsed_seconds"], 4.0)
+        self.assertEqual(window["requests_per_second"], 1.0)
+        self.assertEqual(window["decisions_per_second"], 4.5)
+        self.assertAlmostEqual(result["successful_requests_per_second"], 8/9)
+        self.assertEqual(result["throughput_scope"], "complete_wave_including_ramp_and_drain")
+
+    def test_no_steady_state_estimate_for_short_or_failed_rounds(self):
+        for count, failed in [(4, False), (8, True)]:
+            records = self.records(count)
+            if failed: records[0]["error"] = {"kind": "http"}
+            result = serving_matrix.round_summary(records, 9.0, "case", 2, 1, count)
+            self.assertIn("steady_state", result)
+            self.assertFalse(result["steady_state"]["available"])
+
+
+if __name__ == "__main__":
+    unittest.main()