From 820e0787f840e010484fca0d50ba11fe36c295f2 Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 8 Oct 2026 10:27:57 +0800
Subject: [PATCH 01/11] benchmarks: add frozen HTTP serving matrix runner

---
 benchmarks/serving_matrix.md            |  94 ++++++++
 benchmarks/serving_matrix.py            | 277 ++++++++++++++++++++++++
 tests/benchmarks/test_serving_matrix.py | 242 +++++++++++++++++++++
 3 files changed, 613 insertions(+)
 create mode 100644 benchmarks/serving_matrix.md
 create mode 100644 benchmarks/serving_matrix.py
 create mode 100644 tests/benchmarks/test_serving_matrix.py

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
new file mode 100644
index 00000000..06455aba
--- /dev/null
+++ b/benchmarks/serving_matrix.md
@@ -0,0 +1,94 @@
+# Frozen HTTP serving matrix
+
+`serving_matrix.py` is a Python 3.11+ standard-library client for one already
+ready server. It does not launch servers, reserve devices, or collect GPU data.
+Freeze model, server, hardware, workload and cache provenance in `metadata` before
+running; these fields are retained verbatim, not independently verified.
+
+```sh
+python3 benchmarks/serving_matrix.py /path/to/frozen-plan.json \
+  --output /path/to/new-results-directory
+```
+
+The output directory must not exist. A plan has this shape (replace the example
+request and exact response with the workload's reference):
+
+```json
+{
+  "endpoint": "http://127.0.0.1:8000/v1/systemone",
+  "health_endpoint": "http://127.0.0.1:8000/health",
+  "cases": [
+    {
+      "name": "short",
+      "request": {"model": "served-alias", "state": "frozen input", "questions": {}},
+      "expected_response": {"answers": {}}
+    }
+  ],
+  "concurrency": [1, 8, 16],
+  "requests_per_case": 32,
+  "repetitions": 2,
+  "warmup_per_case": 2,
+  "timeout_seconds": 120,
+  "metadata": {"server_revision": "exact commit", "workload_revision": "exact revision"}
+}
+```
+
+`health_endpoint` is optional and defaults to the endpoint origin's `/health`.
+Case names and positive concurrency values must be unique. `requests_per_case`
+must be at least the largest concurrency value; repetitions and warmup count
+must each equal two. Requests are POSTed unchanged as JSON. There is no token
+injection or additional model selection.
+
+Each case must specify exactly one of `expected_response` or
+`expected_response_text`. JSON expectations compare decoded values recursively,
+including scalar types, with no tolerance or ignored fields. Object key order
+does not matter; array order does. Text expectations compare the response bytes
+with the expectation's UTF-8 encoding, including whitespace. Invalid JSON fails
+a JSON expectation. An HTTP status other than 200 always fails.
+
+The client captures one global health snapshot and one first inference per case,
+then two sequential warmup requests per case. These phases are excluded from
+performance metrics. Each case/concurrency pair receives one excluded feasibility
+wave of `min(concurrency, requests_per_case)` requests, followed by two measured
+rounds with `requests_per_case` requests each. Fixed worker threads send their
+next request only after the prior one completes; at most `concurrency` client
+requests are in flight. There is no arrival-rate simulation.
+
+Any error stops new dispatch and subsequent phases. Requests already in flight
+finish and are retained. A failed health snapshot still permits one first
+inference to preserve readiness evidence, then stops the matrix. No exchange is
+retried, redirected, or routed through an environment proxy. Socket connection
+operations use the declared timeout; once connected, the remaining deadline
+also covers request transmission, headers and the complete response body.
+Standard-library hostname resolution is not bounded by the socket timeout;
+use a resolved IP endpoint when a strict DNS-independent deadline is required.
+
+Outputs are:
+
+- `plan.json`: exact input bytes.
+- `config.json`: plan SHA256, runner SHA256, metadata and client settings.
+- `health.json`: health status, raw bytes, latency and any failure.
+- `responses.jsonl`: every inference exchange in completion order, with case,
+  phase, zero-based request index, concurrency, measured repetition (1 or 2),
+  status, headers, base64 raw bytes, readable UTF-8 text, latency and error.
+- `summary.json`: global failure counts and a summary for every attempted
+  measured round, including an incomplete round when execution stops.
+
+Round wall time starts before worker creation and ends after all workers finish
+and write their records. Successful requests/s and decisions/s divide successful
+counts by this complete elapsed time, including failures and their drain time.
+A decision is one member of the successful response's `answers` map or list;
+non-JSON text responses have zero counted decisions. Client request latency
+includes transport and response validation. Successful p50/p95 use nearest-rank
+quantiles; failure latencies are listed separately and remain in raw records.
+Round summaries include attempted and planned counts, successful counts, failure
+counts and completion status. Excluded-phase failures appear in global counts.
+Exit status is zero only for a complete error-free matrix; any exchange failure
+returns one after writing the summary. Invalid plans and existing directories
+fail before network requests or output overwrite.
+
+CPU-only verification:
+
+```sh
+python3 -m unittest discover -s tests/benchmarks -p test_serving_matrix.py -v
+```
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
new file mode 100644
index 00000000..abba44c8
--- /dev/null
+++ b/benchmarks/serving_matrix.py
@@ -0,0 +1,277 @@
+"""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 hashlib
+import http.client
+import json
+import math
+from pathlib import Path
+import socket
+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 validate_plan(plan):
+    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()
+    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)
+        if not isinstance(case["request"], dict):
+            raise ValueError("request must be a JSON object")
+        if ("expected_response" in case) == ("expected_response_text" in case):
+            raise ValueError("each case needs exactly one expected response")
+        if "expected_response_text" in case and not isinstance(case["expected_response_text"], str):
+            raise ValueError("expected_response_text must be a string")
+    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")
+    if type(plan["repetitions"]) is not int or plan["repetitions"] != 2:
+        raise ValueError("repetitions must be exactly 2")
+    if type(plan["warmup_per_case"]) is not int or plan["warmup_per_case"] != 2:
+        raise ValueError("warmup_per_case must be exactly 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")
+
+
+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 exchange(endpoint, timeout, case=None):
+    """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
+    try:
+        connection.connect()
+        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={"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):
+                decoded = None
+            if "expected_response_text" in case:
+                matches = bytes(payload) == case["expected_response_text"].encode("utf-8")
+            else:
+                matches = exact_json(decoded, case["expected_response"])
+                # Invalid JSON must fail even when the oracle is JSON null.
+                if decoded is None:
+                    try:
+                        decode_json(bytes(payload))
+                    except (ValueError, UnicodeError):
+                        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 (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()
+    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": time.perf_counter() - started,
+        "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)
+    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": len(records) == planned and len(successful) == planned,
+    }
+
+
+def run(plan_path, output):
+    snapshot = plan_path.read_bytes()
+    plan = decode_json(snapshot)
+    validate_plan(plan)
+    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)
+    save("config.json", {
+        "plan_sha256": hashlib.sha256(snapshot).hexdigest(),
+        "runner_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "plan": plan, "timing": "client wall time including response validation",
+        "retries": 0, "redirects": False, "environment_proxy": False,
+    })
+    parts = urlsplit(plan["endpoint"])
+    health_endpoint = plan.get("health_endpoint", urlunsplit((parts.scheme, parts.netloc, "/health", "", "")))
+    health = exchange(health_endpoint, plan["timeout_seconds"])
+    save("health.json", health)
+    all_records = []
+    rounds = []
+    stopped = threading.Event()
+    lock = threading.Lock()
+    with (output / "responses.jsonl").open("w") as sink:
+        def request(case, phase, index, concurrency=1, repetition=None):
+            record = exchange(plan["endpoint"], plan["timeout_seconds"], case)
+            record.update(case=case["name"], phase=phase, index=index,
+                          concurrency=concurrency, repetition=repetition)
+            with lock:
+                if record["error"]:
+                    stopped.set()
+                all_records.append(record)
+                sink.write(json.dumps(record, allow_nan=False) + "\n")
+                sink.flush()
+            return record
+
+        # Preserve first inference even when the independent health snapshot fails.
+        for case in plan["cases"]:
+            request(case, "readiness", 0)
+            if health["error"] or stopped.is_set():
+                stopped.set()
+                break
+        for case in plan["cases"]:
+            for index in range(plan["warmup_per_case"]):
+                if stopped.is_set():
+                    break
+                request(case, "warmup", 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, plan["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,
+        "excluded_phases": ["health", "readiness", "warmup", "feasibility"],
+    }
+    save("summary.json", summary)
+    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..caadad96
--- /dev/null
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -0,0 +1,242 @@
+import hashlib
+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 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=()):
+    state = {"posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
+    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):
+            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):
+            self.rfile.read(int(self.headers["Content-Length"]))
+            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)
+            )
+            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": {"revision": "pinned"},
+        }
+        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_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_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_still_preserves_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"], 1)
+        self.assertEqual(records[0]["phase"], "readiness")
+        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=3)
+        with self.assertRaises(ValueError):
+            self.execute(path)
+
+    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")
+
+
+if __name__ == "__main__":
+    unittest.main()

