From 4b9cccbde841272ff5ace6aee3cf51610fe1324f Mon Sep 17 00:00:00 2001 From: linear3735 Date: Sat, 10 Oct 2026 19:14:22 +0800 Subject: [PATCH 1/7] bench: preserve complete serving evidence and offline summaries Retain planned, attempted, failed and incomplete requests and recompute saved rounds without inference. Changed lines: 886 authored code/test lines and 100 documentation lines (986 total additions + deletions). No third-party source copied. --- benchmarks/README.md | 100 ++++++- benchmarks/bench.py | 382 ++++++++++++++++++++++--- tests/benchmarks/test_bench.py | 504 ++++++++++++++++++++++++++++++++- 3 files changed, 943 insertions(+), 43 deletions(-) diff --git a/benchmarks/README.md b/benchmarks/README.md index c5788477..2200d6ec 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -61,7 +61,7 @@ python benchmarks/bench.py run requests.jsonl \ python benchmarks/bench.py run requests.jsonl \ --endpoint http://127.0.0.1:8000/v1/systemone --model english \ --metadata run-metadata.json --phase measured \ - --concurrency 1 --output /tmp/laya-c1-run1 + --concurrency 1 --warmup 5 --output /tmp/laya-c1-run1 ``` Use the checkpoint's served alias for vllm-jev/native implementations. For the @@ -70,6 +70,15 @@ only the endpoint. Optional authentication comes from `OMNI_JEV_TEST_TOKEN`; the token is not saved. Redirects, proxy environment variables and retries are disabled. `--timeout` is a total deadline for each HTTP exchange, including body receipt; response validation is included in recorded client latency. +Each answer must be an object whose `type` matches the requested question type. +Missing or incorrect types are `invalid_response`; the runner keeps the original +body and does not supply missing fields. +Online validation and offline replay both parse the saved HTTPX-decoded +`response.text` with `json.loads`. No BOM is stripped or missing field supplied. +An initial U+FEFF is rejected by string JSON parsing. HTTPX may replace invalid +encoded bytes with U+FFFD; if that exact saved text is valid JSON and passes the +same schema checks, it is accepted. This is a decoded-text contract, not a +claim that the original HTTP bytes were valid UTF-8. Start with the weight-free preflight, then one feasibility run per configuration. Exclude feasibility from performance comparisons. Use two measured runs at each @@ -77,6 +86,19 @@ of concurrency 1, 8 and 16, with the same request ordering. Reuse the server; report variation rather than silently extending an inconclusive run budget. Use separate output directories; existing results are never overwritten. +For the initial frozen 231-task text campaign, predeclare three complete measured +traversals at concurrency 1, with five warmup requests per traversal and the same +running server. Invoke `run` separately for each output directory. Repeating the +manifest three times still represents 231 independent tasks, not 693 tasks. + +By default warmup cycles through the measured manifest. Use +`--warmup-manifest warmup-requests.jsonl` to supply independent labelled requests +in the same format. They are saved with their checksum and never enter measured +counts. The first entry in `warmup.json` is the first request observed by this +client invocation. It is not a measurement of server process startup, model +loading or service-internal warmup; save those boundaries externally. An +independent warmup manifest does not establish image-path or cache coverage. + Concurrency is a closed-loop count of in-flight HTTP requests. Fixed workers consume the manifest in order; completion order is naturally variable. Each worker sends its next request only when the previous one completes. There is no @@ -86,19 +108,81 @@ Outputs: - `config.json`: manifest/runner SHA256, runtime metadata and client settings. - `requests.jsonl`: exact manifest snapshot. +- `warmup-requests.jsonl`: independent warmup snapshot, when supplied. - `warmup.json`: first-inference readiness check and warmup responses, excluded - from load timing. Warmup failure stops the run. -- `responses.jsonl`: one record per input, in input order, with raw response, - HTTP status, client latency, validated answers and any error. + from load timing. Warmup failure stops the run and saves completion/summary + evidence that all measured requests are unattempted. +- `responses.jsonl`: flushed after each request finishes, in completion order + during replay, then atomically reordered to manifest input order for a complete + traversal. Interrupted runs retain their actual saved completion order, + with the frozen request ID, original response text and its SHA256, HTTP status, + validation result, client latency and any error. Join records to the manifest + by ID; `compare` accepts different completion orders. +- `completion.json`: atomically replaced before each measured request starts and + after it finishes. Contains attempted/completed/active IDs, input/config/response + checksums, measured start, terminal stop reason and real measured wall time. + A failed/interrupted warmup has no measured start/finish/wall or responses file; + `stop_reason` is `warmup_failed` or `warmup_interrupted`, with no attempted IDs. - `summary.json`: successful requests/s and decisions/s, successful-request p50/p95 latency (nearest-rank p95), failure counts by kind, and quality metrics - with denominators. A decision is a question, not an internal candidate branch. + with denominators, failure latency p50/p95 and planned/attempted/unattempted/ + incomplete request and decision counts. A decision is a question, not an + internal candidate branch. Any request error makes the run exit nonzero after saving results. Malformed responses count as failures. Quality and latency summaries cover successful -requests only; failure latencies remain in raw results. Always publish failure -counts alongside those summaries. Different failure rates invalidate a simple -speedup claim. +requests only; completed failure latencies have a separate summary. Always publish +failure counts alongside those summaries. Different failure rates invalidate a +simple speedup claim. + +The request clock spans POST through complete body receipt, JSON parsing and wire +schema validation (`latency_basis=post_to_body_json_and_schema_validation`). It +excludes result hashing and persistence. Measured wall includes per-request +hashing, start/completion saves and closed-loop client bookkeeping; throughput is +successful requests or decisions divided by this saved wall. The real elapsed +time includes the complete traversal's response reorder/save/checksum; final +terminal-state and summary writes occur after that boundary. Use the same +collector version and persistence policy across configurations. Do not derive speedup ratios against +historical body-only client timing. + +A soft interruption saves finished samples and records cancelled requests as +`incomplete`, with `elapsed_until_abort_ms` and no completed `latency_ms`. If a +complete body was received before cancellation, its text remains available for +independent offline scoring. Abort durations do not enter latency percentiles. +Cancellation of the client does not prove the server stopped processing. + +Request counts obey `planned = attempted + unattempted` and +`attempted = successful + failed + incomplete`; decision counts follow the same +partition. `failed` refers to completed failures. The retained `requests` field +counts persisted response records, including explicit incomplete records. +`evidence_complete` means a terminal, complete traversal, including any failures; +it does not certify model quality or a performance comparison. Conditional quality +is this tool's simple metric and must not replace a dataset's pinned scorer. + +## Recompute a saved summary + +```sh +python benchmarks/bench.py summary /tmp/laya-c1-run1 +``` + +This command makes no requests and prints the recomputed summary without changing +saved files. It checks frozen IDs, duplicates and unknown entries, saved manifest, +config and response checksums, and agreement between original response text and +normalized answers/validation. It requires the new `completion.json` evidence; +older runs remain usable by `compare` but cannot acquire unrecorded timing or +interruption boundaries through this command. + +After a hard kill, a valid start snapshot can distinguish unattempted IDs from +attempted IDs without completed records. Without a saved terminal boundary, +`wall_seconds` and both throughput fields are null and `evidence_complete` is +false. Partial files, checksum disagreement or missing claimed completed records +are rejected; inspect the retained originals instead of repairing timing or +declaring a complete baseline. Hashes detect inconsistency, not authenticity +against someone rewriting all evidence and hashes. Interrupted runs provide +partial observations with explicit coverage, not complete traversals. +Flushing records and atomically replacing state cover process interruption/kill +for the successfully written records. They do not promise machine power-loss +durability; this tool does not call `fsync`. ## Compare output fidelity diff --git a/benchmarks/bench.py b/benchmarks/bench.py index ae132888..4ab8403a 100644 --- a/benchmarks/bench.py +++ b/benchmarks/bench.py @@ -20,7 +20,17 @@ def digest(path): def write_json(path, value): - path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") + temporary.replace(path) + + +def response_digest(text): + if text is None: + return None + if not isinstance(text, str): + raise ValueError("raw response must be text") + return hashlib.sha256(text.encode("utf-8")).hexdigest() def load_cases(path): @@ -85,12 +95,18 @@ def sum_allowance(count): def read_answers(case, payload): """Validate wire results and normalize probabilities by label, never by order.""" answers = payload["answers"] - if set(answers) != set(case["request"]["questions"]): + if not isinstance(answers, dict) or set(answers) != set( + case["request"]["questions"] + ): raise ValueError("response question IDs do not match the request") normalized = {} for qid, question in case["request"]["questions"].items(): answer = answers[qid] kind = question["type"] + if not isinstance(answer, dict) or answer.get("type") != kind: + raise ValueError( + "response answer must be an object with the requested type" + ) if kind == "noul": value = answer["noul"] if not number(value): @@ -128,8 +144,10 @@ def read_answers(case, payload): def quality(cases, records): values = {"choice_accuracy": [], "noul_accuracy": [], "score_mae": []} - for case, record in zip(cases, records): - if record["error"]: + by_id = {r["id"]: r for r in records} + for case in cases: + record = by_id.get(case["id"]) + if record is None or record["error"] is not None: continue for qid, target in case["expected"].items(): value = record["answers"][qid]["value"] @@ -145,30 +163,70 @@ def quality(cases, records): } -def summarize(cases, records, elapsed): - successful = [r for r in records if r["error"] is None] +def summarize(cases, records, elapsed, attempted_ids=None): + attempted = set( + attempted_ids if attempted_ids is not None else (r["id"] for r in records) + ) + completed = [r for r in records if r.get("state", "completed") == "completed"] + successful = [r for r in completed if r["error"] is None] + failed = [r for r in completed if r["error"] is not None] + completed_ids = {r["id"] for r in completed} + incomplete = attempted - completed_ids latencies = sorted(r["latency_ms"] for r in successful) + failed_latencies = sorted(r["latency_ms"] for r in failed) decisions = sum(len(r["answers"]) for r in successful) + counts = { + "planned": 0, + "attempted": 0, + "unattempted": 0, + "incomplete": 0, + "failed": 0, + } + failed_ids = {r["id"] for r in failed} + for case in cases: + size = len(case["request"]["questions"]) + counts["planned"] += size + counts["attempted" if case["id"] in attempted else "unattempted"] += size + if case["id"] in incomplete: + counts["incomplete"] += size + if case["id"] in failed_ids: + counts["failed"] += size errors = {} - for record in records: - if record["error"]: - key = record["error_kind"] - errors[key] = errors.get(key, 0) + 1 + for record in failed: + key = record["error_kind"] + errors[key] = errors.get(key, 0) + 1 return { "requests": len(records), + "planned_requests": len(cases), + "attempted_requests": len(attempted), + "unattempted_requests": len(cases) - len(attempted), + "incomplete_requests": len(incomplete), + "completed_requests": len(completed), "successful_requests": len(successful), - "failed_requests": len(records) - len(successful), + "failed_requests": len(failed), "errors": errors, + **{name + "_decisions": count for name, count in counts.items()}, "successful_decisions": decisions, + "completed_decisions": decisions + counts["failed"], "wall_seconds": elapsed, - "requests_per_second": len(successful) / elapsed, - "decisions_per_second": decisions / elapsed, + "requests_per_second": len(successful) / elapsed if elapsed else None, + "decisions_per_second": decisions / elapsed if elapsed else None, + "evidence_complete": elapsed is not None and len(completed) == len(cases), + "latency_basis": "post_to_body_json_and_schema_validation", "successful_latency_p50_ms": statistics.median(latencies) if latencies else None, "successful_latency_p95_ms": latencies[math.ceil(0.95 * len(latencies)) - 1] if latencies else None, + "failed_latency_p50_ms": statistics.median(failed_latencies) + if failed_latencies + else None, + "failed_latency_p95_ms": failed_latencies[ + math.ceil(0.95 * len(failed_latencies)) - 1 + ] + if failed_latencies + else None, "quality_on_successful_requests": quality(cases, records), } @@ -182,6 +240,7 @@ async def request_one(client, endpoint, case, model, timeout): "error_kind": None, "response": None, "answers": {}, + "state": "completed", } try: async with asyncio.timeout(timeout): @@ -191,7 +250,7 @@ async def request_one(client, endpoint, case, model, timeout): record["status"] = response.status_code record["response"] = response.text response.raise_for_status() - record["answers"] = read_answers(case, response.json()) + record["answers"] = read_answers(case, json.loads(record["response"])) except httpx.HTTPStatusError as exc: record.update(error=str(exc), error_kind=f"http_{record['status']}") except (TimeoutError, httpx.TimeoutException) as exc: @@ -202,24 +261,112 @@ async def request_one(client, endpoint, case, model, timeout): record.update(error=str(exc), error_kind="transport") except (ValueError, KeyError, TypeError, AttributeError) as exc: record.update(error=str(exc), error_kind="invalid_response") - record["latency_ms"] = (time.perf_counter() - started) * 1000 + except asyncio.CancelledError: + record.update( + state="incomplete", + error="request cancelled before response validation completed", + error_kind="interrupted", + ) + duration = (time.perf_counter() - started) * 1000 + record["latency_ms"] = duration if record["state"] == "completed" else None + if record["state"] == "incomplete": + record["elapsed_until_abort_ms"] = duration + record["response_sha256"] = response_digest(record["response"]) return record -async def replay(client, endpoint, cases, model, concurrency, timeout): +def measurement_state(output): + state = { + "started_at": None, + "finished_at": None, + "wall_seconds": None, + "stop_reason": None, + "attempted_ids": [], + "completed_ids": [], + "active_ids": [], + "responses_sha256": None, + } + if output is not None: + state["manifest_sha256"] = digest(output / "requests.jsonl") + state["config_sha256"] = digest(output / "config.json") + return state + + +async def replay(client, endpoint, cases, model, concurrency, timeout, output=None): pending = iter(enumerate(cases)) records = [None] * len(cases) + state = measurement_state(output) + responses = None + response_hash = hashlib.sha256() + if output is not None: + state["responses_sha256"] = response_hash.hexdigest() + responses = (output / "responses.jsonl").open( + "w", encoding="utf-8", newline="" + ) + + def save_state(): + if output is not None: + write_json(output / "completion.json", state) async def worker(): for index, case in pending: + state["attempted_ids"].append(case["id"]) + state["active_ids"].append(case["id"]) + save_state() records[index] = await request_one(client, endpoint, case, model, timeout) + record = records[index] + if responses is not None: + line = json.dumps(record, allow_nan=False) + "\n" + responses.write(line) + responses.flush() + response_hash.update(line.encode("utf-8")) + state["responses_sha256"] = response_hash.hexdigest() + if record["state"] == "completed": + state["completed_ids"].append(case["id"]) + state["active_ids"].remove(case["id"]) + save_state() + if record["state"] == "incomplete": + raise asyncio.CancelledError started = time.perf_counter() - await asyncio.gather(*(worker() for _ in range(concurrency))) - return records, time.perf_counter() - started + state["started_at"] = time.time() + save_state() + tasks = [asyncio.create_task(worker()) for _ in range(concurrency)] + try: + await asyncio.gather(*tasks) + if responses is not None: + responses.close() + responses = None + path = output / "responses.jsonl" + temporary = path.with_suffix(".jsonl.tmp") + temporary.write_text( + "".join(json.dumps(r, allow_nan=False) + "\n" for r in records), + encoding="utf-8", + newline="", + ) + temporary.replace(path) + state["responses_sha256"] = digest(path) + state["stop_reason"] = "completed" + except BaseException as exc: + state["stop_reason"] = ( + "interrupted" if isinstance(exc, asyncio.CancelledError) else "error" + ) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + finally: + state["wall_seconds"] = time.perf_counter() - started + state["finished_at"] = time.time() + if responses is not None: + responses.close() + save_state() + return records, state["wall_seconds"] async def run(args, cases): + warmup_manifest = getattr(args, "warmup_manifest", None) + warmup_cases = load_cases(warmup_manifest) if warmup_manifest else cases metadata = json.loads(args.metadata.read_text()) for name in ( "gpu", @@ -247,12 +394,20 @@ async def run(args, cases): "phase": args.phase, "concurrency": args.concurrency, "warmup": args.warmup, + "warmup_manifest_sha256": digest(warmup_manifest) + if warmup_manifest + else None, "timeout_seconds": args.timeout, + "latency_basis": "post_to_body_json_and_schema_validation", "python": platform.python_version(), "httpx": httpx.__version__, }, ) (args.output / "requests.jsonl").write_bytes(args.manifest.read_bytes()) + if warmup_manifest: + (args.output / "warmup-requests.jsonl").write_bytes( + warmup_manifest.read_bytes() + ) headers = {} if token := os.environ.get("OMNI_JEV_TEST_TOKEN"): headers["Authorization"] = f"Bearer {token}" @@ -271,25 +426,174 @@ async def run(args, cases): await request_one( client, args.endpoint, - cases[i % len(cases)], + warmup_cases[i % len(warmup_cases)], args.model, args.timeout, ) ) - if warmup[-1]["error"]: + if warmup[-1]["error"] is not None: write_json(args.output / "warmup.json", warmup) + state = measurement_state(args.output) + state["stop_reason"] = ( + "warmup_interrupted" + if warmup[-1]["state"] == "incomplete" + else "warmup_failed" + ) + write_json(args.output / "completion.json", state) + write_json(args.output / "summary.json", summarize_saved(args.output)) raise ValueError("readiness/warmup failed; see warmup.json") write_json(args.output / "warmup.json", warmup) - records, elapsed = await replay( - client, args.endpoint, cases, args.model, args.concurrency, args.timeout - ) - (args.output / "responses.jsonl").write_text( - "".join(json.dumps(r, allow_nan=False) + "\n" for r in records) - ) - summary = summarize(cases, records, elapsed) + try: + await replay( + client, + args.endpoint, + cases, + args.model, + args.concurrency, + args.timeout, + args.output, + ) + except asyncio.CancelledError: + # Replay has saved the completed samples and the measured stop boundary. + pass + summary = summarize_saved(args.output) write_json(args.output / "summary.json", summary) print(json.dumps(summary, indent=2)) - return int(summary["failed_requests"] > 0) + return int(summary["failed_requests"] > 0 or not summary["evidence_complete"]) + + +def summarize_saved(path): + """Recompute from this runner's saved inputs, responses and timing evidence.""" + config = json.loads((path / "config.json").read_text()) + state = json.loads((path / "completion.json").read_text()) + if digest(path / "config.json") != state["config_sha256"]: + raise ValueError("saved config checksum does not match completion") + if config["latency_basis"] != "post_to_body_json_and_schema_validation": + raise ValueError("unsupported saved latency basis") + manifest_hash = digest(path / "requests.jsonl") + if ( + manifest_hash != config["manifest_sha256"] + or manifest_hash != state["manifest_sha256"] + ): + raise ValueError("saved manifest checksum does not match config/completion") + if config["warmup_manifest_sha256"] is not None: + if digest(path / "warmup-requests.jsonl") != config["warmup_manifest_sha256"]: + raise ValueError("saved warmup manifest checksum does not match config") + responses_path = path / "responses.jsonl" + if state["started_at"] is None: + if state["responses_sha256"] is not None or responses_path.exists(): + raise ValueError("unstarted measurement must not claim responses") + records = [] + elif digest(responses_path) != state["responses_sha256"]: + raise ValueError("saved responses checksum does not match completion") + else: + records = [json.loads(line) for line in responses_path.read_text().splitlines()] + cases = load_cases(path / "requests.jsonl") + by_id = {c["id"]: c for c in cases} + ids = {} + for name in ("attempted_ids", "completed_ids", "active_ids"): + values = state[name] + if len(values) != len(set(values)) or not set(values) <= set(by_id): + raise ValueError(f"duplicate or unknown {name}") + ids[name] = set(values) + if not ids["completed_ids"] <= ids["attempted_ids"]: + raise ValueError("completed IDs must have a saved start") + if not ids["active_ids"] <= ids["attempted_ids"] - ids["completed_ids"]: + raise ValueError("active IDs must be attempted and incomplete") + record_ids = [r["id"] for r in records] + if ( + len(record_ids) != len(set(record_ids)) + or not set(record_ids) <= ids["attempted_ids"] + ): + raise ValueError("duplicate, unknown or unattempted response ID") + completed_ids = set() + for record in records: + if record["response_sha256"] != response_digest(record["response"]): + raise ValueError("raw response checksum mismatch") + if record["state"] == "incomplete": + if ( + record["latency_ms"] is not None + or record["error_kind"] != "interrupted" + or not record["error"] + or record["answers"] + ): + raise ValueError("incomplete response must not claim a latency/answer") + if ( + not number(record["elapsed_until_abort_ms"]) + or record["elapsed_until_abort_ms"] < 0 + ): + raise ValueError("invalid elapsed-until-abort") + continue + if ( + record["state"] != "completed" + or not number(record["latency_ms"]) + or record["latency_ms"] < 0 + ): + raise ValueError("invalid completed response or latency") + completed_ids.add(record["id"]) + answers, error_kind = {}, None + status = record["status"] + if status is None: + if record["response"] is not None or record["error_kind"] not in ( + "timeout", "transport" + ): + raise ValueError("response without status must be a client failure") + error_kind = record["error_kind"] + elif ( + not isinstance(status, int) + or isinstance(status, bool) + or not 100 <= status <= 599 + ): + raise ValueError("invalid HTTP status") + elif not 200 <= status < 300: + error_kind = f"http_{status}" + else: + try: + answers = read_answers( + by_id[record["id"]], json.loads(record["response"]) + ) + except (ValueError, KeyError, TypeError, AttributeError): + error_kind = "invalid_response" + if ( + record["answers"] != answers + or record["error_kind"] != error_kind + or (record["error"] is None) != (error_kind is None) + ): + raise ValueError("raw response and saved validation result disagree") + if completed_ids != ids["completed_ids"]: + raise ValueError("completed IDs do not match saved responses") + elapsed = state["wall_seconds"] + if state["started_at"] is None: + if ( + state["finished_at"] is not None + or elapsed is not None + or any(ids.values()) + or state["stop_reason"] not in ("warmup_failed", "warmup_interrupted") + ): + raise ValueError("invalid unstarted measurement") + elif not number(state["started_at"]): + raise ValueError("invalid measured start boundary") + elif state["finished_at"] is None: + if elapsed is not None or state["stop_reason"] is not None: + raise ValueError("unfinished measurement must not claim a terminal wall") + else: + if ( + not number(state["finished_at"]) + or state["finished_at"] < state["started_at"] + or not number(elapsed) + or elapsed <= 0 + or ids["active_ids"] + or state["stop_reason"] not in ("completed", "interrupted", "error") + ): + raise ValueError("invalid measured completion boundary") + if state["stop_reason"] == "completed" and completed_ids != set(by_id): + raise ValueError("completed traversal is missing response IDs") + summary = summarize(cases, records, elapsed, ids["attempted_ids"]) + summary["stop_reason"] = state["stop_reason"] + summary["evidence_complete"] = ( + summary["evidence_complete"] and state["stop_reason"] == "completed" + ) + return summary def compare(args): @@ -306,13 +610,15 @@ def compare(args): [json.loads(line) for line in (p / "responses.jsonl").read_text().splitlines()] for p in (args.reference, args.candidate) ] - if [r["id"] for r in runs[0]] != [r["id"] for r in runs[1]]: - raise ValueError("response IDs differ") drift, flips, score_drift, failed, compared = 0.0, 0, 0.0, 0, 0 cases = load_cases(args.reference / "requests.jsonl") - if [c["id"] for c in cases] != [r["id"] for r in runs[0]]: - raise ValueError("response IDs do not cover the manifest") - for case, ref, candidate in zip(cases, *runs): + ordered = [] + for records in runs: + by_id = {r["id"]: r for r in records} + if len(by_id) != len(records) or set(by_id) != {c["id"] for c in cases}: + raise ValueError("response IDs do not uniquely cover the manifest") + ordered.append([by_id[c["id"]] for c in cases]) + for case, ref, candidate in zip(cases, *ordered): if ref["error"] or candidate["error"]: failed += 1 continue @@ -381,6 +687,9 @@ def main(): runner.add_argument("--phase", choices=("feasibility", "measured"), required=True) runner.add_argument("--concurrency", type=positive, default=1) runner.add_argument("--warmup", type=positive, default=3) + runner.add_argument( + "--warmup-manifest", type=Path, help="independent labelled warmup requests" + ) runner.add_argument("--timeout", type=positive, default=60) comparison = commands.add_parser("compare") comparison.add_argument("reference", type=Path) @@ -388,8 +697,15 @@ def main(): comparison.add_argument("--max-probability-drift", type=float, required=True) comparison.add_argument("--max-score-drift", type=float, required=True) comparison.add_argument("--max-flips", type=int, required=True) + summary = commands.add_parser( + "summary", help="recompute a saved run without HTTP requests" + ) + summary.add_argument("output", type=Path) args = parser.parse_args() try: + if args.command == "summary": + print(json.dumps(summarize_saved(args.output), indent=2, allow_nan=False)) + return 0 if args.command == "compare": tolerances = ( args.max_probability_drift, diff --git a/tests/benchmarks/test_bench.py b/tests/benchmarks/test_bench.py index 3ca17915..508189e8 100644 --- a/tests/benchmarks/test_bench.py +++ b/tests/benchmarks/test_bench.py @@ -21,7 +21,7 @@ def response_for(case): kind = question["type"] target = case["expected"][qid] if kind == "noul": - answers[qid] = {"noul": float(target)} + answers[qid] = {"type": kind, "noul": float(target)} else: labels = ( list(question["criteria"]) @@ -29,13 +29,105 @@ def response_for(case): else [str(i) for i in range(len(question["criteria"]))] ) answers[qid] = { + "type": kind, kind: target, "probabilities": {k: float(k == str(target)) for k in labels}, } return {"answers": answers} +def saved_inputs(path, cases=CASES): + path.mkdir() + (path / "requests.jsonl").write_text("".join(json.dumps(c) + "\n" for c in cases)) + bench.write_json( + path / "config.json", + { + "manifest_sha256": bench.digest(path / "requests.jsonl"), + "warmup_manifest_sha256": None, + "latency_basis": "post_to_body_json_and_schema_validation", + }, + ) + + +def run_args(root): + metadata = { + key: "mock-only" + for key in ( + "gpu", "gpu_ids", "driver", "cuda", "precision", "model_revision", + "runtime_revision", "cache_policy", "cuda_evidence", "reservation", + ) + } + bench.write_json(root / "metadata.json", metadata) + return argparse.Namespace( + metadata=root / "metadata.json", output=root / "run", + manifest=Path(bench.__file__).with_name("smoke.jsonl"), + endpoint="http://localhost/v1/systemone", model="english", + phase="measured", concurrency=1, warmup=1, timeout=1, + warmup_manifest=None, + ) + + class ReplayTests(unittest.IsolatedAsyncioTestCase): + async def test_legacy_run_namespace_without_optional_warmup_manifest(self): + with tempfile.TemporaryDirectory() as directory: + args = run_args(Path(directory)) + del args.warmup_manifest + + def handler(request): + body = json.loads(request.content) + body.pop("model") + case = next(c for c in CASES if c["request"] == body) + return httpx.Response(200, json=response_for(case)) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + self.assertEqual(await bench.run(args, CASES), 0) + + async def test_warmup_failures_preserve_all_measured_unattempted(self): + for kind in ("missing-type", "wrong-type", "nonobject", "http", "timeout", "cancel"): + with self.subTest(kind=kind), tempfile.TemporaryDirectory() as directory: + args = run_args(Path(directory)) + raw = response_for(CASES[0]) + if kind == "missing-type": + raw["answers"]["route"].pop("type") + elif kind == "wrong-type": + raw["answers"]["route"]["type"] = "noul" + elif kind == "nonobject": + raw["answers"]["route"] = [] + + def handler(request): + if kind == "timeout": + raise httpx.ReadTimeout("controlled warmup timeout", request=request) + if kind == "cancel": + raise asyncio.CancelledError + return httpx.Response(503, text="busy") if kind == "http" else httpx.Response(200, json=raw) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with patch.object(bench.httpx, "AsyncClient", return_value=client): + with self.assertRaisesRegex(ValueError, "readiness/warmup failed"): + await bench.run(args, CASES) + self.assertFalse((args.output / "responses.jsonl").exists()) + saved = json.loads((args.output / "summary.json").read_text()) + self.assertEqual(saved, bench.summarize_saved(args.output)) + self.assertEqual(saved["planned_requests"], 4) + self.assertEqual(saved["unattempted_requests"], 4) + self.assertEqual(saved["planned_decisions"], 6) + self.assertEqual(saved["unattempted_decisions"], 6) + for field in ("attempted_requests", "incomplete_requests", "successful_requests", + "failed_requests", "attempted_decisions", "incomplete_decisions"): + self.assertEqual(saved[field], 0) + for field in ("wall_seconds", "requests_per_second", "decisions_per_second"): + self.assertIsNone(saved[field]) + state = json.loads((args.output / "completion.json").read_text()) + self.assertIsNone(state["started_at"]) + self.assertIsNone(state["finished_at"]) + self.assertEqual(state["attempted_ids"], []) + self.assertEqual(state["stop_reason"], "warmup_interrupted" if kind == "cancel" else "warmup_failed") + self.assertFalse(saved["evidence_complete"]) + async def test_concurrency_order_and_question_counts(self): active, peak = 0, 0 @@ -125,6 +217,7 @@ def handler(request): phase="feasibility", concurrency=2, warmup=2, + warmup_manifest=None, timeout=1, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) @@ -136,6 +229,7 @@ def handler(request): self.assertEqual(len(calls), 6) summary = json.loads((args.output / "summary.json").read_text()) self.assertEqual(summary["requests"], 4) + self.assertEqual(bench.summarize_saved(args.output), summary) self.assertEqual( len(json.loads((args.output / "warmup.json").read_text())), 2 ) @@ -146,6 +240,115 @@ def handler(request): with self.assertRaises(FileExistsError): await bench.run(args, CASES) + async def test_independent_warmup_does_not_enter_measurement(self): + warmup_case = json.loads(json.dumps(CASES[0])) + warmup_case["id"] = "warmup-only" + warmup_case["request"]["state"] = "independent warmup" + calls = [] + + def handler(request): + body = json.loads(request.content) + body.pop("model") + calls.append(body) + case = ( + warmup_case if body == warmup_case["request"] else next( + c for c in CASES if c["request"] == body + ) + ) + return httpx.Response(200, json=response_for(case)) + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + metadata = { + key: "mock-only" + for key in ( + "gpu", "gpu_ids", "driver", "cuda", "precision", + "model_revision", "runtime_revision", "cache_policy", + "cuda_evidence", "reservation", + ) + } + bench.write_json(root / "metadata.json", metadata) + warmup_manifest = root / "warmup.jsonl" + warmup_manifest.write_text(json.dumps(warmup_case) + "\n") + args = argparse.Namespace( + metadata=root / "metadata.json", + output=root / "run", + manifest=Path(bench.__file__).with_name("smoke.jsonl"), + endpoint="http://localhost/v1/systemone", + model="english", + phase="measured", + concurrency=1, + warmup=2, + timeout=1, + warmup_manifest=warmup_manifest, + ) + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + self.assertEqual(await bench.run(args, CASES), 0) + self.assertEqual(calls[:2], [warmup_case["request"]] * 2) + records = [ + json.loads(line) + for line in (args.output / "responses.jsonl").read_text().splitlines() + ] + self.assertEqual([r["id"] for r in records], [c["id"] for c in CASES]) + self.assertEqual( + (args.output / "warmup-requests.jsonl").read_bytes(), + warmup_manifest.read_bytes(), + ) + self.assertEqual(bench.summarize_saved(args.output)["planned_requests"], 4) + (args.output / "warmup-requests.jsonl").write_text("changed") + with self.assertRaisesRegex(ValueError, "warmup manifest checksum"): + bench.summarize_saved(args.output) + + async def test_body_and_validation_are_inside_latency_clock(self): + clock = [10.0] + body = json.dumps(response_for(CASES[0])).encode() + + class Body(httpx.AsyncByteStream): + async def __aiter__(self): + clock[0] += 1 + yield body + + def validate(case, payload): + clock[0] += 2 + return original(case, payload) + + original = bench.read_answers + async with httpx.AsyncClient( + transport=httpx.MockTransport( + lambda request: httpx.Response(200, stream=Body()) + ) + ) as client: + with ( + patch.object(bench.time, "perf_counter", side_effect=lambda: clock[0]), + patch.object(bench, "read_answers", side_effect=validate), + ): + record = await bench.request_one( + client, "http://localhost/v1/systemone", CASES[0], "english", 1 + ) + self.assertEqual(record["latency_ms"], 3000) + self.assertIsNone(record["error"]) + + async def test_cancel_after_body_retains_raw_without_completed_latency(self): + async with httpx.AsyncClient( + transport=httpx.MockTransport( + lambda request: httpx.Response(200, json=response_for(CASES[0])) + ) + ) as client: + with patch.object( + bench, "read_answers", side_effect=asyncio.CancelledError + ): + record = await bench.request_one( + client, "http://localhost/v1/systemone", CASES[0], "english", 1 + ) + self.assertEqual(record["state"], "incomplete") + self.assertEqual(json.loads(record["response"]), response_for(CASES[0])) + self.assertIsNone(record["latency_ms"]) + self.assertEqual(record["answers"], {}) + async def test_total_deadline(self): async def handler(request): await asyncio.sleep(1) @@ -157,6 +360,285 @@ async def handler(request): ) self.assertEqual(record["error_kind"], "timeout") + async def test_transport_failure_with_empty_error_is_excluded_from_quality(self): + def handler(request): + raise httpx.ConnectError("", request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + record = await bench.request_one( + client, "http://localhost/v1/systemone", CASES[0], "english", 1 + ) + self.assertEqual(record["error_kind"], "transport") + summary = bench.summarize(CASES[:1], [record], 1) + self.assertEqual(summary["failed_requests"], 1) + self.assertEqual( + summary["quality_on_successful_requests"]["choice_accuracy"]["count"], 0 + ) + + +class PersistenceTests(unittest.IsolatedAsyncioTestCase): + async def test_bom_and_invalid_utf8_online_offline_text_decoding_agree(self): + valid = json.dumps(response_for(CASES[0])).encode("utf-8") + for label, body, error_kind in ( + ("bom", b"\xef\xbb\xbf" + valid, "invalid_response"), + ("invalid-utf8", valid[:-1] + b', "unused": "\xff"}', None), + ): + with self.subTest(label=label), tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + saved_inputs(path, CASES[:1]) + expected_text = httpx.Response(200, content=body).text + async with httpx.AsyncClient(transport=httpx.MockTransport( + lambda request: httpx.Response(200, content=body) + )) as client: + records, elapsed = await bench.replay( + client, "http://localhost/v1/systemone", CASES[:1], "english", 1, 1, path + ) + self.assertEqual(records[0]["response"], expected_text) + self.assertEqual(records[0]["error_kind"], error_kind) + self.assertTrue(expected_text.startswith("\ufeff") if label == "bom" else "\ufffd" in expected_text) + expected = bench.summarize(CASES[:1], records, elapsed) + expected["stop_reason"] = "completed" + self.assertEqual(bench.summarize_saved(path), expected) + + async def collect(self, path): + saved_inputs(path) + + def handler(request): + body = json.loads(request.content) + body.pop("model") + case = next(c for c in CASES if c["request"] == body) + return httpx.Response(200, json=response_for(case)) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + return await bench.replay( + client, "http://localhost/v1/systemone", CASES, "english", 2, 1, path + ) + + def replace_records(self, path, records): + (path / "responses.jsonl").write_text( + "".join(json.dumps(r) + "\n" for r in records) + ) + state = json.loads((path / "completion.json").read_text()) + state["responses_sha256"] = bench.digest(path / "responses.jsonl") + bench.write_json(path / "completion.json", state) + + async def test_saved_http_timeout_and_typed_validation_failures(self): + calls = 0 + missing_type = response_for(CASES[1]) + missing_type["answers"]["refund"].pop("type") + + def handler(request): + nonlocal calls + calls += 1 + if calls == 1: + return httpx.Response(503, text="busy") + if calls == 2: + return httpx.Response(200, json=missing_type) + if calls == 3: + raise httpx.ReadTimeout("controlled timeout", request=request) + return httpx.Response(200, json=response_for(CASES[3])) + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + saved_inputs(path) + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler) + ) as client: + records, elapsed = await bench.replay( + client, "http://localhost/v1/systemone", CASES, "english", 1, 1, path + ) + summary = bench.summarize_saved(path) + self.assertEqual( + summary["errors"], + {"http_503": 1, "invalid_response": 1, "timeout": 1}, + ) + self.assertEqual(summary["completed_requests"], 4) + self.assertEqual(summary["failed_requests"], 3) + self.assertEqual(summary["failed_decisions"], 3) + self.assertIsNotNone(summary["failed_latency_p50_ms"]) + self.assertEqual(json.loads(records[1]["response"]), missing_type) + self.assertEqual(records[1]["answers"], {}) + expected = bench.summarize(CASES, records, elapsed) + expected["stop_reason"] = "completed" + self.assertEqual(summary, expected) + + async def test_out_of_order_records_recompute_by_frozen_id(self): + release = asyncio.Event() + later_completed = asyncio.Event() + + async def handler(request): + body = json.loads(request.content) + body.pop("model") + case = next(c for c in CASES if c["request"] == body) + if case["id"] == CASES[0]["id"]: + await release.wait() + else: + later_completed.set() + return httpx.Response(200, json=response_for(case)) + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + saved_inputs(path) + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler) + ) as client: + task = asyncio.create_task(bench.replay( + client, "http://localhost/v1/systemone", CASES, "english", 2, 1, path + )) + await later_completed.wait() + partial = [json.loads(line) for line in (path / "responses.jsonl").read_text().splitlines()] + self.assertTrue(partial) + self.assertNotIn(CASES[0]["id"], [r["id"] for r in partial]) + partial_summary = bench.summarize_saved(path) + self.assertEqual(partial_summary["successful_requests"], len(partial)) + self.assertEqual(partial_summary["incomplete_requests"], 1) + self.assertIsNone(partial_summary["wall_seconds"]) + release.set() + records, elapsed = await task + saved = [ + json.loads(line) + for line in (path / "responses.jsonl").read_text().splitlines() + ] + self.assertEqual([r["id"] for r in saved], [c["id"] for c in CASES]) + self.assertEqual([r["id"] for r in records], [c["id"] for c in CASES]) + expected = bench.summarize(CASES, records, elapsed) + expected["stop_reason"] = "completed" + self.assertEqual(bench.summarize_saved(path), expected) + args = argparse.Namespace( + reference=path, + candidate=path, + max_probability_drift=0, + max_score_drift=0, + max_flips=0, + ) + with redirect_stdout(io.StringIO()): + self.assertEqual(bench.compare(args), 0) + + async def test_cancel_saves_completed_incomplete_and_unattempted(self): + slow_started = asyncio.Event() + calls = 0 + + async def handler(request): + nonlocal calls + calls += 1 + if calls == 1: + return httpx.Response(200, json=response_for(CASES[0])) + slow_started.set() + await asyncio.Event().wait() + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + saved_inputs(path) + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler) + ) as client: + task = asyncio.create_task(bench.replay( + client, "http://localhost/v1/systemone", CASES, "english", 1, 60, path + )) + await slow_started.wait() + before = [ + json.loads(line) + for line in (path / "responses.jsonl").read_text().splitlines() + ] + self.assertEqual([r["id"] for r in before], [CASES[0]["id"]]) + state = json.loads((path / "completion.json").read_text()) + self.assertEqual(state["attempted_ids"], [c["id"] for c in CASES[:2]]) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + summary = bench.summarize_saved(path) + for name, count in ( + ("planned_requests", 4), ("attempted_requests", 2), + ("successful_requests", 1), ("failed_requests", 0), + ("incomplete_requests", 1), ("unattempted_requests", 2), + ): + self.assertEqual(summary[name], count) + self.assertEqual(summary["planned_decisions"], 6) + self.assertEqual(summary["attempted_decisions"], 2) + self.assertEqual(summary["unattempted_decisions"], 4) + self.assertEqual(summary["incomplete_decisions"], 1) + self.assertEqual(summary["stop_reason"], "interrupted") + self.assertFalse(summary["evidence_complete"]) + self.assertIsNotNone(summary["wall_seconds"]) + records = [ + json.loads(line) + for line in (path / "responses.jsonl").read_text().splitlines() + ] + self.assertEqual(records[1]["error_kind"], "interrupted") + self.assertIsNone(records[1]["latency_ms"]) + + async def test_unfinished_measurement_never_invents_wall_or_throughput(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + await self.collect(path) + records = [ + json.loads(line) + for line in (path / "responses.jsonl").read_text().splitlines() + ] + self.replace_records(path, records[:-1]) + state = json.loads((path / "completion.json").read_text()) + state.update( + finished_at=None, wall_seconds=None, stop_reason=None, + completed_ids=[r["id"] for r in records[:-1]], + active_ids=[records[-1]["id"]], + ) + bench.write_json(path / "completion.json", state) + summary = bench.summarize_saved(path) + self.assertEqual(summary["incomplete_requests"], 1) + self.assertIsNone(summary["wall_seconds"]) + self.assertIsNone(summary["requests_per_second"]) + self.assertIsNone(summary["decisions_per_second"]) + self.assertFalse(summary["evidence_complete"]) + + async def test_id_coverage_duplicate_and_unknown_rejected(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + await self.collect(path) + original = [ + json.loads(line) + for line in (path / "responses.jsonl").read_text().splitlines() + ] + for records in ( + original[:-1], original + [original[0]], + [{**original[0], "id": "unknown"}] + original[1:], + ): + self.replace_records(path, records) + with self.assertRaises(ValueError): + bench.summarize_saved(path) + + async def test_checksums_and_raw_normalized_consistency_rejected(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + await self.collect(path) + files = { + name: (path / name).read_bytes() + for name in ( + "requests.jsonl", "config.json", "responses.jsonl", "completion.json" + ) + } + for name in ("requests.jsonl", "config.json", "responses.jsonl"): + (path / name).write_bytes(files[name] + b"\n") + with self.assertRaisesRegex(ValueError, "checksum"): + bench.summarize_saved(path) + (path / name).write_bytes(files[name]) + for change in ("raw", "normalized", "validation"): + records = [ + json.loads(line) + for line in files["responses.jsonl"].decode().splitlines() + ] + if change == "raw": + records[0]["response"] += " " + elif change == "normalized": + qid = next(iter(records[0]["answers"])) + records[0]["answers"][qid]["value"] = "changed" + else: + records[0]["error_kind"] = "invalid_response" + records[0]["error"] = "changed" + self.replace_records(path, records) + with self.assertRaises(ValueError): + bench.summarize_saved(path) + (path / "completion.json").write_bytes(files["completion.json"]) + class ValidationTests(unittest.TestCase): def test_all_primitives_and_invalid_distributions(self): @@ -192,6 +674,7 @@ def test_rounded_probabilities_at_the_sum_boundary(self): {"A": 0.1130, "B": 0.5135, "C": 0.3151, "D": 0.0583}, ): answer = { + "type": "choice", "choice": max(probabilities, key=probabilities.get), "probabilities": probabilities, } @@ -199,9 +682,26 @@ def test_rounded_probabilities_at_the_sum_boundary(self): off = {"A": 0.4, "B": 0.3, "C": 0.2, "D": 0.09} with self.assertRaises(ValueError): bench.read_answers( - case, {"answers": {"q": {"choice": "A", "probabilities": off}}} + case, + {"answers": {"q": { + "type": "choice", "choice": "A", "probabilities": off + }}}, ) + def test_missing_wrong_and_nonobject_answer_types(self): + for case in CASES: + for qid in case["request"]["questions"]: + for mutation in (None, "wrong", []): + payload = response_for(case) + if mutation is None: + payload["answers"][qid].pop("type") + elif isinstance(mutation, str): + payload["answers"][qid]["type"] = mutation + else: + payload["answers"][qid] = mutation + with self.assertRaises(ValueError): + bench.read_answers(case, payload) + def test_duplicate_ids(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "requests.jsonl" From 7ac2b11b5707d392d6d8880891422f2e1d64fe7c Mon Sep 17 00:00:00 2001 From: linear3735 Date: Sat, 10 Oct 2026 22:47:33 +0800 Subject: [PATCH 2/7] bench: verify warmup evidence and keep invalid responses isolated Document rejection of torn snapshots and add the latency leaderboard. Preserve historical runs for replay with their original collector. Changed lines (additions plus deletions): 16 source, 72 tests, 95 documentation. --- benchmarks/README.md | 19 ++++++-- benchmarks/bench.py | 16 +++++-- benchmarks/latency-leaderboard.md | 76 +++++++++++++++++++++++++++++++ tests/benchmarks/test_bench.py | 72 ++++++++++++++++++++++++++++- 4 files changed, 171 insertions(+), 12 deletions(-) create mode 100644 benchmarks/latency-leaderboard.md diff --git a/benchmarks/README.md b/benchmarks/README.md index 2200d6ec..1cc84a2f 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -5,6 +5,10 @@ First harness for #39: replay the same labelled requests against CUDA-backed required (`python -m pip install -r benchmarks/requirements.txt`). Reuse an existing benchmark environment when available. +The [latency leaderboard](latency-leaderboard.md) records the five-model H800 +comparison and its timing rules. Its historical body-receipt clock is kept +separate from this runner's body/JSON/schema-validation clock. + The runner is an HTTP client: it neither launches models nor reserves or accesses a GPU. Start each server separately through the host's verified GPU scheduler, using the same exact reserved device and pinned Laya checkpoint. Verify CUDA @@ -167,21 +171,26 @@ python benchmarks/bench.py summary /tmp/laya-c1-run1 This command makes no requests and prints the recomputed summary without changing saved files. It checks frozen IDs, duplicates and unknown entries, saved manifest, -config and response checksums, and agreement between original response text and +config, warmup and response checksums, and agreement between original response text and normalized answers/validation. It requires the new `completion.json` evidence; older runs remain usable by `compare` but cannot acquire unrecorded timing or interruption boundaries through this command. +Runs recorded before `warmup_sha256` was added must use their original collector +for offline summaries. Do not add a checksum to historical evidence to make it +look verified at collection time. After a hard kill, a valid start snapshot can distinguish unattempted IDs from attempted IDs without completed records. Without a saved terminal boundary, `wall_seconds` and both throughput fields are null and `evidence_complete` is -false. Partial files, checksum disagreement or missing claimed completed records -are rejected; inspect the retained originals instead of repairing timing or +false. A response flush and its completion snapshot are separate writes. A kill +between them can leave an extra response with a stale checksum or completed-ID +list. Such runs, partial files and missing claimed records are rejected; inspect +the retained originals instead of repairing timing or declaring a complete baseline. Hashes detect inconsistency, not authenticity against someone rewriting all evidence and hashes. Interrupted runs provide partial observations with explicit coverage, not complete traversals. -Flushing records and atomically replacing state cover process interruption/kill -for the successfully written records. They do not promise machine power-loss +Flushing records and atomically replacing state preserve consistent snapshots +when both writes finish. They do not promise machine power-loss durability; this tool does not call `fsync`. ## Compare output fidelity diff --git a/benchmarks/bench.py b/benchmarks/bench.py index 4ab8403a..1eb4ba71 100644 --- a/benchmarks/bench.py +++ b/benchmarks/bench.py @@ -74,11 +74,12 @@ def load_cases(path): def number(value): - return ( - isinstance(value, (int, float)) - and not isinstance(value, bool) - and math.isfinite(value) - ) + if not isinstance(value, (int, float)) or isinstance(value, bool): + return False + try: + return math.isfinite(value) + except OverflowError: + return False WIRE_DECIMALS = 4 # workers round probabilities on the wire to four decimal places @@ -289,6 +290,7 @@ def measurement_state(output): if output is not None: state["manifest_sha256"] = digest(output / "requests.jsonl") state["config_sha256"] = digest(output / "config.json") + state["warmup_sha256"] = digest(output / "warmup.json") return state @@ -468,6 +470,10 @@ def summarize_saved(path): state = json.loads((path / "completion.json").read_text()) if digest(path / "config.json") != state["config_sha256"]: raise ValueError("saved config checksum does not match completion") + if "warmup_sha256" not in state: + raise ValueError("saved warmup checksum missing; use the collector that produced this run") + if not (path / "warmup.json").is_file() or digest(path / "warmup.json") != state["warmup_sha256"]: + raise ValueError("saved warmup checksum does not match completion") if config["latency_basis"] != "post_to_body_json_and_schema_validation": raise ValueError("unsupported saved latency basis") manifest_hash = digest(path / "requests.jsonl") diff --git a/benchmarks/latency-leaderboard.md b/benchmarks/latency-leaderboard.md new file mode 100644 index 00000000..134dbbf6 --- /dev/null +++ b/benchmarks/latency-leaderboard.md @@ -0,0 +1,76 @@ +# Public231 request latency leaderboard + +This board ranks deployed model/service configurations on the same 231 public +text tasks. It measures complete HTTP requests at concurrency 1 on one H800; +it does not isolate kernel speed, model size or same-model acceleration. + +## H800 group: Slurm 415076, 2026-10-07 UTC + +Rank uses pooled p50, ascending. Equal full-precision values share a rank. p95 has +its own ordering: a low median does not imply low tail latency. + +| p50 rank | Deployment | p50 ms | p95 ms | p95 rank | Successful / planned | +|---:|---|---:|---:|---:|---:| +| 1 | Open-Jev-9B, Omni native | 42.688 | 327.057 | 4 | 693 / 693 | +| 2 | Open-Jev-27B-v1.1, Omni native | 68.877 | 1068.817 | 5 | 693 / 693 | +| 3 | Intern-Decision-2B, official HF | 79.054 | 175.276 | 2 | 693 / 693 | +| 4 | Intern-Decision-0.8B, official HF | 80.493 | 175.237 | 1 | 693 / 693 | +| 5 | Intern-Decision-4B, official HF | 104.552 | 235.291 | 3 | 693 / 693 | + +There were no measured failures, invalid responses or missing requests. Each +deployment had five client warmups once, followed by three 231-request traversals. +Warmups and process startup are excluded. Repetition gives 693 timing samples +per deployment, with 231 unique tasks. + +## Clock and controls + +The historical client starts immediately before POST and stops after HTTPX reads +the complete body and assigns `response.text`. JSON decoding and schema validation +occur after the timer stops. Quantiles use sorted samples, k=(n−1)q and linear +interpolation; pooled p95 is recomputed from all samples, not averaged from round +p95 values. The [immutable report](https://github.com/linear3735/system1-agents/blob/40bf742e2b84ad6157bf2990dd5e4bab73571702/benchmarks/results/public231-h800.md) +includes per-round measurements and exact deployment revisions. + +Inputs and scorer are fixed to [JevBench 7ce310c7](https://github.com/fstandhartinger/jevbench/tree/7ce310c7262ed49cc85853339a8a42459298e3f3), +canonical dataset SHA-256 (`dataset_hash` of the pinned tasks, recorded as +`canonical_dataset_sha256` in the numeric attachment's source manifest): +`dc3995d8ae1e2fc8e81ce38431add509eb8bb39b85aadfd0c7c32079382dde51`. +All arms used BF16, concurrency 1, thinking off, no retries and the same request +order. Intern used upstream HF at T=1 with Torch fallback for unavailable optional +fast paths. Open-Jev used native Rust/CUDA with checkpoint temperatures, merged +LoRA/head and graph/prefix cache off; 9B was sequential and 27B packed candidates. +Different architectures, templates and context policies limit what ratios mean. + +The published report is available now. A numeric audit attachment is prepared but +not yet published; it contains all 3465 measured timings and 25 warmups, probability +vectors and source hashes without question/option text. Raw responses remain +outside the repository. Accuracy is maintained separately in +[System1-Agents](https://github.com/ThinkFlowLab/system1-agents/pull/69). + +## New runner group: Slurm 416135 + +The [serving-evidence PR](https://github.com/ThinkFlowLab/system1-omni/pull/131) +reports a separate two-model run: 9B pooled p50/p95 43.082/326.968 ms and +27B 69.459/1066.037 ms. That client includes body decoding, JSON and schema +validation and uses nearest-rank p95. It performs five warmups per traversal. +Those values do not enter the historical five-model ranking above. No speedup +ratio is derived between these clocks or campaigns. + +## Add a result + +Use the existing [runner and offline summary](README.md#run-against-a-ready-gpu-server), +with a pinned manifest, actual deployment metadata and fresh external output +directories. Predeclare hardware, precision, concurrency, request order, cache +state, timeouts, warmups, repeats, clock boundaries and quantile method. + +Publish per-round and pooled p50/p95 with full planned/attempted/successful/failed/ +incomplete/unattempted counts. Keep failure latencies separate. In this board, a +run with failures or incomplete coverage remains visible but unranked; a fast +success subset cannot establish a winner. Startup and throughput use their own +boundaries, and ticket episode duration is not request latency. + +Append a dated campaign and auditable evidence without replacing prior results. +Compare new deployments with a reference measured in the same campaign. Different +hardware, workloads, precision, concurrency, clocks or quantile methods need +separate groups. Image replacement, repeated-image/cache workloads and higher +concurrency have not been measured here; add their own frozen inputs before ranking. diff --git a/tests/benchmarks/test_bench.py b/tests/benchmarks/test_bench.py index 508189e8..69b7b947 100644 --- a/tests/benchmarks/test_bench.py +++ b/tests/benchmarks/test_bench.py @@ -38,6 +38,7 @@ def response_for(case): def saved_inputs(path, cases=CASES): path.mkdir() + bench.write_json(path / "warmup.json", []) (path / "requests.jsonl").write_text("".join(json.dumps(c) + "\n" for c in cases)) bench.write_json( path / "config.json", @@ -122,6 +123,7 @@ def handler(request): for field in ("wall_seconds", "requests_per_second", "decisions_per_second"): self.assertIsNone(saved[field]) state = json.loads((args.output / "completion.json").read_text()) + self.assertEqual(state["warmup_sha256"], bench.digest(args.output / "warmup.json")) self.assertIsNone(state["started_at"]) self.assertIsNone(state["finished_at"]) self.assertEqual(state["attempted_ids"], []) @@ -230,6 +232,8 @@ def handler(request): summary = json.loads((args.output / "summary.json").read_text()) self.assertEqual(summary["requests"], 4) self.assertEqual(bench.summarize_saved(args.output), summary) + state = json.loads((args.output / "completion.json").read_text()) + self.assertEqual(state["warmup_sha256"], bench.digest(args.output / "warmup.json")) self.assertEqual( len(json.loads((args.output / "warmup.json").read_text())), 2 ) @@ -462,6 +466,35 @@ def handler(request): expected["stop_reason"] = "completed" self.assertEqual(summary, expected) + async def test_large_integer_responses_are_saved_failures(self): + choice, noul = response_for(CASES[0]), response_for(CASES[1]) + probabilities = choice["answers"]["route"]["probabilities"] + probabilities[next(iter(probabilities))] = 10**400 + noul["answers"]["refund"]["noul"] = 10**400 + payloads = [choice, noul, *(response_for(case) for case in CASES[2:])] + replies = iter(httpx.Response(200, json=payload) for payload in payloads) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + saved_inputs(path) + async with httpx.AsyncClient( + transport=httpx.MockTransport(lambda _: next(replies)) + ) as client: + records, elapsed = await bench.replay( + client, "http://localhost/v1/systemone", CASES, "english", 1, 1, path + ) + summary = bench.summarize_saved(path) + self.assertEqual(summary["errors"], {"invalid_response": 2}) + self.assertEqual(summary["completed_requests"], 4) + self.assertEqual(summary["successful_requests"], 2) + self.assertTrue(summary["evidence_complete"]) + for record, payload in zip(records[:2], payloads[:2]): + self.assertEqual(json.loads(record["response"]), payload) + self.assertEqual(record["response_sha256"], bench.response_digest(record["response"])) + self.assertEqual(record["answers"], {}) + expected = bench.summarize(CASES, records, elapsed) + expected["stop_reason"] = "completed" + self.assertEqual(summary, expected) + async def test_out_of_order_records_recompute_by_frozen_id(self): release = asyncio.Event() later_completed = asyncio.Event() @@ -590,6 +623,41 @@ async def test_unfinished_measurement_never_invents_wall_or_throughput(self): self.assertIsNone(summary["decisions_per_second"]) self.assertFalse(summary["evidence_complete"]) + async def test_flushed_response_with_stale_completion_is_rejected(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + await self.collect(path) + records = (path / "responses.jsonl").read_bytes().splitlines(keepends=True) + state = json.loads((path / "completion.json").read_text()) + state.update( + finished_at=None, wall_seconds=None, stop_reason=None, + completed_ids=state["completed_ids"][:-1], + active_ids=[CASES[-1]["id"]], + responses_sha256=bench.hashlib.sha256(b"".join(records[:-1])).hexdigest(), + ) + bench.write_json(path / "completion.json", state) + with self.assertRaisesRegex(ValueError, "saved responses checksum"): + bench.summarize_saved(path) + state["responses_sha256"] = bench.digest(path / "responses.jsonl") + bench.write_json(path / "completion.json", state) + with self.assertRaisesRegex(ValueError, "completed IDs"): + bench.summarize_saved(path) + + async def test_missing_warmup_evidence_is_rejected(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + await self.collect(path) + state = json.loads((path / "completion.json").read_text()) + state.pop("warmup_sha256") + bench.write_json(path / "completion.json", state) + with self.assertRaisesRegex(ValueError, "saved warmup checksum missing"): + bench.summarize_saved(path) + state["warmup_sha256"] = bench.digest(path / "warmup.json") + bench.write_json(path / "completion.json", state) + (path / "warmup.json").unlink() + with self.assertRaisesRegex(ValueError, "saved warmup checksum"): + bench.summarize_saved(path) + async def test_id_coverage_duplicate_and_unknown_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "run" @@ -613,10 +681,10 @@ async def test_checksums_and_raw_normalized_consistency_rejected(self): files = { name: (path / name).read_bytes() for name in ( - "requests.jsonl", "config.json", "responses.jsonl", "completion.json" + "requests.jsonl", "config.json", "responses.jsonl", "completion.json", "warmup.json" ) } - for name in ("requests.jsonl", "config.json", "responses.jsonl"): + for name in ("requests.jsonl", "config.json", "responses.jsonl", "warmup.json"): (path / name).write_bytes(files[name] + b"\n") with self.assertRaisesRegex(ValueError, "checksum"): bench.summarize_saved(path) From 0a55fce06787bc69d735f13853d93212e842b017 Mon Sep 17 00:00:00 2001 From: linear3735 Date: Mon, 12 Oct 2026 10:29:53 +0800 Subject: [PATCH 3/7] docs: clarify pooled latency and dataset hash provenance --- benchmarks/README.md | 10 ++++++++++ benchmarks/latency-leaderboard.md | 10 ++++++++++ 2 files changed, 20 insertions(+) diff --git a/benchmarks/README.md b/benchmarks/README.md index 1cc84a2f..7b8ba930 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -179,6 +179,16 @@ Runs recorded before `warmup_sha256` was added must use their original collector for offline summaries. Do not add a checksum to historical evidence to make it look verified at collection time. +For pooled percentiles, first validate each of the three measured rounds with its +original collector's `summary`; pool only rounds with the same deployment, +manifest and timing/collection settings. Concatenate `latency_ms` from their +`responses.jsonl` records where `state` is `completed` and `error` is `null`, +keeping every repetition and excluding warmups and unsuccessful records. For +sorted samples `x` and `n = len(x)`, p50 is `statistics.median(x)` and p95 is +`x[math.ceil(0.95 * n) - 1]` (zero-based nearest rank; both null if empty). +Recompute from these samples, never average round percentiles; report per-round +and pooled request counts alongside them. + After a hard kill, a valid start snapshot can distinguish unattempted IDs from attempted IDs without completed records. Without a saved terminal boundary, `wall_seconds` and both throughput fields are null and `evidence_complete` is diff --git a/benchmarks/latency-leaderboard.md b/benchmarks/latency-leaderboard.md index 134dbbf6..30c4e89a 100644 --- a/benchmarks/latency-leaderboard.md +++ b/benchmarks/latency-leaderboard.md @@ -35,6 +35,16 @@ Inputs and scorer are fixed to [JevBench 7ce310c7](https://github.com/fstandhart canonical dataset SHA-256 (`dataset_hash` of the pinned tasks, recorded as `canonical_dataset_sha256` in the numeric attachment's source manifest): `dc3995d8ae1e2fc8e81ce38431add509eb8bb39b85aadfd0c7c32079382dde51`. + +The input SHA-256 +`00d1f10ca1142499a598b5f1c6602d329cce1cef04c961c46fc0318daa2c67d3` +in [issue #130](https://github.com/ThinkFlowLab/system1-omni/issues/130) hashes the +exported public231 JSONL request-manifest bytes for the separate two-model run +(Slurm 416135). The `dataset_hash` above hashes sorted canonical JevBench task +records for this five-model campaign (Slurm 415076). Both reference JevBench +`7ce310c7`, but hash different representations and are not interchangeable file +checksums. + All arms used BF16, concurrency 1, thinking off, no retries and the same request order. Intern used upstream HF at T=1 with Torch fallback for unavailable optional fast paths. Open-Jev used native Rust/CUDA with checkpoint temperatures, merged From 36c244b6bc8715af8a1958f5d55a6acd35c98edd Mon Sep 17 00:00:00 2001 From: linear3735 Date: Tue, 13 Oct 2026 14:20:03 +0800 Subject: [PATCH 4/7] bench: fix two response boundary cases --- benchmarks/bench.py | 6 +-- tests/benchmarks/test_bench.py | 86 ++++++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 3 deletions(-) diff --git a/benchmarks/bench.py b/benchmarks/bench.py index 1eb4ba71..3973ce58 100644 --- a/benchmarks/bench.py +++ b/benchmarks/bench.py @@ -260,7 +260,7 @@ async def request_one(client, endpoint, case, model, timeout): ) except httpx.RequestError as exc: record.update(error=str(exc), error_kind="transport") - except (ValueError, KeyError, TypeError, AttributeError) as exc: + except (ValueError, KeyError, TypeError, AttributeError, RecursionError) as exc: record.update(error=str(exc), error_kind="invalid_response") except asyncio.CancelledError: record.update( @@ -548,7 +548,7 @@ def summarize_saved(path): elif ( not isinstance(status, int) or isinstance(status, bool) - or not 100 <= status <= 599 + or not 100 <= status <= 999 ): raise ValueError("invalid HTTP status") elif not 200 <= status < 300: @@ -558,7 +558,7 @@ def summarize_saved(path): answers = read_answers( by_id[record["id"]], json.loads(record["response"]) ) - except (ValueError, KeyError, TypeError, AttributeError): + except (ValueError, KeyError, TypeError, AttributeError, RecursionError): error_kind = "invalid_response" if ( record["answers"] != answers diff --git a/tests/benchmarks/test_bench.py b/tests/benchmarks/test_bench.py index 69b7b947..8f8b9bde 100644 --- a/tests/benchmarks/test_bench.py +++ b/tests/benchmarks/test_bench.py @@ -495,6 +495,92 @@ async def test_large_integer_responses_are_saved_failures(self): expected["stop_reason"] = "completed" self.assertEqual(summary, expected) + async def test_run_saves_deep_json_and_nonstandard_http_failures(self): + nested = "[" * 10000 + "0" + "]" * 10000 + with self.assertRaises(RecursionError): + json.loads(nested) + for status, body, error_kind in ( + (200, nested, "invalid_response"), + (600, "busy", "http_600"), + (999, "busy", "http_999"), + ): + with self.subTest(status=status), tempfile.TemporaryDirectory() as directory: + args = run_args(Path(directory)) + replies = iter([ + httpx.Response(200, json=response_for(CASES[0])), + httpx.Response(status, text=body), + *(httpx.Response(200, json=response_for(case)) for case in CASES[1:]), + ]) + calls = [] + + def handler(request): + payload = json.loads(request.content) + payload.pop("model") + calls.append(payload) + return next(replies) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + self.assertEqual(await bench.run(args, CASES), 1) + self.assertEqual(calls, [CASES[0]["request"], *(c["request"] for c in CASES)]) + warmup = json.loads((args.output / "warmup.json").read_text()) + self.assertEqual(len(warmup), 1) + self.assertIsNone(warmup[0]["error"]) + records = [ + json.loads(line) + for line in (args.output / "responses.jsonl").read_text().splitlines() + ] + self.assertEqual([r["id"] for r in records], [c["id"] for c in CASES]) + failed = records[0] + self.assertEqual(failed["status"], status) + self.assertEqual(failed["response"], body) + self.assertEqual(failed["response_sha256"], bench.response_digest(body)) + self.assertEqual(failed["error_kind"], error_kind) + self.assertTrue(failed["error"]) + self.assertEqual(failed["answers"], {}) + self.assertEqual(failed["state"], "completed") + self.assertIsNotNone(failed["latency_ms"]) + self.assertTrue(all(r["error"] is None for r in records[1:])) + state = json.loads((args.output / "completion.json").read_text()) + self.assertEqual(state["active_ids"], []) + self.assertEqual(state["attempted_ids"], [c["id"] for c in CASES]) + self.assertEqual(state["completed_ids"], state["attempted_ids"]) + self.assertEqual(state["responses_sha256"], bench.digest(args.output / "responses.jsonl")) + summary = json.loads((args.output / "summary.json").read_text()) + for field, count in ( + ("requests", 4), ("planned_requests", 4), ("attempted_requests", 4), + ("completed_requests", 4), ("successful_requests", 3), ("failed_requests", 1), + ("unattempted_requests", 0), ("incomplete_requests", 0), + ("planned_decisions", 6), ("attempted_decisions", 6), + ("completed_decisions", 6), ("successful_decisions", 5), ("failed_decisions", 1), + ("unattempted_decisions", 0), ("incomplete_decisions", 0), + ): + self.assertEqual(summary[field], count) + self.assertEqual(summary["errors"], {error_kind: 1}) + self.assertEqual(summary["stop_reason"], "completed") + self.assertTrue(summary["evidence_complete"]) + self.assertEqual(summary, bench.summarize_saved(args.output)) + output = io.StringIO() + with ( + patch("sys.argv", ["bench.py", "summary", str(args.output)]), + redirect_stdout(output), + ): + self.assertEqual(bench.main(), 0) + self.assertEqual(json.loads(output.getvalue()), summary) + + async def test_saved_invalid_http_status_is_rejected(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "run" + records, _ = await self.collect(path) + for status in (True, "200", 99, 1000): + with self.subTest(status=status): + self.replace_records(path, [{**records[0], "status": status}, *records[1:]]) + with self.assertRaisesRegex(ValueError, "invalid HTTP status"): + bench.summarize_saved(path) + async def test_out_of_order_records_recompute_by_frozen_id(self): release = asyncio.Event() later_completed = asyncio.Event() From 8e42b4b298caeaadea6519bd7bb123c08993ece6 Mon Sep 17 00:00:00 2001 From: linear3735 Date: Tue, 20 Oct 2026 16:30:36 +0800 Subject: [PATCH 5/7] bench: bind saved evidence to validation and wire protocols --- benchmarks/README.md | 43 ++- benchmarks/bench.py | 231 +++++++++++++-- tests/benchmarks/test_bench.py | 4 +- tests/benchmarks/test_saved_protocols.py | 353 +++++++++++++++++++++++ tests/benchmarks/test_score_wire.py | 225 +++++++++++++++ 5 files changed, 833 insertions(+), 23 deletions(-) create mode 100644 tests/benchmarks/test_saved_protocols.py create mode 100644 tests/benchmarks/test_score_wire.py diff --git a/benchmarks/README.md b/benchmarks/README.md index 7b8ba930..114734a3 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -24,8 +24,9 @@ The JSONL format is one object per request: ``` Use a criteria map and a label target for Choice; an ordered criteria list and a -numeric level target for Score. Every question needs a target. The runner adds -only the backend model alias; all other request fields are replayed unchanged. +numeric level target for Score. Every question needs a target. With the default +`--protocol systemone`, the runner adds only the backend model alias; all other +request fields are replayed unchanged. Keep source dataset revision, selected record IDs, sampling seed, transformation code revision, token lengths and truncation decisions with the manifest. Freeze short/near-limit and option-count slices as separate manifests. @@ -75,8 +76,18 @@ the token is not saved. Redirects, proxy environment variables and retries are disabled. `--timeout` is a total deadline for each HTTP exchange, including body receipt; response validation is included in recorded client latency. Each answer must be an object whose `type` matches the requested question type. -Missing or incorrect types are `invalid_response`; the runner keeps the original -body and does not supply missing fields. +Missing or incorrect System One answer types are `invalid_response`; the runner +keeps the original body and does not supply missing fields. +For JEV-VL, use `--protocol jev-vl` with a manifest containing one Choice question +per request. The adapter checks the returned options, index, label and maximal +probability, then supplies the requested `type` in the normalized answer envelope. +Both online collection and offline replay use this same adapter. An image-only +state object is sent as a media-parts list; other state objects remain text data. +Use `--score-decimals 2`, `3` or `4` to match the model's scalar Score precision +(default `4`); this is separate from the GPU dtype in metadata. Probabilities +remain four-decimal wire values. Score validation bounds their independent +rounding together with the scalar's rounding, and accepts Decider's all-zero +Score distribution only with its complete isolated zero-fit metadata. Online validation and offline replay both parse the saved HTTPX-decoded `response.text` with `json.loads`. No BOM is stripped or missing field supplied. An initial U+FEFF is rejected by string JSON parsing. HTTPX may replace invalid @@ -179,6 +190,30 @@ Runs recorded before `warmup_sha256` was added must use their original collector for offline summaries. Do not add a checksum to historical evidence to make it look verified at collection time. +New runs save `protocol`, `score_decimals` and +`validation_protocol=probabilities-4dp-model-score-v1` in the checksummed config. +Offline replay validates these settings before interpreting response records. +Unknown protocol/version values and unsupported or non-integer precisions are +rejected, even if the config checksum matches. Missing protocol and precision +selectors default to `systemone` and `4`. + +For supported archives without `validation_protocol`, the reader uses +`systemone-4dp-score-abs1e-4-v1`: the original typed System One validation and +`math.isclose(..., abs_tol=1e-4)` Score rule. This legacy contract only permits +System One and four-decimal Score. It does not reinterpret old failures using +the newer independent-rounding allowance. These defaults do not make pre-hash +archives readable: runs without the completion/warmup checksums still require +their producing collector. Do not rewrite historical configs or hashes. + +The Score rules reuse [#93](https://github.com/InferMatrixLab/system1-omni/pull/93), +and the wire adapters reuse [#141](https://github.com/InferMatrixLab/system1-omni/pull/141). +The weekly consumer must require `evidence_complete` to be exactly true alongside +zero failures before aggregating measured passes. That consumer change belongs +to #141; retain its companion guard when combining the branches. +CPU regression coverage includes both protocols, supported precisions, +interrupted/torn snapshots, raw-response hashes and online/offline/CLI agreement, +alongside the retained deep-JSON and HTTP 600/999 cases. + For pooled percentiles, first validate each of the three measured rounds with its original collector's `summary`; pool only rounds with the same deployment, manifest and timing/collection settings. Concatenate `latency_ms` from their diff --git a/benchmarks/bench.py b/benchmarks/bench.py index 3973ce58..87df5e46 100644 --- a/benchmarks/bench.py +++ b/benchmarks/bench.py @@ -83,6 +83,25 @@ def number(value): WIRE_DECIMALS = 4 # workers round probabilities on the wire to four decimal places +VALIDATION_PROTOCOL = "probabilities-4dp-model-score-v1" +LEGACY_VALIDATION_PROTOCOL = "systemone-4dp-score-abs1e-4-v1" + + +def validation_settings(config): + version = config.get("validation_protocol", LEGACY_VALIDATION_PROTOCOL) + protocol = config.get("protocol", "systemone") + decimals = config.get("score_decimals", 4) + if version not in (LEGACY_VALIDATION_PROTOCOL, VALIDATION_PROTOCOL): + raise ValueError("unsupported saved validation protocol") + if protocol not in ("systemone", "jev-vl"): + raise ValueError("unsupported wire protocol") + if type(decimals) is not int or decimals not in (2, 3, 4): + raise ValueError("score_decimals must be 2, 3 or 4") + if version == LEGACY_VALIDATION_PROTOCOL and ( + protocol != "systemone" or decimals != 4 + ): + raise ValueError("legacy validation requires System One and four-decimal Score") + return protocol, decimals, version def sum_allowance(count): @@ -93,7 +112,36 @@ def sum_allowance(count): return count * 0.5 * 10**-WIRE_DECIMALS + 1e-9 -def read_answers(case, payload): +def score_allowance(count, score_decimals=4): + """Bound independent rounding of the scalar and its weighted probabilities.""" + return ( + 0.5 * 10**-score_decimals + + 0.5 * 10**-WIRE_DECIMALS * count * (count - 1) / 2 + + 1e-9 + ) + + +def zero_score_fit(answer, probabilities): + """Recognize Decider's isolated Score response when no level has any fit.""" + level_fit = answer.get("level_fit") + return ( + answer.get("type") == "score" + and all(p == 0 for p in probabilities.values()) + and isinstance(level_fit, dict) + and set(level_fit) == set(probabilities) + and all(v == 0 and number(v) for v in level_fit.values()) + and all( + answer.get(key) == 0 and number(answer[key]) + for key in ("score", "confidence", "x_p_max", "fit_mass") + ) + and answer.get("certainty") == 1 + and number(answer["certainty"]) + ) + + +def read_answers( + case, payload, score_decimals=4, validation_protocol=VALIDATION_PROTOCOL +): """Validate wire results and normalize probabilities by label, never by order.""" answers = payload["answers"] if not isinstance(answers, dict) or set(answers) != set( @@ -126,7 +174,12 @@ def read_answers(case, payload): if any(not number(p) or not 0 <= p <= 1 for p in probabilities.values()): raise ValueError("probabilities must be finite and in [0, 1]") total = math.fsum(probabilities.values()) - if abs(total - 1) > sum_allowance(len(probabilities)): + if abs(total - 1) > sum_allowance(len(probabilities)) and not ( + validation_protocol == VALIDATION_PROTOCOL + and kind == "score" + and total == 0 + and zero_score_fit(answer, probabilities) + ): raise ValueError("probabilities must sum to one") if kind == "choice": if value not in probabilities or probabilities[value] != max( @@ -136,8 +189,20 @@ def read_answers(case, payload): elif not number(value): raise ValueError("decision value must be finite") if kind == "score": - expected = sum(int(k) * p for k, p in probabilities.items()) - if not math.isclose(value, expected, abs_tol=1e-4): + if validation_protocol == LEGACY_VALIDATION_PROTOCOL: + expected = sum(int(k) * p for k, p in probabilities.items()) + valid = math.isclose(value, expected, abs_tol=1e-4) + else: + if not 0 <= value <= len(probabilities) - 1: + raise ValueError("score must be in the level range") + expected = math.fsum(int(k) * p for k, p in probabilities.items()) + valid = math.isclose( + value, + expected, + rel_tol=0, + abs_tol=score_allowance(len(probabilities), score_decimals), + ) + if not valid: raise ValueError("score must equal the expected level") normalized[qid] = {"value": value, "probabilities": probabilities} return normalized @@ -232,7 +297,80 @@ def summarize(cases, records, elapsed, attempted_ids=None): } -async def request_one(client, endpoint, case, model, timeout): +def wire_request(case, model, protocol): + if protocol == "systemone": + return {**case["request"], "model": model} + if protocol != "jev-vl": + raise ValueError("unsupported wire protocol") + questions = case["request"]["questions"] + if len(questions) != 1: + raise ValueError("JEV-VL requires one question per request") + question = next(iter(questions.values())) + if question["type"] != "choice": + raise ValueError("the JEV-VL benchmark adapter supports choice only") + state = case["request"]["state"] + if isinstance(state, dict) and set(state) == {"image"}: + # JEV-VL treats a plain object as JSON text; media must be a parts list. + state = [state] + return { + "model": model, + "kind": "choice", + "state": state, + "question": question["instructions"], + "options": [f"{k}: {v}" for k, v in question["criteria"].items()], + } + + +def wire_answers( + case, + payload, + model, + protocol, + score_decimals=4, + validation_protocol=VALIDATION_PROTOCOL, +): + if protocol == "systemone": + return read_answers(case, payload, score_decimals, validation_protocol) + request = wire_request(case, model, protocol) + qid, question = next(iter(case["request"]["questions"].items())) + labels = list(question["criteria"]) + index = payload["choice_index"] + if ( + payload["kind"] != "choice" + or payload["effective_kind"] != "choice" + or payload["options"] != request["options"] + or type(index) is not int + or not 0 <= index < len(labels) + or payload["choice"] != request["options"][index] + or len(payload["probabilities"]) != len(labels) + ): + raise ValueError("JEV-VL decision does not match the request") + return read_answers( + case, + { + "answers": { + qid: { + "type": question["type"], + "choice": labels[index], + "probabilities": dict(zip(labels, payload["probabilities"])), + } + } + }, + score_decimals, + validation_protocol, + ) + + +async def request_one( + client, + endpoint, + case, + model, + timeout, + protocol="systemone", + score_decimals=4, + validation_protocol=VALIDATION_PROTOCOL, +): started = time.perf_counter() record = { "id": case["id"], @@ -246,12 +384,19 @@ async def request_one(client, endpoint, case, model, timeout): try: async with asyncio.timeout(timeout): response = await client.post( - endpoint, json={**case["request"], "model": model} + endpoint, json=wire_request(case, model, protocol) ) record["status"] = response.status_code record["response"] = response.text response.raise_for_status() - record["answers"] = read_answers(case, json.loads(record["response"])) + record["answers"] = wire_answers( + case, + json.loads(record["response"]), + model, + protocol, + score_decimals, + validation_protocol, + ) except httpx.HTTPStatusError as exc: record.update(error=str(exc), error_kind=f"http_{record['status']}") except (TimeoutError, httpx.TimeoutException) as exc: @@ -294,7 +439,19 @@ def measurement_state(output): return state -async def replay(client, endpoint, cases, model, concurrency, timeout, output=None): +async def replay( + client, + endpoint, + cases, + model, + concurrency, + timeout, + output=None, + *, + protocol="systemone", + score_decimals=4, + validation_protocol=VALIDATION_PROTOCOL, +): pending = iter(enumerate(cases)) records = [None] * len(cases) state = measurement_state(output) @@ -302,9 +459,7 @@ async def replay(client, endpoint, cases, model, concurrency, timeout, output=No response_hash = hashlib.sha256() if output is not None: state["responses_sha256"] = response_hash.hexdigest() - responses = (output / "responses.jsonl").open( - "w", encoding="utf-8", newline="" - ) + responses = (output / "responses.jsonl").open("w", encoding="utf-8", newline="") def save_state(): if output is not None: @@ -315,7 +470,16 @@ async def worker(): state["attempted_ids"].append(case["id"]) state["active_ids"].append(case["id"]) save_state() - records[index] = await request_one(client, endpoint, case, model, timeout) + records[index] = await request_one( + client, + endpoint, + case, + model, + timeout, + protocol, + score_decimals, + validation_protocol, + ) record = records[index] if responses is not None: line = json.dumps(record, allow_nan=False) + "\n" @@ -367,6 +531,13 @@ async def worker(): async def run(args, cases): + protocol, score_decimals, version = validation_settings( + { + "protocol": getattr(args, "protocol", "systemone"), + "score_decimals": getattr(args, "score_decimals", 4), + "validation_protocol": VALIDATION_PROTOCOL, + } + ) warmup_manifest = getattr(args, "warmup_manifest", None) warmup_cases = load_cases(warmup_manifest) if warmup_manifest else cases metadata = json.loads(args.metadata.read_text()) @@ -391,6 +562,9 @@ async def run(args, cases): "manifest_sha256": digest(args.manifest), "runner_sha256": digest(Path(__file__)), "metadata": metadata, + "validation_protocol": version, + "score_decimals": score_decimals, + "protocol": protocol, "endpoint": args.endpoint, "model": args.model, "phase": args.phase, @@ -431,6 +605,9 @@ async def run(args, cases): warmup_cases[i % len(warmup_cases)], args.model, args.timeout, + protocol, + score_decimals, + version, ) ) if warmup[-1]["error"] is not None: @@ -454,6 +631,9 @@ async def run(args, cases): args.concurrency, args.timeout, args.output, + protocol=protocol, + score_decimals=score_decimals, + validation_protocol=version, ) except asyncio.CancelledError: # Replay has saved the completed samples and the measured stop boundary. @@ -471,9 +651,15 @@ def summarize_saved(path): if digest(path / "config.json") != state["config_sha256"]: raise ValueError("saved config checksum does not match completion") if "warmup_sha256" not in state: - raise ValueError("saved warmup checksum missing; use the collector that produced this run") - if not (path / "warmup.json").is_file() or digest(path / "warmup.json") != state["warmup_sha256"]: + raise ValueError( + "saved warmup checksum missing; use the collector that produced this run" + ) + if ( + not (path / "warmup.json").is_file() + or digest(path / "warmup.json") != state["warmup_sha256"] + ): raise ValueError("saved warmup checksum does not match completion") + protocol, score_decimals, version = validation_settings(config) if config["latency_basis"] != "post_to_body_json_and_schema_validation": raise ValueError("unsupported saved latency basis") manifest_hash = digest(path / "requests.jsonl") @@ -541,7 +727,8 @@ def summarize_saved(path): status = record["status"] if status is None: if record["response"] is not None or record["error_kind"] not in ( - "timeout", "transport" + "timeout", + "transport", ): raise ValueError("response without status must be a client failure") error_kind = record["error_kind"] @@ -555,8 +742,13 @@ def summarize_saved(path): error_kind = f"http_{status}" else: try: - answers = read_answers( - by_id[record["id"]], json.loads(record["response"]) + answers = wire_answers( + by_id[record["id"]], + json.loads(record["response"]), + config.get("model"), + protocol, + score_decimals, + version, ) except (ValueError, KeyError, TypeError, AttributeError, RecursionError): error_kind = "invalid_response" @@ -608,6 +800,7 @@ def compare(args): for p in (args.reference, args.candidate) ] for path, config in zip((args.reference, args.candidate), configs): + validation_settings(config) if digest(path / "requests.jsonl") != config["manifest_sha256"]: raise ValueError("saved manifest checksum does not match config") if configs[0]["manifest_sha256"] != configs[1]["manifest_sha256"]: @@ -686,6 +879,10 @@ def main(): runner.add_argument( "--model", required=True, help="backend's alias for the pinned checkpoint" ) + runner.add_argument( + "--protocol", choices=("systemone", "jev-vl"), default="systemone" + ) + runner.add_argument("--score-decimals", type=int, choices=(2, 3, 4), default=4) runner.add_argument("--metadata", type=Path, required=True) runner.add_argument( "--output", type=Path, required=True, help="new directory; never overwritten" diff --git a/tests/benchmarks/test_bench.py b/tests/benchmarks/test_bench.py index 8f8b9bde..83299e32 100644 --- a/tests/benchmarks/test_bench.py +++ b/tests/benchmarks/test_bench.py @@ -316,9 +316,9 @@ async def __aiter__(self): clock[0] += 1 yield body - def validate(case, payload): + def validate(case, payload, *settings): clock[0] += 2 - return original(case, payload) + return original(case, payload, *settings) original = bench.read_answers async with httpx.AsyncClient( diff --git a/tests/benchmarks/test_saved_protocols.py b/tests/benchmarks/test_saved_protocols.py new file mode 100644 index 00000000..13dda752 --- /dev/null +++ b/tests/benchmarks/test_saved_protocols.py @@ -0,0 +1,353 @@ +"""Online collection and saved replay must use the same declared wire contract.""" + +import asyncio +import copy +import io +import json +import subprocess +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + +import httpx +from test_bench import CASES, response_for, run_args + +from benchmarks import bench + +VALIDATION = "probabilities-4dp-model-score-v1" +LEGACY_VALIDATION = "systemone-4dp-score-abs1e-4-v1" + + +def score_case(target=0.88, count=2): + return { + "id": "rounded-score", + "request": { + "state": "A saved CPU response fixture", + "questions": { + "q": { + "type": "score", + "instructions": "Choose a level", + "criteria": [str(i) for i in range(count)], + } + }, + }, + "expected": {"q": target}, + } + + +def score_response(value, probabilities): + return { + "answers": { + "q": {"type": "score", "score": value, "probabilities": probabilities} + } + } + + +def jev_response(case): + question = next(iter(case["request"]["questions"].values())) + options = [f"{key}: {value}" for key, value in question["criteria"].items()] + return { + "kind": "choice", + "effective_kind": "choice", + "options": options, + "probabilities": [1.0, 0.0], + "choice_index": 0, + "choice": options[0], + } + + +def replace_config(path, change): + config = json.loads((path / "config.json").read_text()) + change(config) + bench.write_json(path / "config.json", config) + state = json.loads((path / "completion.json").read_text()) + state["config_sha256"] = bench.digest(path / "config.json") + bench.write_json(path / "completion.json", state) + + +class SavedProtocolTests(unittest.IsolatedAsyncioTestCase): + async def collect(self, root, cases, payload, protocol="systemone", decimals=4): + args = run_args(root) + args.manifest = root / "manifest.jsonl" + args.manifest.write_text("".join(json.dumps(case) + "\n" for case in cases)) + args.protocol = protocol + args.score_decimals = decimals + client = httpx.AsyncClient( + transport=httpx.MockTransport(lambda _: httpx.Response(200, json=payload)) + ) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + self.assertEqual(await bench.run(args, cases), 0) + return args.output + + def assert_cli_summary(self, path, expected): + result = subprocess.run( + [sys.executable, str(Path(bench.__file__)), "summary", str(path)], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(json.loads(result.stdout), expected) + + async def test_protocol_and_precision_roundtrip_through_warmup_and_cli(self): + for protocol in ("systemone", "jev-vl"): + for decimals in (2, 3, 4): + with ( + self.subTest(protocol=protocol, decimals=decimals), + tempfile.TemporaryDirectory() as directory, + ): + scalar = round(0.8766, decimals) + case = score_case(scalar) if protocol == "systemone" else CASES[0] + payload = ( + score_response(scalar, {"0": 0.1234, "1": 0.8766}) + if protocol == "systemone" + else jev_response(case) + ) + path = await self.collect( + Path(directory), [case], payload, protocol, decimals + ) + config = json.loads((path / "config.json").read_text()) + self.assertEqual(config["protocol"], protocol) + self.assertEqual(config["score_decimals"], decimals) + self.assertEqual(config["validation_protocol"], VALIDATION) + warmup = json.loads((path / "warmup.json").read_text()) + records = [ + json.loads(row) + for row in (path / "responses.jsonl").read_text().splitlines() + ] + for record in [*warmup, *records]: + self.assertIsNone(record["error"]) + self.assertEqual( + record["response_sha256"], + bench.response_digest(record["response"]), + ) + summary = json.loads((path / "summary.json").read_text()) + self.assertTrue(summary["evidence_complete"]) + self.assertEqual(summary["successful_requests"], 1) + self.assertEqual(bench.summarize_saved(path), summary) + self.assert_cli_summary(path, summary) + + async def test_unknown_saved_contracts_are_rejected_after_config_hash_matches(self): + with tempfile.TemporaryDirectory() as directory: + path = await self.collect( + Path(directory), [CASES[0]], response_for(CASES[0]) + ) + config = json.loads((path / "config.json").read_text()) + mutations = ( + ("protocol", "unknown"), + ("protocol", None), + ("validation_protocol", "future-v999"), + ("validation_protocol", None), + ("score_decimals", 9), + ("score_decimals", "2"), + ("score_decimals", 2.0), + ("score_decimals", True), + ("score_decimals", None), + ) + for key, value in mutations: + with self.subTest(key=key, value=value): + replace_config( + path, + lambda saved, key=key, value=value: saved.update( + config | {key: value} + ), + ) + with self.assertRaisesRegex( + ValueError, "protocol|precision|score_decimals|validation" + ): + bench.summarize_saved(path) + + async def test_missing_contract_fields_keep_legacy_systemone_four_decimal_defaults( + self, + ): + with tempfile.TemporaryDirectory() as directory: + path = await self.collect( + Path(directory), [CASES[0]], response_for(CASES[0]) + ) + replace_config( + path, + lambda config: [ + config.pop(key) + for key in ("protocol", "score_decimals", "validation_protocol") + ], + ) + summary = bench.summarize_saved(path) + self.assertTrue(summary["evidence_complete"]) + self.assertEqual(summary["failed_requests"], 0) + self.assert_cli_summary(path, summary) + + async def test_legacy_score_failure_is_not_reinterpreted_by_the_new_rounding_rule( + self, + ): + case = score_case(2.003, 4) + payload = score_response( + 2.003, {"0": 0.0337, "1": 0.0974, "2": 0.701, "3": 0.1678} + ) + with tempfile.TemporaryDirectory() as directory: + path = await self.collect(Path(directory), [case], payload) + records = [ + json.loads(row) + for row in (path / "responses.jsonl").read_text().splitlines() + ] + records[0].update( + answers={}, + error="score must equal the expected level", + error_kind="invalid_response", + ) + (path / "responses.jsonl").write_text( + "".join(json.dumps(record) + "\n" for record in records) + ) + state = json.loads((path / "completion.json").read_text()) + state["responses_sha256"] = bench.digest(path / "responses.jsonl") + bench.write_json(path / "completion.json", state) + replace_config( + path, + lambda config: [ + config.pop(key) + for key in ("protocol", "score_decimals", "validation_protocol") + ], + ) + summary = bench.summarize_saved(path) + self.assertTrue(summary["evidence_complete"]) + self.assertEqual(summary["errors"], {"invalid_response": 1}) + self.assertEqual(summary["successful_requests"], 0) + self.assert_cli_summary(path, summary) + replace_config( + path, lambda config: config.update(validation_protocol=VALIDATION) + ) + with self.assertRaisesRegex(ValueError, "saved validation result disagree"): + bench.summarize_saved(path) + + async def test_legacy_contract_cannot_claim_another_precision_or_protocol(self): + with tempfile.TemporaryDirectory() as directory: + path = await self.collect( + Path(directory), [CASES[0]], response_for(CASES[0]) + ) + config = json.loads((path / "config.json").read_text()) + for change in ({"protocol": "jev-vl"}, {"score_decimals": 2}): + with self.subTest(change=change): + replace_config( + path, + lambda saved, change=change: saved.update( + config | {"validation_protocol": LEGACY_VALIDATION} | change + ), + ) + with self.assertRaisesRegex( + ValueError, + "legacy|protocol|precision|score_decimals|validation", + ): + bench.summarize_saved(path) + + async def test_raw_hash_and_torn_snapshot_checks_apply_to_both_protocols(self): + for protocol in ("systemone", "jev-vl"): + with ( + self.subTest(protocol=protocol), + tempfile.TemporaryDirectory() as directory, + ): + payload = ( + response_for(CASES[0]) + if protocol == "systemone" + else jev_response(CASES[0]) + ) + path = await self.collect( + Path(directory), [CASES[0]], payload, protocol + ) + responses = (path / "responses.jsonl").read_bytes() + state_bytes = (path / "completion.json").read_bytes() + records = [json.loads(row) for row in responses.decode().splitlines()] + records[0]["response"] += " " + (path / "responses.jsonl").write_text( + "".join(json.dumps(record) + "\n" for record in records) + ) + state = json.loads(state_bytes) + state["responses_sha256"] = bench.digest(path / "responses.jsonl") + bench.write_json(path / "completion.json", state) + with self.assertRaisesRegex(ValueError, "raw response checksum"): + bench.summarize_saved(path) + (path / "responses.jsonl").write_bytes(responses + b'{"id":') + (path / "completion.json").write_bytes(state_bytes) + with self.assertRaisesRegex(ValueError, "saved responses checksum"): + bench.summarize_saved(path) + + async def test_interrupted_dual_protocol_runs_recompute_without_claiming_completion( + self, + ): + for protocol in ("systemone", "jev-vl"): + with ( + self.subTest(protocol=protocol), + tempfile.TemporaryDirectory() as directory, + ): + root = Path(directory) + cases = [CASES[0], {**CASES[0], "id": "uncompleted-choice"}] + args = run_args(root) + args.protocol = protocol + args.score_decimals = 4 + args.manifest = root / "manifest.jsonl" + args.manifest.write_text( + "".join(json.dumps(case) + "\n" for case in cases) + ) + waiting = asyncio.Event() + calls = 0 + + async def handler(request, protocol=protocol, waiting=waiting): + nonlocal calls + calls += 1 + if calls < 3: + payload = ( + response_for(CASES[0]) + if protocol == "systemone" + else jev_response(CASES[0]) + ) + return httpx.Response(200, json=payload) + waiting.set() + await asyncio.Event().wait() + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + task = asyncio.create_task(bench.run(args, cases)) + await asyncio.wait_for(waiting.wait(), 2) + task.cancel() + self.assertEqual(await task, 1) + summary = json.loads((args.output / "summary.json").read_text()) + self.assertFalse(summary["evidence_complete"]) + self.assertEqual(summary["failed_requests"], 0) + self.assertEqual(summary["incomplete_requests"], 1) + self.assertEqual(summary, bench.summarize_saved(args.output)) + self.assert_cli_summary(args.output, summary) + + +class AdapterContractTests(unittest.TestCase): + def test_systemone_missing_or_wrong_answer_type_still_fails(self): + case = CASES[0] + qid = next(iter(case["request"]["questions"])) + for value in (None, "score"): + payload = copy.deepcopy(response_for(case)) + if value is None: + payload["answers"][qid].pop("type") + else: + payload["answers"][qid]["type"] = value + with self.subTest(value=value), self.assertRaises(ValueError): + bench.wire_answers(case, payload, "test", "systemone") + + def test_jev_normalization_preserves_the_requested_type_for_strict_validation(self): + case = CASES[0] + result = bench.wire_answers(case, jev_response(case), "test", "jev-vl") + self.assertEqual(result, bench.read_answers(case, response_for(case))) + for field in ("kind", "effective_kind"): + with self.subTest(field=field), self.assertRaises(ValueError): + bench.wire_answers( + case, jev_response(case) | {field: "score"}, "test", "jev-vl" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/benchmarks/test_score_wire.py b/tests/benchmarks/test_score_wire.py new file mode 100644 index 00000000..59d4a751 --- /dev/null +++ b/tests/benchmarks/test_score_wire.py @@ -0,0 +1,225 @@ +"""Score wire contracts: model precision and Decider's isolated zero-fit output.""" + +import argparse +import asyncio +import copy +import io +import json +import random +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + +import httpx + +from benchmarks import bench + + +def case_for(count): + return { + "request": { + "questions": { + "q": {"type": "score", "criteria": [str(i) for i in range(count)]} + } + } + } + + +def payload(score, probabilities): + return { + "answers": { + "q": {"type": "score", "score": score, "probabilities": probabilities} + } + } + + +class ScoreWireTests(unittest.TestCase): + def test_two_decimal_contract_is_explicit(self): + response = payload(0.88, {"0": 0.1234, "1": 0.8766}) + with self.assertRaisesRegex(ValueError, "expected level"): + bench.read_answers(case_for(2), response) + self.assertEqual( + bench.read_answers(case_for(2), response, score_decimals=2)["q"]["value"], + 0.88, + ) + with self.assertRaisesRegex(ValueError, "expected level"): + bench.read_answers( + case_for(2), payload(0.9, {"0": 0.1234, "1": 0.8766}), score_decimals=2 + ) + + def test_observed_laya_independent_rounding(self): + probabilities = {"0": 0.0337, "1": 0.0974, "2": 0.701, "3": 0.1678} + result = bench.read_answers(case_for(4), payload(2.003, probabilities)) + self.assertEqual(result["q"]["probabilities"], probabilities) + with self.assertRaisesRegex(ValueError, "expected level"): + bench.read_answers(case_for(4), payload(2.007, probabilities)) + + def test_independently_rounded_distributions(self): + rng = random.Random(93) + for count in range(2, 11): + for decimals in (2, 3, 4): + for _ in range(32): + weights = [rng.random() for _ in range(count)] + probabilities = [w / sum(weights) for w in weights] + score = round( + sum(i * p for i, p in enumerate(probabilities)), decimals + ) + wire = { + str(i): round(probabilities[i], 4) + for i in reversed(range(count)) + } + self.assertEqual( + bench.read_answers( + case_for(count), + payload(score, wire), + score_decimals=decimals, + )["q"]["probabilities"], + wire, + ) + + def test_exact_score_range_and_invalid_distribution(self): + for score in (-0.00001, 1.00001, float("nan"), True): + with self.subTest(score=score), self.assertRaises(ValueError): + bench.read_answers( + case_for(2), payload(score, {"0": 0, "1": 1}), score_decimals=2 + ) + for probabilities in ( + {"0": 0, "1": 0}, + {"0": 0.2, "1": 0.2}, + {"0": True, "1": 0}, + ): + with self.assertRaises(ValueError): + bench.read_answers(case_for(2), payload(0, probabilities)) + + def test_only_complete_zero_fit_metadata_is_accepted(self): + response = payload(0, {"0": 0, "1": 0}) + answer = response["answers"]["q"] + answer.update( + type="score", + level_fit={"0": 0, "1": 0}, + confidence=0, + x_p_max=0, + fit_mass=0, + certainty=1, + ) + bench.read_answers(case_for(2), response) + for key in ( + "type", + "level_fit", + "confidence", + "x_p_max", + "fit_mass", + "certainty", + ): + missing = copy.deepcopy(response) + del missing["answers"]["q"][key] + with self.subTest(key=key), self.assertRaises(ValueError): + bench.read_answers(case_for(2), missing) + for key in ("score", "confidence", "x_p_max", "fit_mass", "certainty"): + bad = copy.deepcopy(response) + bad["answers"]["q"][key] = True + with self.subTest(key=key), self.assertRaises(ValueError): + bench.read_answers(case_for(2), bad) + bad = copy.deepcopy(response) + bad["answers"]["q"]["level_fit"]["0"] = 10**1000 + with self.assertRaises(ValueError): + bench.read_answers(case_for(2), bad) + + def test_transport_threads_model_precision(self): + async def check(): + response = payload(0.88, {"0": 0.1234, "1": 0.8766}) + async with httpx.AsyncClient( + transport=httpx.MockTransport( + lambda _: httpx.Response(200, json=response) + ) + ) as client: + records, _ = await bench.replay( + client, + "http://localhost/v1/systemone", + [case_for(2) | {"id": "score"}], + "decider", + 1, + 1, + score_decimals=2, + ) + self.assertIsNone(records[0]["error"]) + + bad = payload(0, {"0": 0, "1": 0}) + bad["answers"]["q"].update( + type="score", + level_fit={"0": 10**1000, "1": 0}, + confidence=0, + x_p_max=0, + fit_mass=0, + certainty=1, + ) + async with httpx.AsyncClient( + transport=httpx.MockTransport(lambda _: httpx.Response(200, json=bad)) + ) as client: + record = await bench.request_one( + client, + "http://localhost/v1/systemone", + case_for(2) | {"id": "score"}, + "decider", + 1, + score_decimals=2, + ) + self.assertEqual(record["error_kind"], "invalid_response") + + asyncio.run(check()) + + def test_run_saves_precision_and_uses_it_for_warmup(self): + async def check(): + case = case_for(2) | {"id": "score", "expected": {"q": 0.88}} + case["request"]["state"] = "CPU fixture" + response = payload(0.88, {"0": 0.1234, "1": 0.8766}) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + metadata = { + key: "CPU fixture" + for key in ( + "gpu", + "gpu_ids", + "driver", + "cuda", + "precision", + "model_revision", + "runtime_revision", + "cache_policy", + "cuda_evidence", + "reservation", + ) + } + bench.write_json(root / "metadata.json", metadata) + (root / "manifest.jsonl").write_text(json.dumps(case) + "\n") + args = argparse.Namespace( + metadata=root / "metadata.json", + output=root / "run", + manifest=root / "manifest.jsonl", + endpoint="http://localhost/v1/systemone", + model="decider", + phase="feasibility", + concurrency=1, + warmup=1, + timeout=1, + score_decimals=2, + ) + client = httpx.AsyncClient( + transport=httpx.MockTransport( + lambda _: httpx.Response(200, json=response) + ) + ) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + self.assertEqual(await bench.run(args, [case]), 0) + config = json.loads((args.output / "config.json").read_text()) + self.assertEqual(config["score_decimals"], 2) + self.assertEqual( + config["validation_protocol"], "probabilities-4dp-model-score-v1" + ) + + asyncio.run(check()) From 82c108ede95c495ec5b57eba52fbdd044fc59116 Mon Sep 17 00:00:00 2001 From: linear3735 Date: Wed, 21 Oct 2026 22:50:14 +0800 Subject: [PATCH 6/7] bench: reject adapter-incompatible manifests before creating the archive With --protocol jev-vl, wire_request raises before any HTTP exchange for a manifest the adapter cannot express (multi-question or non-choice cases). request_one recorded that failure as status=None with error_kind invalid_response, and summarize_saved then rejected the record ("response without status must be a client failure"), leaving a saved archive the tool itself could never summarize again. run() now pre-flights wire_request over the manifest and warmup cases and raises a startup error naming the case before the output directory exists, so every saved archive stays summarizable. Co-Authored-By: Claude Code --- benchmarks/bench.py | 10 ++++ tests/benchmarks/test_saved_protocols.py | 61 ++++++++++++++++++++++++ 2 files changed, 71 insertions(+) diff --git a/benchmarks/bench.py b/benchmarks/bench.py index 87df5e46..b30cdc34 100644 --- a/benchmarks/bench.py +++ b/benchmarks/bench.py @@ -555,6 +555,16 @@ async def run(args, cases): ): if not metadata.get(name): raise ValueError(f"metadata requires {name}") + # A case the adapter cannot wire fails before any HTTP exchange; reject it + # before creating the output directory so a saved archive stays summarizable. + for case in cases if warmup_cases is cases else [*cases, *warmup_cases]: + try: + wire_request(case, args.model, protocol) + except (ValueError, KeyError, TypeError, AttributeError) as exc: + raise ValueError( + f"case {case['id']!r} is incompatible with the {protocol} " + f"wire protocol: {exc}" + ) from exc args.output.mkdir(parents=True, exist_ok=False) write_json( args.output / "config.json", diff --git a/tests/benchmarks/test_saved_protocols.py b/tests/benchmarks/test_saved_protocols.py index 13dda752..c36b26be 100644 --- a/tests/benchmarks/test_saved_protocols.py +++ b/tests/benchmarks/test_saved_protocols.py @@ -325,6 +325,67 @@ async def handler(request, protocol=protocol, waiting=waiting): self.assert_cli_summary(args.output, summary) +def choice_case(case_id): + return { + "id": case_id, + "request": { + "state": "A saved CPU state fixture", + "questions": { + "q": { + "type": "choice", + "instructions": "Pick one", + "criteria": {"a": "A", "b": "B"}, + } + }, + }, + "expected": {"q": "a"}, + } + + +def multi_question_case(): + case = choice_case("two-questions") + case["request"]["questions"]["r"] = { + "type": "choice", + "instructions": "Pick another", + "criteria": {"a": "A", "b": "B"}, + } + case["expected"]["r"] = "a" + return case + + +class WirePreflightTests(unittest.IsolatedAsyncioTestCase): + def jev_args(self, root): + args = run_args(root) + args.manifest = root / "manifest.jsonl" + args.protocol = "jev-vl" + return args + + async def assert_rejected_before_any_artifact(self, args, cases): + with self.assertRaisesRegex( + ValueError, "incompatible with the jev-vl wire protocol" + ): + await bench.run(args, cases) + self.assertFalse(args.output.exists()) + + async def test_incompatible_cases_are_rejected_before_any_artifact(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.jev_args(root) + args.manifest.write_text(json.dumps(choice_case("ok")) + "\n") + await self.assert_rejected_before_any_artifact( + args, [choice_case("ok"), multi_question_case(), score_case()] + ) + + async def test_incompatible_warmup_manifest_is_rejected_before_any_artifact(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.jev_args(root) + args.manifest.write_text(json.dumps(choice_case("ok")) + "\n") + args.warmup_manifest = root / "warmup.jsonl" + args.warmup_manifest.write_text(json.dumps(multi_question_case()) + "\n") + await self.assert_rejected_before_any_artifact(args, [choice_case("ok")]) + + class AdapterContractTests(unittest.TestCase): def test_systemone_missing_or_wrong_answer_type_still_fails(self): case = CASES[0] From ef13e0ccd144409304f256d91db7bf509bc31f7c Mon Sep 17 00:00:00 2001 From: linear3735 Date: Fri, 23 Oct 2026 13:06:09 +0800 Subject: [PATCH 7/7] bench: preserve response bytes and reject invalid UTF-8 --- benchmarks/README.md | 35 ++-- benchmarks/bench.py | 58 +++++- tests/benchmarks/test_bench.py | 8 +- tests/benchmarks/test_response_bytes.py | 237 +++++++++++++++++++++++ tests/benchmarks/test_saved_protocols.py | 2 +- 5 files changed, 320 insertions(+), 20 deletions(-) create mode 100644 tests/benchmarks/test_response_bytes.py diff --git a/benchmarks/README.md b/benchmarks/README.md index 2c6ae70b..807cc777 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -88,12 +88,12 @@ Use `--score-decimals 2`, `3` or `4` to match the model's scalar Score precision remain four-decimal wire values. Score validation bounds their independent rounding together with the scalar's rounding, and accepts Decider's all-zero Score distribution only with its complete isolated zero-fit metadata. -Online validation and offline replay both parse the saved HTTPX-decoded -`response.text` with `json.loads`. No BOM is stripped or missing field supplied. -An initial U+FEFF is rejected by string JSON parsing. HTTPX may replace invalid -encoded bytes with U+FFFD; if that exact saved text is valid JSON and passes the -same schema checks, it is accepted. This is a decoded-text contract, not a -claim that the original HTTP bytes were valid UTF-8. +New runs preserve HTTPX `response.content` bytes and decode them as strict UTF-8 +before JSON/schema validation, both online and offline. Invalid UTF-8 is an +`invalid_response`, even in an otherwise valid answer's unused field. HTTP charset +headers do not override this JSON contract. No BOM is stripped or missing field +supplied; an initial U+FEFF remains invalid. These bytes are HTTPX's body after +HTTP content decoding, not compressed wire bytes or HTTP framing. For Decider, pass `--score-decimals 2` in both feasibility and measured runs; its scalar Score is rounded to two decimals while probabilities use four. @@ -145,9 +145,11 @@ Outputs: - `responses.jsonl`: flushed after each request finishes, in completion order during replay, then atomically reordered to manifest input order for a complete traversal. Interrupted runs retain their actual saved completion order, - with the frozen request ID, original response text and its SHA256, HTTP status, + with the frozen request ID, lossless `response_bytes_base64`, its SHA256, HTTP status, validation result, client latency and any error. Join records to the manifest - by ID; `compare` accepts different completion orders. + by ID; `compare` accepts different completion orders. The `response` text is a + display copy decoded with replacement, never the new format's validation input. + No HTTP body is represented by null; an empty HTTP body has an empty base64 string. - `completion.json`: atomically replaced before each measured request starts and after it finishes. Contains attempted/completed/active IDs, input/config/response checksums, measured start, terminal stop reason and real measured wall time. @@ -177,7 +179,7 @@ historical body-only client timing. A soft interruption saves finished samples and records cancelled requests as `incomplete`, with `elapsed_until_abort_ms` and no completed `latency_ms`. If a -complete body was received before cancellation, its text remains available for +complete body was received before cancellation, its bytes remain available for independent offline scoring. Abort durations do not enter latency percentiles. Cancellation of the client does not prove the server stopped processing. @@ -197,7 +199,7 @@ python benchmarks/bench.py summary /tmp/laya-c1-run1 This command makes no requests and prints the recomputed summary without changing saved files. It checks frozen IDs, duplicates and unknown entries, saved manifest, -config, warmup and response checksums, and agreement between original response text and +config, warmup and response checksums, and agreement between saved response bodies and normalized answers/validation. It requires the new `completion.json` evidence; older runs remain usable by `compare` but cannot acquire unrecorded timing or interruption boundaries through this command. @@ -212,6 +214,15 @@ Unknown protocol/version values and unsupported or non-integer precisions are rejected, even if the config checksum matches. Missing protocol and precision selectors default to `systemone` and `4`. +The independent `response_format=httpx-content-base64-utf8-v1` selector binds the +new byte evidence and strict decoder to the config checksum. Offline replay +checks base64, the original byte SHA256 and the display text's consistency before +revalidating answers. Unknown response formats are rejected. Supported archives +without this selector, or with `httpx-text-v1`, retain their original saved-text +parsing and text SHA256 semantics, including HTTPX replacement decoding. Such +archives cannot establish the validity of the original UTF-8 bytes. Do not add +byte fields or rewrite old configs/hashes to make historical evidence look verified. + For supported archives without `validation_protocol`, the reader uses `systemone-4dp-score-abs1e-4-v1`: the original typed System One validation and `math.isclose(..., abs_tol=1e-4)` Score rule. This legacy contract only permits @@ -342,6 +353,8 @@ rounding contract (default: four decimals; Decider: two). Probability precision remains four decimals. The validator bounds independent scalar and weighted probability rounding, requires the exact level range, and preserves wire values. The same optional `score_decimals` argument is accepted by `read_answers`, -`request_one` and `replay`; run configuration records it for reproducibility. +`request_one` and `replay`; pass it by keyword when calling the collector, since +`replay` retains its positional `output` argument. Run configuration records it +for reproducibility. Decider's isolated all-zero Score distribution requires complete consistent zero-fit metadata; other non-normalized distributions remain invalid. diff --git a/benchmarks/bench.py b/benchmarks/bench.py index b30cdc34..4bd60447 100755 --- a/benchmarks/bench.py +++ b/benchmarks/bench.py @@ -3,6 +3,7 @@ import argparse import asyncio +import base64 import hashlib import json import math @@ -33,6 +34,43 @@ def response_digest(text): return hashlib.sha256(text.encode("utf-8")).hexdigest() +RESPONSE_FORMAT = "httpx-content-base64-utf8-v1" +LEGACY_RESPONSE_FORMAT = "httpx-text-v1" + + +def saved_response_format(config): + value = config.get("response_format", LEGACY_RESPONSE_FORMAT) + if value not in (RESPONSE_FORMAT, LEGACY_RESPONSE_FORMAT): + raise ValueError("unsupported saved response format") + return value + + +def response_body(record, response_format): + if response_format == LEGACY_RESPONSE_FORMAT: + body = record["response"] + checksum = response_digest(body) + else: + encoded = record["response_bytes_base64"] + if encoded is not None and not isinstance(encoded, str): + raise ValueError("saved response base64 must be text") + body = base64.b64decode(encoded, validate=True) if encoded is not None else None + checksum = hashlib.sha256(body).hexdigest() if body is not None else None + text = body.decode("utf-8", errors="replace") if body is not None else None + if record["response"] != text: + raise ValueError("saved response text does not match its bytes") + if body is None and record["status"] is not None: + raise ValueError("HTTP response is missing its saved bytes") + if record["response_sha256"] != checksum: + raise ValueError("raw response checksum mismatch") + return body + + +def response_payload(body): + if isinstance(body, bytes): + body = body.decode("utf-8") + return json.loads(body) + + def load_cases(path): cases = [json.loads(line) for line in path.read_text().splitlines() if line.strip()] if not cases or len({c["id"] for c in cases}) != len(cases): @@ -378,20 +416,24 @@ async def request_one( "error": None, "error_kind": None, "response": None, + "response_bytes_base64": None, "answers": {}, "state": "completed", } + body = None try: async with asyncio.timeout(timeout): response = await client.post( endpoint, json=wire_request(case, model, protocol) ) record["status"] = response.status_code - record["response"] = response.text + body = response.content + record["response_bytes_base64"] = base64.b64encode(body).decode("ascii") + record["response"] = body.decode("utf-8", errors="replace") response.raise_for_status() record["answers"] = wire_answers( case, - json.loads(record["response"]), + response_payload(body), model, protocol, score_decimals, @@ -417,7 +459,9 @@ async def request_one( record["latency_ms"] = duration if record["state"] == "completed" else None if record["state"] == "incomplete": record["elapsed_until_abort_ms"] = duration - record["response_sha256"] = response_digest(record["response"]) + record["response_sha256"] = ( + hashlib.sha256(body).hexdigest() if body is not None else None + ) return record @@ -575,6 +619,7 @@ async def run(args, cases): "validation_protocol": version, "score_decimals": score_decimals, "protocol": protocol, + "response_format": RESPONSE_FORMAT, "endpoint": args.endpoint, "model": args.model, "phase": args.phase, @@ -670,6 +715,7 @@ def summarize_saved(path): ): raise ValueError("saved warmup checksum does not match completion") protocol, score_decimals, version = validation_settings(config) + response_format = saved_response_format(config) if config["latency_basis"] != "post_to_body_json_and_schema_validation": raise ValueError("unsupported saved latency basis") manifest_hash = digest(path / "requests.jsonl") @@ -710,8 +756,7 @@ def summarize_saved(path): raise ValueError("duplicate, unknown or unattempted response ID") completed_ids = set() for record in records: - if record["response_sha256"] != response_digest(record["response"]): - raise ValueError("raw response checksum mismatch") + body = response_body(record, response_format) if record["state"] == "incomplete": if ( record["latency_ms"] is not None @@ -754,7 +799,7 @@ def summarize_saved(path): try: answers = wire_answers( by_id[record["id"]], - json.loads(record["response"]), + response_payload(body), config.get("model"), protocol, score_decimals, @@ -811,6 +856,7 @@ def compare(args): ] for path, config in zip((args.reference, args.candidate), configs): validation_settings(config) + saved_response_format(config) if digest(path / "requests.jsonl") != config["manifest_sha256"]: raise ValueError("saved manifest checksum does not match config") if configs[0]["manifest_sha256"] != configs[1]["manifest_sha256"]: diff --git a/tests/benchmarks/test_bench.py b/tests/benchmarks/test_bench.py index 83299e32..7d0ed4ae 100644 --- a/tests/benchmarks/test_bench.py +++ b/tests/benchmarks/test_bench.py @@ -1,5 +1,6 @@ import argparse import asyncio +import base64 import io import json import tempfile @@ -46,6 +47,7 @@ def saved_inputs(path, cases=CASES): "manifest_sha256": bench.digest(path / "requests.jsonl"), "warmup_manifest_sha256": None, "latency_basis": "post_to_body_json_and_schema_validation", + "response_format": bench.RESPONSE_FORMAT, }, ) @@ -381,11 +383,11 @@ def handler(request): class PersistenceTests(unittest.IsolatedAsyncioTestCase): - async def test_bom_and_invalid_utf8_online_offline_text_decoding_agree(self): + async def test_bom_and_invalid_utf8_online_offline_rejection_agrees(self): valid = json.dumps(response_for(CASES[0])).encode("utf-8") for label, body, error_kind in ( ("bom", b"\xef\xbb\xbf" + valid, "invalid_response"), - ("invalid-utf8", valid[:-1] + b', "unused": "\xff"}', None), + ("invalid-utf8", valid[:-1] + b', "unused": "\xff"}', "invalid_response"), ): with self.subTest(label=label), tempfile.TemporaryDirectory() as directory: path = Path(directory) / "run" @@ -398,6 +400,8 @@ async def test_bom_and_invalid_utf8_online_offline_text_decoding_agree(self): client, "http://localhost/v1/systemone", CASES[:1], "english", 1, 1, path ) self.assertEqual(records[0]["response"], expected_text) + self.assertEqual(base64.b64decode(records[0]["response_bytes_base64"]), body) + self.assertEqual(records[0]["response_sha256"], bench.hashlib.sha256(body).hexdigest()) self.assertEqual(records[0]["error_kind"], error_kind) self.assertTrue(expected_text.startswith("\ufeff") if label == "bom" else "\ufffd" in expected_text) expected = bench.summarize(CASES[:1], records, elapsed) diff --git a/tests/benchmarks/test_response_bytes.py b/tests/benchmarks/test_response_bytes.py new file mode 100644 index 00000000..9b2e0d1a --- /dev/null +++ b/tests/benchmarks/test_response_bytes.py @@ -0,0 +1,237 @@ +"""Response archives must retain and validate the bytes received by HTTPX.""" + +import base64 +import copy +import hashlib +import io +import json +import subprocess +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + +import httpx +from test_bench import CASES, response_for, run_args +from test_saved_protocols import jev_response, replace_config + +from benchmarks import bench + +FORMAT = "httpx-content-base64-utf8-v1" + + +def body_for(protocol, note="original"): + payload = ( + response_for(CASES[0]) if protocol == "systemone" else jev_response(CASES[0]) + ) + return json.dumps(payload | {"note": note}, ensure_ascii=False).encode("utf-8") + + +class ResponseBytesTests(unittest.IsolatedAsyncioTestCase): + async def collect( + self, root, raw, protocol="systemone", status=200, warmup=0, headers=None + ): + args = run_args(root) + args.protocol = protocol + args.warmup = warmup + args.manifest = root / "manifest.jsonl" + args.manifest.write_text(json.dumps(CASES[0]) + "\n") + + def handler(request): + if raw is None: + raise httpx.ConnectError( + "controlled transport failure", request=request + ) + return httpx.Response(status, content=raw, headers=headers) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with ( + patch.object(bench.httpx, "AsyncClient", return_value=client), + redirect_stdout(io.StringIO()), + ): + if warmup: + with self.assertRaisesRegex(ValueError, "readiness/warmup failed"): + await bench.run(args, [CASES[0]]) + else: + await bench.run(args, [CASES[0]]) + return args.output + + def records(self, path): + return [ + json.loads(line) + for line in (path / "responses.jsonl").read_text().splitlines() + ] + + def assert_original_bytes(self, record, raw): + self.assertEqual( + base64.b64decode(record["response_bytes_base64"], validate=True), raw + ) + self.assertEqual(record["response_sha256"], hashlib.sha256(raw).hexdigest()) + self.assertEqual(record["response"], raw.decode("utf-8", errors="replace")) + + def assert_cli_summary(self, path, expected): + result = subprocess.run( + [sys.executable, str(Path(bench.__file__)), "summary", str(path)], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(json.loads(result.stdout), expected) + + async def test_invalid_utf8_roundtrips_as_failed_response_in_both_protocols(self): + for protocol in ("systemone", "jev-vl"): + with ( + self.subTest(protocol=protocol), + tempfile.TemporaryDirectory() as directory, + ): + raw = body_for(protocol).replace(b"original", b"\xff") + path = await self.collect(Path(directory), raw, protocol) + record = self.records(path)[0] + self.assertEqual(record["error_kind"], "invalid_response") + self.assertEqual(record["answers"], {}) + self.assert_original_bytes(record, raw) + config = json.loads((path / "config.json").read_text()) + self.assertEqual(config["response_format"], FORMAT) + summary = json.loads((path / "summary.json").read_text()) + self.assertEqual(summary["errors"], {"invalid_response": 1}) + self.assertEqual(summary["successful_requests"], 0) + self.assertEqual(summary, bench.summarize_saved(path)) + self.assert_cli_summary(path, summary) + + async def test_utf8_decoding_ignores_a_conflicting_http_charset(self): + with tempfile.TemporaryDirectory() as directory: + raw = body_for("systemone", "原始 café 😀") + path = await self.collect( + Path(directory), + raw, + headers={"content-type": "application/json; charset=latin-1"}, + ) + record = self.records(path)[0] + self.assertIsNone(record["error_kind"]) + self.assert_original_bytes(record, raw) + self.assertEqual(bench.summarize_saved(path)["successful_requests"], 1) + + async def test_empty_body_bom_and_unusual_http_statuses_preserve_bytes(self): + for raw, status, kind in ( + (b"", 200, "invalid_response"), + (b"\xef\xbb\xbf" + body_for("systemone"), 200, "invalid_response"), + (b"\xff", 600, "http_600"), + (b"\xfe", 999, "http_999"), + ): + with ( + self.subTest(status=status, raw=raw[:3]), + tempfile.TemporaryDirectory() as directory, + ): + path = await self.collect(Path(directory), raw, status=status) + record = self.records(path)[0] + self.assertEqual(record["error_kind"], kind) + self.assert_original_bytes(record, raw) + self.assertEqual(bench.summarize_saved(path)["errors"], {kind: 1}) + + async def test_no_http_body_is_distinct_from_an_empty_body(self): + with tempfile.TemporaryDirectory() as directory: + path = await self.collect(Path(directory), None) + record = self.records(path)[0] + self.assertEqual(record["error_kind"], "transport") + for field in ("response", "response_bytes_base64", "response_sha256"): + self.assertIsNone(record[field]) + self.assertEqual(bench.summarize_saved(path)["errors"], {"transport": 1}) + + async def test_inner_byte_evidence_is_checked_after_outer_checksum_is_resigned( + self, + ): + with tempfile.TemporaryDirectory() as directory: + path = await self.collect(Path(directory), body_for("systemone")) + records = self.records(path) + mutations = ( + {"response_bytes_base64": "not*base64"}, + {"response_bytes_base64": 1}, + {"response_bytes_base64": None}, + {"response_sha256": hashlib.sha256(b"other").hexdigest()}, + {"response": records[0]["response"] + " "}, + ) + for change in mutations: + with self.subTest(change=change): + changed = copy.deepcopy(records) + changed[0].update(change) + (path / "responses.jsonl").write_text(json.dumps(changed[0]) + "\n") + state = json.loads((path / "completion.json").read_text()) + state["responses_sha256"] = bench.digest(path / "responses.jsonl") + bench.write_json(path / "completion.json", state) + with self.assertRaises(ValueError): + bench.summarize_saved(path) + + async def test_unknown_response_format_is_not_guessed(self): + for value in ("future-format-v99", None, 1): + with self.subTest(value=value), tempfile.TemporaryDirectory() as directory: + path = await self.collect(Path(directory), body_for("systemone")) + replace_config( + path, + lambda config, value=value: config.update(response_format=value), + ) + with self.assertRaisesRegex(ValueError, "response|format"): + bench.summarize_saved(path) + + async def test_legacy_text_archive_keeps_its_replacement_decoding_semantics(self): + for explicit in (False, True): + with ( + self.subTest(explicit=explicit), + tempfile.TemporaryDirectory() as directory, + ): + raw = body_for("systemone").replace(b"original", b"\xff") + path = await self.collect(Path(directory), raw) + record = self.records(path)[0] + record.pop("response_bytes_base64", None) + record["response"] = httpx.Response(200, content=raw).text + record["response_sha256"] = hashlib.sha256( + record["response"].encode() + ).hexdigest() + record["error"] = record["error_kind"] = None + record["answers"] = bench.wire_answers( + CASES[0], json.loads(record["response"]), "english", "systemone" + ) + (path / "responses.jsonl").write_text(json.dumps(record) + "\n") + state = json.loads((path / "completion.json").read_text()) + state["responses_sha256"] = bench.digest(path / "responses.jsonl") + bench.write_json(path / "completion.json", state) + + def legacy_config(config, explicit=explicit): + config.pop("response_format", None) + if explicit: + config["response_format"] = "httpx-text-v1" + + replace_config(path, legacy_config) + before = { + name: (path / name).read_bytes() + for name in ("config.json", "responses.jsonl") + } + summary = bench.summarize_saved(path) + self.assertEqual(summary["successful_requests"], 1) + self.assertEqual(summary["failed_requests"], 0) + self.assert_cli_summary(path, summary) + self.assertEqual( + before, {name: (path / name).read_bytes() for name in before} + ) + + async def test_failed_warmup_retains_the_original_bytes_and_unattempted_denominator( + self, + ): + with tempfile.TemporaryDirectory() as directory: + raw = body_for("systemone").replace(b"original", b"\xff") + path = await self.collect(Path(directory), raw, warmup=1) + record = json.loads((path / "warmup.json").read_text())[0] + self.assertEqual(record["error_kind"], "invalid_response") + self.assert_original_bytes(record, raw) + self.assertFalse((path / "responses.jsonl").exists()) + summary = bench.summarize_saved(path) + self.assertEqual(summary["attempted_requests"], 0) + self.assertEqual(summary["unattempted_requests"], 1) + self.assertFalse(summary["evidence_complete"]) + self.assert_cli_summary(path, summary) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/benchmarks/test_saved_protocols.py b/tests/benchmarks/test_saved_protocols.py index c36b26be..dd994d74 100644 --- a/tests/benchmarks/test_saved_protocols.py +++ b/tests/benchmarks/test_saved_protocols.py @@ -268,7 +268,7 @@ async def test_raw_hash_and_torn_snapshot_checks_apply_to_both_protocols(self): state = json.loads(state_bytes) state["responses_sha256"] = bench.digest(path / "responses.jsonl") bench.write_json(path / "completion.json", state) - with self.assertRaisesRegex(ValueError, "raw response checksum"): + with self.assertRaisesRegex(ValueError, "response text does not match its bytes"): bench.summarize_saved(path) (path / "responses.jsonl").write_bytes(responses + b'{"id":') (path / "completion.json").write_bytes(state_bytes)