| """Validate and recompute a saved Vons browser benchmark report. |
| |
| The browser page is responsible for collecting raw samples. This tool is the |
| second, offline boundary: it rejects malformed samples and recomputes summary |
| statistics instead of trusting the values emitted by the page. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import math |
| import re |
| from pathlib import Path |
| from typing import Any |
|
|
| TIMING_FIELDS = ( |
| "tokenizeMs", |
| "prepareFeedMs", |
| "inferenceAndReadbackMs", |
| "postprocessMs", |
| "totalRequestMs", |
| ) |
| PROVIDERS = {"wasm", "webgpu"} |
| BACKENDS = {"direct", "diffusion"} |
| SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") |
|
|
|
|
| def file_sha256(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def _finite_number(value: Any, label: str, *, allow_none: bool = False) -> float | None: |
| if value is None and allow_none: |
| return None |
| if isinstance(value, bool) or not isinstance(value, (int, float)): |
| raise TypeError(f"{label} must be a finite number or null") |
| numeric = float(value) |
| if not math.isfinite(numeric) or numeric < 0: |
| raise ValueError(f"{label} must be a finite non-negative number") |
| return numeric |
|
|
|
|
| def _positive_integer(value: Any, label: str) -> int: |
| if isinstance(value, bool) or not isinstance(value, int) or value < 1: |
| raise ValueError(f"{label} must be a positive integer") |
| return value |
|
|
|
|
| def _nonnegative_integer(value: Any, label: str) -> int: |
| if isinstance(value, bool) or not isinstance(value, int) or value < 0: |
| raise ValueError(f"{label} must be a non-negative integer") |
| return value |
|
|
|
|
| def _sha256(value: Any, label: str, *, allow_none: bool = False) -> str | None: |
| if value is None and allow_none: |
| return None |
| if not isinstance(value, str) or SHA256_PATTERN.fullmatch(value) is None: |
| raise ValueError(f"{label} must be a lowercase SHA-256 hex string" ) |
| return value |
|
|
|
|
| def percentile(values: list[float], fraction: float) -> float | None: |
| if not values: |
| return None |
| ordered = sorted(values) |
| index = (len(ordered) - 1) * fraction |
| lower = math.floor(index) |
| upper = math.ceil(index) |
| if lower == upper: |
| return ordered[lower] |
| return ordered[lower] + (ordered[upper] - ordered[lower]) * (index - lower) |
|
|
|
|
| def _assert_close(actual: Any, expected: float | None, label: str) -> None: |
| if expected is None: |
| if actual is not None: |
| raise ValueError(f"{label} must be null, got {actual!r}") |
| return |
| if not isinstance(actual, (int, float)) or not math.isclose( |
| float(actual), expected, rel_tol=1e-9, abs_tol=1e-9 |
| ): |
| raise ValueError(f"{label} does not match recomputed value {expected!r}: {actual!r}") |
|
|
|
|
| def verify_report(report_path: Path, *, allow_incomplete: bool = False) -> dict[str, Any]: |
| report = json.loads(report_path.read_text(encoding="utf-8")) |
| if not isinstance(report, dict): |
| raise TypeError("browser benchmark report must be an object") |
| if report.get("schema") != "vons.browser-benchmark/v1": |
| raise ValueError("unexpected browser benchmark schema") |
|
|
| sessions = _positive_integer(report.get("sessions"), "sessions") |
| warmup = _positive_integer(report.get("warmup_excluded_per_session"), "warmup_excluded_per_session") |
| repeats = _positive_integer(report.get("repeats_per_session"), "repeats_per_session") |
| _positive_integer(report.get("cases"), "cases") |
| if not isinstance(report.get("run_id"), str) or not report["run_id"]: |
| raise TypeError("run_id is required") |
| backend = report.get("backend") |
| if backend not in BACKENDS: |
| raise ValueError("backend must be direct or diffusion") |
| provider = report.get("requested_provider") |
| if provider not in PROVIDERS: |
| raise ValueError("requested_provider must be wasm or webgpu") |
| environment = report.get("environment") |
| if ( |
| not isinstance(environment, dict) |
| or not isinstance(environment.get("user_agent"), str) |
| or not environment["user_agent"] |
| ): |
| raise TypeError("environment.user_agent is required") |
| if not isinstance(environment.get("cross_origin_isolated"), bool): |
| raise TypeError("environment.cross_origin_isolated must be boolean") |
| runtime_info = report.get("runtime_info") |
| if runtime_info is not None and not isinstance(runtime_info, dict): |
| raise TypeError("runtime_info must be an object or null") |
| if not isinstance(report.get("execution_provider_evidence"), str) or not report["execution_provider_evidence"]: |
| raise TypeError("execution_provider_evidence is required") |
| samples = report.get("samples") |
| session_records = report.get("session_records") |
| if not isinstance(samples, list) or not isinstance(session_records, list): |
| raise TypeError("samples and session_records must be lists") |
|
|
| seen_slots: set[tuple[int, int]] = set() |
| successful: list[dict[str, Any]] = [] |
| failures = 0 |
| observed_providers: set[str] = set() |
| for index, sample in enumerate(samples, start=1): |
| if not isinstance(sample, dict): |
| raise TypeError(f"sample {index} must be an object") |
| session_start = sample.get("session_start") |
| repeat = sample.get("repeat") |
| if ( |
| isinstance(session_start, bool) |
| or not isinstance(session_start, int) |
| or not 1 <= session_start <= sessions |
| ): |
| raise ValueError(f"sample {index} has an invalid session_start") |
| if isinstance(repeat, bool) or not isinstance(repeat, int) or not 0 <= repeat < repeats: |
| raise ValueError(f"sample {index} has an invalid repeat") |
| slot = (session_start, repeat) |
| if slot in seen_slots: |
| raise ValueError(f"duplicate sample slot {slot}") |
| seen_slots.add(slot) |
| if sample.get("requested_provider") != provider: |
| raise ValueError(f"sample {index} requested_provider disagrees with report") |
| if sample.get("backend") != backend: |
| raise ValueError(f"sample {index} backend disagrees with report") |
| if not isinstance(sample.get("run_id"), str) or not sample["run_id"]: |
| raise TypeError(f"sample {index}.run_id is required") |
| if not isinstance(sample.get("case_id"), str) or not sample["case_id"]: |
| raise TypeError(f"sample {index}.case_id is required") |
| _sha256(sample.get("bundle_hash"), f"sample {index}.bundle_hash") |
| _sha256(sample.get("tokenizer_hash"), f"sample {index}.tokenizer_hash") |
| _sha256(sample.get("input_hash"), f"sample {index}.input_hash") |
| noise_hash = _sha256(sample.get("noise_hash"), f"sample {index}.noise_hash", allow_none=True) |
| if backend == "diffusion" and noise_hash is None: |
| raise ValueError(f"sample {index}.noise_hash is required for diffusion") |
| if backend == "direct" and noise_hash is not None: |
| raise ValueError(f"sample {index}.noise_hash must be null for direct") |
| observed = sample.get("observed_provider") |
| if observed is not None: |
| if observed not in PROVIDERS: |
| raise ValueError(f"sample {index} has an invalid observed_provider") |
| observed_providers.add(observed) |
| status = sample.get("status") |
| if status not in {"ok", "error"}: |
| raise ValueError(f"sample {index} status must be ok or error") |
| if not isinstance(sample.get("observed_provider_status"), str) or not sample["observed_provider_status"]: |
| raise ValueError(f"sample {index} requires observed_provider_status") |
| for field in TIMING_FIELDS: |
| timing = _finite_number(sample.get(field), f"sample {index}.{field}", allow_none=True) |
| if status == "ok" and timing is None: |
| raise ValueError(f"successful sample {index}.{field} must be numeric") |
| if status == "error" and timing is not None: |
| raise ValueError(f"failed sample {index}.{field} must be null") |
| shape_fields = ("live_candidates", "allocated_candidates", "live_tokens", "sequence_length") |
| for field in shape_fields: |
| value = sample.get(field) |
| if status == "ok": |
| _nonnegative_integer(value, f"sample {index}.{field}") |
| elif value is not None: |
| raise ValueError(f"failed sample {index}.{field} must be null") |
| if status == "ok" and sample["allocated_candidates"] < sample["live_candidates"]: |
| raise ValueError(f"sample {index} allocated_candidates is below live_candidates") |
| for field in ("js_heap_used_bytes_before", "js_heap_used_bytes_after"): |
| _finite_number(sample.get(field), f"sample {index}.{field}", allow_none=True) |
| if status == "ok": |
| successful.append(sample) |
| else: |
| failures += 1 |
|
|
| session_record_ids: set[int] = set() |
| session_failures: list[int] = [] |
| for index, record in enumerate(session_records, start=1): |
| if not isinstance(record, dict): |
| raise TypeError(f"session record {index} must be an object") |
| session_start = record.get("session_start") |
| if ( |
| isinstance(session_start, bool) |
| or not isinstance(session_start, int) |
| or not 1 <= session_start <= sessions |
| ): |
| raise ValueError(f"session record {index} has an invalid session_start") |
| if session_start in session_record_ids: |
| raise ValueError(f"duplicate session record {session_start}") |
| session_record_ids.add(session_start) |
| _finite_number(record.get("load_ms"), f"session record {index}.load_ms") |
| if record.get("status") not in {"ok", "error"}: |
| raise ValueError(f"session record {index} status must be ok or error") |
| if record["status"] == "error": |
| session_failures.append(session_start) |
|
|
| incomplete_reasons: list[str] = [] |
| if len(session_records) != sessions: |
| incomplete_reasons.append(f"expected {sessions} session records, got {len(session_records)}") |
| if session_failures: |
| incomplete_reasons.append(f"session creation failed for starts {session_failures}") |
| expected_samples = sessions * repeats |
| if len(samples) != expected_samples: |
| incomplete_reasons.append(f"expected {expected_samples} timed samples, got {len(samples)}") |
| if incomplete_reasons and not allow_incomplete: |
| raise ValueError("incomplete browser benchmark: " + "; ".join(incomplete_reasons)) |
|
|
| total_values = [float(sample["totalRequestMs"]) for sample in successful] |
| inference_values = [float(sample["inferenceAndReadbackMs"]) for sample in successful] |
| before_values = [ |
| float(sample["js_heap_used_bytes_before"]) |
| for sample in successful |
| if sample.get("js_heap_used_bytes_before") is not None |
| ] |
| after_values = [ |
| float(sample["js_heap_used_bytes_after"]) |
| for sample in successful |
| if sample.get("js_heap_used_bytes_after") is not None |
| ] |
| recomputed_summary = { |
| "successful_samples": len(successful), |
| "failed_samples": failures, |
| "total_request_ms_p50": percentile(total_values, 0.5), |
| "total_request_ms_p95": percentile(total_values, 0.95), |
| "inference_and_readback_ms_p50": percentile(inference_values, 0.5), |
| "inference_and_readback_ms_p95": percentile(inference_values, 0.95), |
| "js_heap_before_max": max(before_values) if before_values else None, |
| "js_heap_after_max": max(after_values) if after_values else None, |
| } |
| supplied_summary = report.get("summary") |
| if supplied_summary is not None: |
| if not isinstance(supplied_summary, dict): |
| raise TypeError("summary must be an object") |
| for key, expected in recomputed_summary.items(): |
| _assert_close(supplied_summary.get(key), expected, f"summary.{key}") |
|
|
| return { |
| "schema": "vons.browser-benchmark-verification/v1", |
| "measurement_label": "verified_browser_benchmark_raw_samples", |
| "complete": not incomplete_reasons, |
| "incomplete_reasons": incomplete_reasons, |
| "report": str(report_path), |
| "report_sha256": file_sha256(report_path), |
| "requested_provider": provider, |
| "observed_providers": sorted(observed_providers), |
| "sessions": sessions, |
| "warmup_excluded_per_session": warmup, |
| "repeats_per_session": repeats, |
| "sample_rows": len(samples), |
| "successful_samples": len(successful), |
| "failed_samples": failures, |
| "recomputed_summary": recomputed_summary, |
| "memory_note": "JS heap values are before/after observations only; null means unavailable; no GPU peak is inferred.", |
| } |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--report", type=Path, required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| parser.add_argument( |
| "--allow-incomplete", |
| action="store_true", |
| help="validate and summarize a partial diagnostic run without treating it as complete", |
| ) |
| args = parser.parse_args() |
| result = verify_report(args.report, allow_incomplete=args.allow_incomplete) |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
| print(json.dumps(result, indent=2, sort_keys=True)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