From 93071aed201affb3d5ed9f0a8d42f3f6013ff594 Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 8 Oct 2026 10:29:43 +0800
Subject: [PATCH 02/11] benchmarks: alternate frozen request variants within
 matrix cases

---
 benchmarks/serving_matrix.md            | 24 ++++++++++-
 benchmarks/serving_matrix.py            | 56 ++++++++++++++++++-------
 tests/benchmarks/test_serving_matrix.py | 33 ++++++++++++++-
 3 files changed, 94 insertions(+), 19 deletions(-)

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
index 06455aba..71e6ce3a 100644
--- a/benchmarks/serving_matrix.md
+++ b/benchmarks/serving_matrix.md
@@ -47,7 +47,29 @@ with the expectation's UTF-8 encoding, including whitespace. Invalid JSON fails
 a JSON expectation. An HTTP status other than 200 always fails.
 
 The client captures one global health snapshot and one first inference per case,
-then two sequential warmup requests per case. These phases are excluded from
+then two sequential warmup requests per case. A case may instead provide a
+nonempty `variants` list, replacing its top-level request and expectation:
+
+```json
+{
+  "name": "mixed-shapes",
+  "variants": [
+    {"name": "short", "request": {"state": "short input"}, "expected_response": {"answers": {}}},
+    {"name": "long", "request": {"state": "longer input"}, "expected_response": {"answers": {}}}
+  ]
+}
+```
+
+Each variant needs its own unique name within the case, request and exact
+expectation. Each variant receives an independent first inference and two
+warmup requests. Feasibility and measured requests cycle through variants using
+`index % len(variants)`, starting at variant zero in each wave/round. Summaries
+aggregate the mixed case; decision counts use each actual successful response.
+Every inference record includes `variant` and its zero-based `variant_index`;
+an original single-request case uses its case name as its sole variant name.
+Choose a measured request count divisible by the variant count for equal shares.
+
+Readiness, warmup and feasibility phases are excluded from
 performance metrics. Each case/concurrency pair receives one excluded feasibility
 wave of `min(concurrency, requests_per_case)` requests, followed by two measured
 rounds with `requests_per_case` requests each. Fixed worker threads send their
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
index abba44c8..83f708ef 100644
--- a/benchmarks/serving_matrix.py
+++ b/benchmarks/serving_matrix.py
@@ -21,6 +21,10 @@ def reject_constant(value):
     return json.loads(data, parse_constant=reject_constant)
 
 
+def case_variants(case):
+    return case["variants"] if "variants" in case else [case]
+
+
 def validate_plan(plan):
     for key in ("endpoint", "health_endpoint"):
         if key not in plan and key == "health_endpoint":
@@ -38,12 +42,23 @@ def validate_plan(plan):
         if not isinstance(name, str) or not name or name in names:
             raise ValueError("case names must be nonempty and unique")
         names.add(name)
-        if not isinstance(case["request"], dict):
-            raise ValueError("request must be a JSON object")
-        if ("expected_response" in case) == ("expected_response_text" in case):
-            raise ValueError("each case needs exactly one expected response")
-        if "expected_response_text" in case and not isinstance(case["expected_response_text"], str):
-            raise ValueError("expected_response_text must be a string")
+        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")
+            if "expected_response_text" in variant and not isinstance(variant["expected_response_text"], str):
+                raise ValueError("expected_response_text must be a string")
     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:
@@ -196,10 +211,15 @@ def save(name, value):
     stopped = threading.Event()
     lock = threading.Lock()
     with (output / "responses.jsonl").open("w") as sink:
-        def request(case, phase, index, concurrency=1, repetition=None):
-            record = exchange(plan["endpoint"], plan["timeout_seconds"], case)
+        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)
             record.update(case=case["name"], phase=phase, index=index,
-                          concurrency=concurrency, repetition=repetition)
+                          concurrency=concurrency, repetition=repetition,
+                          variant=variant["name"], variant_index=variant_index)
             with lock:
                 if record["error"]:
                     stopped.set()
@@ -210,15 +230,19 @@ def request(case, phase, index, concurrency=1, repetition=None):
 
         # Preserve first inference even when the independent health snapshot fails.
         for case in plan["cases"]:
-            request(case, "readiness", 0)
-            if health["error"] or stopped.is_set():
-                stopped.set()
+            if stopped.is_set():
                 break
-        for case in plan["cases"]:
-            for index in range(plan["warmup_per_case"]):
-                if stopped.is_set():
+            for variant_index in range(len(case_variants(case))):
+                request(case, "readiness", 0, variant_index=variant_index)
+                if health["error"] or stopped.is_set():
+                    stopped.set()
                     break
-                request(case, "warmup", index)
+        for case in plan["cases"]:
+            for variant_index in range(len(case_variants(case))):
+                for index in range(plan["warmup_per_case"]):
+                    if stopped.is_set():
+                        break
+                    request(case, "warmup", index, variant_index=variant_index)
 
         def wave(case, phase, concurrency, count, repetition=None):
             next_index = 0
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
index caadad96..e3342c52 100644
--- a/tests/benchmarks/test_serving_matrix.py
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -17,7 +17,7 @@
 
 
 @contextmanager
-def server(actions=None, health_status=200, barrier_indices=()):
+def server(actions=None, health_status=200, barrier_indices=(), responses_by_tag=None):
     state = {"posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
     lock = threading.Lock()
     barrier = threading.Barrier(len(barrier_indices)) if barrier_indices else None
@@ -34,7 +34,7 @@ def do_GET(self):
             self.wfile.write(b'{"ready":true}')
 
         def do_POST(self):
-            self.rfile.read(int(self.headers["Content-Length"]))
+            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
             with lock:
                 state["posts"] += 1
                 index = state["posts"]
@@ -43,6 +43,8 @@ def do_POST(self):
             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()
             try:
                 if index in barrier_indices:
                     barrier.wait(timeout=5)
@@ -237,6 +239,33 @@ def test_refuses_existing_output(self):
             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"])
+
 
 if __name__ == "__main__":
     unittest.main()

From 116be3f519fb4bb0228e7ec4be3f3aa0b5fa948b Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 8 Oct 2026 10:34:26 +0800
Subject: [PATCH 03/11] benchmarks: reject nonfinite plans and stop on failed
 health

---
 benchmarks/serving_matrix.md            |  9 +++++---
 benchmarks/serving_matrix.py            | 10 ++++++---
 tests/benchmarks/test_serving_matrix.py | 28 +++++++++++++++++++++----
 3 files changed, 37 insertions(+), 10 deletions(-)

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
index 71e6ce3a..85eeffe8 100644
--- a/benchmarks/serving_matrix.md
+++ b/benchmarks/serving_matrix.md
@@ -37,7 +37,10 @@ request and exact response with the workload's reference):
 Case names and positive concurrency values must be unique. `requests_per_case`
 must be at least the largest concurrency value; repetitions and warmup count
 must each equal two. Requests are POSTed unchanged as JSON. There is no token
-injection or additional model selection.
+injection or additional model selection. The complete decoded plan, including
+requests, expectations and metadata, must serialize as finite JSON. Nonfinite
+values, including exponent overflow such as `1e309`, fail before creating output
+or making HTTP requests.
 
 Each case must specify exactly one of `expected_response` or
 `expected_response_text`. JSON expectations compare decoded values recursively,
@@ -77,8 +80,8 @@ next request only after the prior one completes; at most `concurrency` client
 requests are in flight. There is no arrival-rate simulation.
 
 Any error stops new dispatch and subsequent phases. Requests already in flight
-finish and are retained. A failed health snapshot still permits one first
-inference to preserve readiness evidence, then stops the matrix. No exchange is
+finish and are retained. A failed health snapshot is saved with the summary and
+stops the matrix before any readiness inference is dispatched. No exchange is
 retried, redirected, or routed through an environment proxy. Socket connection
 operations use the declared timeout; once connected, the remaining deadline
 also covers request transmission, headers and the complete response body.
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
index 83f708ef..58efd83d 100644
--- a/benchmarks/serving_matrix.py
+++ b/benchmarks/serving_matrix.py
@@ -26,6 +26,10 @@ def case_variants(case):
 
 
 def validate_plan(plan):
+    try:
+        json.dumps(plan, allow_nan=False)
+    except ValueError as exc:
+        raise ValueError("plan must contain only finite JSON values") from exc
     for key in ("endpoint", "health_endpoint"):
         if key not in plan and key == "health_endpoint":
             continue
@@ -209,6 +213,8 @@ def save(name, value):
     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):
@@ -228,14 +234,12 @@ def request(case, phase, index, concurrency=1, repetition=None, variant_index=No
                 sink.flush()
             return record
 
-        # Preserve first inference even when the independent health snapshot fails.
         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 health["error"] or stopped.is_set():
-                    stopped.set()
+                if stopped.is_set():
                     break
         for case in plan["cases"]:
             for variant_index in range(len(case_variants(case))):
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
index e3342c52..4facfea4 100644
--- a/tests/benchmarks/test_serving_matrix.py
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -18,7 +18,7 @@
 
 @contextmanager
 def server(actions=None, health_status=200, barrier_indices=(), responses_by_tag=None):
-    state = {"posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
+    state = {"gets": 0, "posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
     lock = threading.Lock()
     barrier = threading.Barrier(len(barrier_indices)) if barrier_indices else None
 
@@ -27,6 +27,7 @@ def log_message(self, *args):
             pass
 
         def do_GET(self):
+            state["gets"] += 1
             if self.path == "/redirected":
                 state["redirect_hits"] += 1
             self.send_response(health_status)
@@ -186,12 +187,13 @@ def test_does_not_follow_redirect(self):
         self.assertTrue((output / "health.json").exists())
         self.assertEqual(summary["failures"], {"http": 1})
 
-    def test_health_failure_still_preserves_readiness(self):
+    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"], 1)
-        self.assertEqual(records[0]["phase"], "readiness")
+        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})
 
@@ -230,6 +232,24 @@ def test_rejects_invalid_budget_before_creating_output(self):
         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_refuses_existing_output(self):
         path = self.plan("http://127.0.0.1:1")
         output = self.root / "results"

From 8c7df74a3375bfbc6c6f106208b42e5ed9322d7c Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 8 Oct 2026 10:39:37 +0800
Subject: [PATCH 04/11] benchmarks: retain deeply nested failures and validate
 UTF-8 oracles

---
 benchmarks/serving_matrix.md            |  6 +++-
 benchmarks/serving_matrix.py            | 15 +++++++---
 tests/benchmarks/test_serving_matrix.py | 38 +++++++++++++++++++++++++
 3 files changed, 54 insertions(+), 5 deletions(-)

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
index 85eeffe8..0249ed01 100644
--- a/benchmarks/serving_matrix.md
+++ b/benchmarks/serving_matrix.md
@@ -47,7 +47,11 @@ Each case must specify exactly one of `expected_response` or
 including scalar types, with no tolerance or ignored fields. Object key order
 does not matter; array order does. Text expectations compare the response bytes
 with the expectation's UTF-8 encoding, including whitespace. Invalid JSON fails
-a JSON expectation. An HTTP status other than 200 always fails.
+a JSON expectation. Text expectations must be encodable as UTF-8; lone surrogates
+are rejected before output creation or HTTP. With JSON expectations, responses
+that exceed decoding or recursive comparison limits count as correctness failures,
+retaining raw bytes and the failure summary. An HTTP status other than 200 always
+fails.
 
 The client captures one global health snapshot and one first inference per case,
 then two sequential warmup requests per case. A case may instead provide a
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
index 58efd83d..bd034d15 100644
--- a/benchmarks/serving_matrix.py
+++ b/benchmarks/serving_matrix.py
@@ -28,8 +28,8 @@ def case_variants(case):
 def validate_plan(plan):
     try:
         json.dumps(plan, allow_nan=False)
-    except ValueError as exc:
-        raise ValueError("plan must contain only finite JSON values") from exc
+    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
@@ -63,6 +63,11 @@ def validate_plan(plan):
                 raise ValueError("each variant needs exactly one expected response")
             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:
@@ -139,7 +144,7 @@ def expire():
         elif case:
             try:
                 decoded = decode_json(bytes(payload))
-            except (ValueError, UnicodeError):
+            except (ValueError, UnicodeError, RecursionError):
                 decoded = None
             if "expected_response_text" in case:
                 matches = bytes(payload) == case["expected_response_text"].encode("utf-8")
@@ -149,12 +154,14 @@ def expire():
                 if decoded is None:
                     try:
                         decode_json(bytes(payload))
-                    except (ValueError, UnicodeError):
+                    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}"}
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
index 4facfea4..47c86e86 100644
--- a/tests/benchmarks/test_serving_matrix.py
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -250,6 +250,44 @@ def test_nonfinite_plan_values_fail_before_http_or_output(self):
                     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"

From cd530dcc43dd3f3ce564b9b7bae27579c7d53c86 Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 8 Oct 2026 14:13:38 +0800
Subject: [PATCH 05/11] chore: retain only core implementation in PR diff

---
 benchmarks/serving_matrix.md            | 123 ---------
 tests/benchmarks/test_serving_matrix.py | 329 ------------------------
 2 files changed, 452 deletions(-)
 delete mode 100644 benchmarks/serving_matrix.md
 delete mode 100644 tests/benchmarks/test_serving_matrix.py

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
deleted file mode 100644
index 0249ed01..00000000
--- a/benchmarks/serving_matrix.md
+++ /dev/null
@@ -1,123 +0,0 @@
-# Frozen HTTP serving matrix
-
-`serving_matrix.py` is a Python 3.11+ standard-library client for one already
-ready server. It does not launch servers, reserve devices, or collect GPU data.
-Freeze model, server, hardware, workload and cache provenance in `metadata` before
-running; these fields are retained verbatim, not independently verified.
-
-```sh
-python3 benchmarks/serving_matrix.py /path/to/frozen-plan.json \
-  --output /path/to/new-results-directory
-```
-
-The output directory must not exist. A plan has this shape (replace the example
-request and exact response with the workload's reference):
-
-```json
-{
-  "endpoint": "http://127.0.0.1:8000/v1/systemone",
-  "health_endpoint": "http://127.0.0.1:8000/health",
-  "cases": [
-    {
-      "name": "short",
-      "request": {"model": "served-alias", "state": "frozen input", "questions": {}},
-      "expected_response": {"answers": {}}
-    }
-  ],
-  "concurrency": [1, 8, 16],
-  "requests_per_case": 32,
-  "repetitions": 2,
-  "warmup_per_case": 2,
-  "timeout_seconds": 120,
-  "metadata": {"server_revision": "exact commit", "workload_revision": "exact revision"}
-}
-```
-
-`health_endpoint` is optional and defaults to the endpoint origin's `/health`.
-Case names and positive concurrency values must be unique. `requests_per_case`
-must be at least the largest concurrency value; repetitions and warmup count
-must each equal two. Requests are POSTed unchanged as JSON. There is no token
-injection or additional model selection. The complete decoded plan, including
-requests, expectations and metadata, must serialize as finite JSON. Nonfinite
-values, including exponent overflow such as `1e309`, fail before creating output
-or making HTTP requests.
-
-Each case must specify exactly one of `expected_response` or
-`expected_response_text`. JSON expectations compare decoded values recursively,
-including scalar types, with no tolerance or ignored fields. Object key order
-does not matter; array order does. Text expectations compare the response bytes
-with the expectation's UTF-8 encoding, including whitespace. Invalid JSON fails
-a JSON expectation. Text expectations must be encodable as UTF-8; lone surrogates
-are rejected before output creation or HTTP. With JSON expectations, responses
-that exceed decoding or recursive comparison limits count as correctness failures,
-retaining raw bytes and the failure summary. An HTTP status other than 200 always
-fails.
-
-The client captures one global health snapshot and one first inference per case,
-then two sequential warmup requests per case. A case may instead provide a
-nonempty `variants` list, replacing its top-level request and expectation:
-
-```json
-{
-  "name": "mixed-shapes",
-  "variants": [
-    {"name": "short", "request": {"state": "short input"}, "expected_response": {"answers": {}}},
-    {"name": "long", "request": {"state": "longer input"}, "expected_response": {"answers": {}}}
-  ]
-}
-```
-
-Each variant needs its own unique name within the case, request and exact
-expectation. Each variant receives an independent first inference and two
-warmup requests. Feasibility and measured requests cycle through variants using
-`index % len(variants)`, starting at variant zero in each wave/round. Summaries
-aggregate the mixed case; decision counts use each actual successful response.
-Every inference record includes `variant` and its zero-based `variant_index`;
-an original single-request case uses its case name as its sole variant name.
-Choose a measured request count divisible by the variant count for equal shares.
-
-Readiness, warmup and feasibility phases are excluded from
-performance metrics. Each case/concurrency pair receives one excluded feasibility
-wave of `min(concurrency, requests_per_case)` requests, followed by two measured
-rounds with `requests_per_case` requests each. Fixed worker threads send their
-next request only after the prior one completes; at most `concurrency` client
-requests are in flight. There is no arrival-rate simulation.
-
-Any error stops new dispatch and subsequent phases. Requests already in flight
-finish and are retained. A failed health snapshot is saved with the summary and
-stops the matrix before any readiness inference is dispatched. No exchange is
-retried, redirected, or routed through an environment proxy. Socket connection
-operations use the declared timeout; once connected, the remaining deadline
-also covers request transmission, headers and the complete response body.
-Standard-library hostname resolution is not bounded by the socket timeout;
-use a resolved IP endpoint when a strict DNS-independent deadline is required.
-
-Outputs are:
-
-- `plan.json`: exact input bytes.
-- `config.json`: plan SHA256, runner SHA256, metadata and client settings.
-- `health.json`: health status, raw bytes, latency and any failure.
-- `responses.jsonl`: every inference exchange in completion order, with case,
-  phase, zero-based request index, concurrency, measured repetition (1 or 2),
-  status, headers, base64 raw bytes, readable UTF-8 text, latency and error.
-- `summary.json`: global failure counts and a summary for every attempted
-  measured round, including an incomplete round when execution stops.
-
-Round wall time starts before worker creation and ends after all workers finish
-and write their records. Successful requests/s and decisions/s divide successful
-counts by this complete elapsed time, including failures and their drain time.
-A decision is one member of the successful response's `answers` map or list;
-non-JSON text responses have zero counted decisions. Client request latency
-includes transport and response validation. Successful p50/p95 use nearest-rank
-quantiles; failure latencies are listed separately and remain in raw records.
-Round summaries include attempted and planned counts, successful counts, failure
-counts and completion status. Excluded-phase failures appear in global counts.
-Exit status is zero only for a complete error-free matrix; any exchange failure
-returns one after writing the summary. Invalid plans and existing directories
-fail before network requests or output overwrite.
-
-CPU-only verification:
-
-```sh
-python3 -m unittest discover -s tests/benchmarks -p test_serving_matrix.py -v
-```
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
deleted file mode 100644
index 47c86e86..00000000
--- a/tests/benchmarks/test_serving_matrix.py
+++ /dev/null
@@ -1,329 +0,0 @@
-import hashlib
-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 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):
-    state = {"gets": 0, "posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
-    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
-            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"])))
-            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()
-            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": {"revision": "pinned"},
-        }
-        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_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_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=3)
-        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"])
-
-
-if __name__ == "__main__":
-    unittest.main()

From fc96d38a03264485be06de8ef198310d21c66928 Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 8 Oct 2026 14:23:43 +0800
Subject: [PATCH 06/11] test: restore focused regression coverage

---
 tests/benchmarks/test_serving_matrix.py | 329 ++++++++++++++++++++++++
 1 file changed, 329 insertions(+)
 create mode 100644 tests/benchmarks/test_serving_matrix.py

diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
new file mode 100644
index 00000000..47c86e86
--- /dev/null
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -0,0 +1,329 @@
+import hashlib
+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 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):
+    state = {"gets": 0, "posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
+    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
+            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"])))
+            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()
+            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": {"revision": "pinned"},
+        }
+        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_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_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=3)
+        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"])
+
+
+if __name__ == "__main__":
+    unittest.main()

From 9778a75a6de025a367a68baaaa5b0d5b70b8b5ad Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Fri, 9 Oct 2026 08:42:54 +0800
Subject: [PATCH 07/11] fix(bench): require runtime provenance and authenticate
 matrix requests

---
 benchmarks/README.md                    |   6 ++
 benchmarks/serving_matrix.example.json  |  42 ++++++++++
 benchmarks/serving_matrix.md            | 100 ++++++++++++++++++++++++
 benchmarks/serving_matrix.py            |  21 ++++-
 tests/benchmarks/test_serving_matrix.py |  62 ++++++++++++++-
 5 files changed, 224 insertions(+), 7 deletions(-)
 create mode 100644 benchmarks/serving_matrix.example.json
 create mode 100644 benchmarks/serving_matrix.md

diff --git a/benchmarks/README.md b/benchmarks/README.md
index c5788477..454f1d43 100644
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -175,3 +175,9 @@ python -m unittest discover -s tests/benchmarks -p 'test_*.py' -v
 [`laya_mps/`](laya_mps/README.md) holds the scripts behind the numbers of the
 [Apple Silicon recipe](../recipe/laya/apple-silicon.md): in-process and HTTP latency, paired comparisons,
 profiling and the report with its output-parity section.
+
+## 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.
diff --git a/benchmarks/serving_matrix.example.json b/benchmarks/serving_matrix.example.json
new file mode 100644
index 00000000..ec8ed6b7
--- /dev/null
+++ b/benchmarks/serving_matrix.example.json
@@ -0,0 +1,42 @@
+{
+  "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"
+  }
+}
diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
new file mode 100644
index 00000000..1e171e40
--- /dev/null
+++ b/benchmarks/serving_matrix.md
@@ -0,0 +1,100 @@
+# 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`: exactly 2; `warmup_per_case`: exactly 2 per variant.
+- `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`. Record pinned revisions and the actual reserved device. 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.
+
+## 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, two excluded
+warmups per variant, then one excluded feasibility wave at each concurrency and
+exactly two measured rounds. 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 connection establishment, complete response read and
+oracle validation. 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; failure latencies and kinds are retained separately. First-use, capture and
+loading costs must be evaluated separately from warmed results.
+
+Output includes the byte-exact `plan.json`, plan/runner SHA-256 in `config.json`,
+health response, `responses.jsonl` with phase/variant 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
index bd034d15..9d27fbb8 100644
--- a/benchmarks/serving_matrix.py
+++ b/benchmarks/serving_matrix.py
@@ -8,6 +8,7 @@
 import http.client
 import json
 import math
+import os
 from pathlib import Path
 import socket
 import threading
@@ -83,6 +84,13 @@ def validate_plan(plan):
         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",
+    ):
+        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):
@@ -95,7 +103,7 @@ def exact_json(left, right):
     return left == right
 
 
-def exchange(endpoint, timeout, case=None):
+def exchange(endpoint, timeout, case=None, request_headers=None):
     """One connection, no proxy, redirect, pooling or retry; retain raw bytes."""
     started = time.perf_counter()
     parts = urlsplit(endpoint)
@@ -129,7 +137,7 @@ def expire():
         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={"Content-Type": "application/json"} if case else {})
+                           headers={**(request_headers or {}), **({"Content-Type": "application/json"} if case else {})})
         response = connection.getresponse()
         status = response.status
         headers = dict(response.getheaders())
@@ -201,6 +209,11 @@ 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):
@@ -215,7 +228,7 @@ def save(name, value):
     })
     parts = urlsplit(plan["endpoint"])
     health_endpoint = plan.get("health_endpoint", urlunsplit((parts.scheme, parts.netloc, "/health", "", "")))
-    health = exchange(health_endpoint, plan["timeout_seconds"])
+    health = exchange(health_endpoint, plan["timeout_seconds"], request_headers=request_headers)
     save("health.json", health)
     all_records = []
     rounds = []
@@ -229,7 +242,7 @@ def request(case, phase, index, concurrency=1, repetition=None, variant_index=No
             if variant_index is None:
                 variant_index = index % len(variants)
             variant = variants[variant_index]
-            record = exchange(plan["endpoint"], plan["timeout_seconds"], variant)
+            record = exchange(plan["endpoint"], plan["timeout_seconds"], variant, request_headers)
             record.update(case=case["name"], phase=phase, index=index,
                           concurrency=concurrency, repetition=repetition,
                           variant=variant["name"], variant_index=variant_index)
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
index 47c86e86..e7e9ce3a 100644
--- a/tests/benchmarks/test_serving_matrix.py
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -17,8 +17,8 @@
 
 
 @contextmanager
-def server(actions=None, health_status=200, barrier_indices=(), responses_by_tag=None):
-    state = {"gets": 0, "posts": 0, "active": 0, "peak": 0, "redirect_hits": 0}
+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
 
@@ -28,6 +28,11 @@ def log_message(self, *args):
 
         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)
@@ -36,6 +41,11 @@ def do_GET(self):
 
         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"]
@@ -91,7 +101,13 @@ def plan(self, origin, **overrides):
                        "expected_response": EXPECTED}],
             "concurrency": [1, 2], "requests_per_case": 4,
             "repetitions": 2, "warmup_per_case": 2,
-            "timeout_seconds": 2, "metadata": {"revision": "pinned"},
+            "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",
+            },
         }
         plan.update(overrides)
         path = self.root / "input.json"
@@ -106,6 +122,46 @@ def execute(self, path):
         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_closed_loop_counts_metrics_and_provenance(self):
         with server() as (origin, state):
             path = self.plan(origin)

From a9da233f2a6e38150cbd86cc3578a6a1969303ee Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Fri, 9 Oct 2026 21:10:02 +0800
Subject: [PATCH 08/11] fix(bench): compare native timers and report matrix
 measurement scope

---
 benchmarks/serving_matrix.example.json  |   3 +-
 benchmarks/serving_matrix.md            |  76 +++++++--
 benchmarks/serving_matrix.py            | 155 ++++++++++++++++--
 tests/benchmarks/test_serving_matrix.py | 199 +++++++++++++++++++++++-
 4 files changed, 406 insertions(+), 27 deletions(-)

diff --git a/benchmarks/serving_matrix.example.json b/benchmarks/serving_matrix.example.json
index ec8ed6b7..9859dc78 100644
--- a/benchmarks/serving_matrix.example.json
+++ b/benchmarks/serving_matrix.example.json
@@ -37,6 +37,7 @@
     "runtime_revision": "local-test-server-v1",
     "cache_policy": "none",
     "cuda_evidence": "N/A: no CUDA execution",
-    "reservation": "N/A: local CPU fixture"
+    "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
index 1e171e40..f7b2bdfc 100644
--- a/benchmarks/serving_matrix.md
+++ b/benchmarks/serving_matrix.md
@@ -29,11 +29,14 @@ correctness, timeout or transport failure was recorded.
   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`: exactly 2; `warmup_per_case`: exactly 2 per variant.
+- `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`. Record pinned revisions and the actual reserved device. CPU
+  `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.
 
@@ -43,6 +46,31 @@ 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 …`
@@ -54,21 +82,51 @@ and retries are disabled. Use HTTPS for authenticated remote endpoints.
 
 ## Phases and metrics
 
-One health request precedes one readiness request per variant, two excluded
-warmups per variant, then one excluded feasibility wave at each concurrency and
-exactly two measured rounds. Each worker sends its next request after the previous
+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 connection establishment, complete response read and
-oracle validation. Round requests/s and decisions/s use actual successful counts
+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.
+
+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; failure latencies and kinds are retained separately. First-use, capture and
+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.
+The four-request/two-worker example deliberately has no interior-window estimate.
+
+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`,
-health response, `responses.jsonl` with phase/variant and raw base64/text/status/
+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
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
index 9d27fbb8..6034650d 100644
--- a/benchmarks/serving_matrix.py
+++ b/benchmarks/serving_matrix.py
@@ -4,13 +4,17 @@
 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
@@ -26,6 +30,34 @@ 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)
@@ -42,6 +74,11 @@ def validate_plan(plan):
     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:
@@ -62,6 +99,12 @@ def validate_plan(plan):
                 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:
@@ -75,10 +118,10 @@ def validate_plan(plan):
         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")
-    if type(plan["repetitions"]) is not int or plan["repetitions"] != 2:
-        raise ValueError("repetitions must be exactly 2")
-    if type(plan["warmup_per_case"]) is not int or plan["warmup_per_case"] != 2:
-        raise ValueError("warmup_per_case must be exactly 2")
+    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")
@@ -86,7 +129,7 @@ def validate_plan(plan):
         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",
+        "runtime_revision", "cache_policy", "cuda_evidence", "reservation", "host",
     ):
         value = plan["metadata"].get(name)
         if not value or (isinstance(value, str) and not value.strip()):
@@ -103,7 +146,25 @@ def exact_json(left, right):
     return left == right
 
 
-def exchange(endpoint, timeout, case=None, request_headers=None):
+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)
@@ -117,8 +178,11 @@ def exchange(endpoint, timeout, case=None, request_headers=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")
@@ -157,7 +221,7 @@ def expire():
             if "expected_response_text" in case:
                 matches = bytes(payload) == case["expected_response_text"].encode("utf-8")
             else:
-                matches = exact_json(decoded, case["expected_response"])
+                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:
@@ -177,11 +241,13 @@ def expire():
         if timer:
             timer.cancel()
         connection.close()
+    completed_at = time.perf_counter()
     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": time.perf_counter() - started,
+        "latency_seconds": completed_at - started, "connect_seconds": connect_seconds,
+        "completed_at_monotonic_seconds": completed_at,
         "successful_decisions": decisions, "error": error,
     }
 
@@ -191,6 +257,24 @@ def round_summary(records, elapsed, case, concurrency, repetition, planned):
     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),
@@ -201,10 +285,40 @@ def round_summary(records, elapsed, case, concurrency, repetition, planned):
         "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": len(records) == planned and len(successful) == planned,
+        "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)
@@ -220,12 +334,19 @@ def save(name, value):
         (output / name).write_text(json.dumps(value, indent=2, allow_nan=False) + "\n")
 
     (output / "plan.json").write_bytes(snapshot)
-    save("config.json", {
+    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 response validation",
+        "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)
@@ -242,7 +363,8 @@ def request(case, phase, index, concurrency=1, repetition=None, variant_index=No
             if variant_index is None:
                 variant_index = index % len(variants)
             variant = variants[variant_index]
-            record = exchange(plan["endpoint"], plan["timeout_seconds"], variant, request_headers)
+            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)
@@ -263,7 +385,7 @@ def request(case, phase, index, concurrency=1, repetition=None, variant_index=No
                     break
         for case in plan["cases"]:
             for variant_index in range(len(case_variants(case))):
-                for index in range(plan["warmup_per_case"]):
+                for index in range(warmup):
                     if stopped.is_set():
                         break
                     request(case, "warmup", index, variant_index=variant_index)
@@ -296,7 +418,7 @@ def worker():
                 if stopped.is_set():
                     break
                 wave(case, "feasibility", concurrency, min(concurrency, plan["requests_per_case"]))
-                for repetition in range(1, plan["repetitions"] + 1):
+                for repetition in range(1, repetitions + 1):
                     if stopped.is_set():
                         break
                     records, elapsed = wave(case, "measured", concurrency, plan["requests_per_case"], repetition)
@@ -306,9 +428,12 @@ def worker():
         "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
 
 
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
index e7e9ce3a..fe8e1a13 100644
--- a/tests/benchmarks/test_serving_matrix.py
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -1,4 +1,9 @@
 import hashlib
+import base64
+from datetime import datetime
+import platform
+import socket
+import statistics
 import json
 import os
 import tempfile
@@ -56,6 +61,8 @@ def do_POST(self):
             )
             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)
@@ -106,7 +113,7 @@ def plan(self, origin, **overrides):
                 "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",
+                "reservation": "N/A: local CPU test", "host": "synthetic-loopback",
             },
         }
         plan.update(overrides)
@@ -162,6 +169,164 @@ def test_invalid_environment_bearer_token_fails_before_network_or_output(self):
                     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)
@@ -284,7 +449,7 @@ def test_rejects_invalid_budget_before_creating_output(self):
         with self.assertRaises(ValueError):
             self.execute(path)
         self.assertFalse((self.root / "results").exists())
-        path = self.plan("http://127.0.0.1:1", repetitions=3)
+        path = self.plan("http://127.0.0.1:1", repetitions=1)
         with self.assertRaises(ValueError):
             self.execute(path)
 
@@ -381,5 +546,35 @@ def test_mixed_variants_alternate_and_count_actual_decisions(self):
             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()

From cbdd473d4386cbfc3b24aaa56e0113d80f1dc671 Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Sat, 10 Oct 2026 22:06:21 +0800
Subject: [PATCH 09/11] docs(benchmarks): clarify short interior windows

---
 benchmarks/serving_matrix.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
index f7b2bdfc..3ae26293 100644
--- a/benchmarks/serving_matrix.md
+++ b/benchmarks/serving_matrix.md
@@ -109,7 +109,8 @@ interior completion window: sort retained monotonic completion timestamps, exclu
 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.
-The four-request/two-worker example deliberately has no interior-window estimate.
+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

From 4f2e34d78499e945c867e4a2dea3a100e8236bde Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Tue, 13 Oct 2026 13:27:19 +0800
Subject: [PATCH 10/11] fix(bench): enforce matrix deadline through validation
 and close

---
 benchmarks/serving_matrix.md            |  7 ++++
 benchmarks/serving_matrix.py            |  4 ++
 tests/benchmarks/test_serving_matrix.py | 53 ++++++++++++++++++++++++-
 3 files changed, 63 insertions(+), 1 deletion(-)

diff --git a/benchmarks/serving_matrix.md b/benchmarks/serving_matrix.md
index 3ae26293..4c64f88e 100644
--- a/benchmarks/serving_matrix.md
+++ b/benchmarks/serving_matrix.md
@@ -96,6 +96,13 @@ 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
diff --git a/benchmarks/serving_matrix.py b/benchmarks/serving_matrix.py
index 6034650d..37c04085 100644
--- a/benchmarks/serving_matrix.py
+++ b/benchmarks/serving_matrix.py
@@ -242,6 +242,10 @@ def expire():
             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"),
diff --git a/tests/benchmarks/test_serving_matrix.py b/tests/benchmarks/test_serving_matrix.py
index fe8e1a13..5f813a60 100644
--- a/tests/benchmarks/test_serving_matrix.py
+++ b/tests/benchmarks/test_serving_matrix.py
@@ -13,7 +13,7 @@
 from contextlib import contextmanager
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 from benchmarks import serving_matrix
 
@@ -393,6 +393,57 @@ def test_total_body_deadline_retains_partial_raw_bytes(self):
         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):

From 2d25d16b5a3d6842a76438918ede3993526d3f8e Mon Sep 17 00:00:00 2001
From: levius <2114377220@qq.com>
Date: Thu, 15 Oct 2026 09:43:59 +0800
Subject: [PATCH 11/11] docs(bench): keep matrix guide independent of Score
 precision addition

---
 benchmarks/README.md | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/benchmarks/README.md b/benchmarks/README.md
index 454f1d43..71dc35ea 100644
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -170,14 +170,14 @@ result is not a native-engine speedup. Metal and vision benchmarks follow CUDA.
 python -m unittest discover -s tests/benchmarks -p 'test_*.py' -v
 ```
 
-## Laya on Apple Silicon
-
-[`laya_mps/`](laya_mps/README.md) holds the scripts behind the numbers of the
-[Apple Silicon recipe](../recipe/laya/apple-silicon.md): in-process and HTTP latency, paired comparisons,
-profiling and the report with its output-parity section.
-
 ## 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
+[Apple Silicon recipe](../recipe/laya/apple-silicon.md): in-process and HTTP latency, paired comparisons,
+profiling and the report with its output-parity section.