"""Summarize a JSONL stream of Vons browser benchmark samples. Inputs are one raw browser sample per JSONL line. Each line must carry the full per-sample schema plus a provenance record that describes the session configuration. The tool performs strict validation before producing any summary: identity duplicates, malformed types, non-finite timings, mixed experimental conditions, and diffusion rows without noise hashes all cause the tool to fail before the output path is touched. Successful rows contribute to linearly interpolated p50/p95 percentiles per timing stage. Failed rows are counted explicitly and retain nullable timing slots. Heap maxima are derived only from actually-measured observations. Measurement completeness (>= 5 independent session starts, >= 10 warmups excluded per session, >= 100 timed rows total) is reported as a separate flag without elevating partial runs to "complete". Because the page harness performs in-page reloads only, "process cold" is never claimed, and the requested provider is never represented as proven. """ from __future__ import annotations import argparse import hashlib import json import math import re from collections.abc import Iterator from pathlib import Path from typing import Any TIMING_FIELDS: tuple[str, ...] = ( "tokenizeMs", "prepareFeedMs", "inferenceAndReadbackMs", "postprocessMs", "totalRequestMs", ) HEAP_FIELDS: tuple[str, ...] = ( "js_heap_used_bytes_before", "js_heap_used_bytes_after", ) PROVIDERS = {"wasm", "webgpu"} BACKENDS = {"direct", "diffusion"} SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") REQUIRED_IDENTITY_FIELDS: tuple[str, ...] = ("run_id", "session_start", "repeat") REQUIRED_TOP_FIELDS: tuple[str, ...] = ( "run_id", "session_start", "repeat", "case_id", "backend", "requested_provider", "observed_provider", "bundle_hash", "tokenizer_hash", "input_hash", "candidate_count", "padded_option_slots", "sequence_length", "seed", "ort_version", "status", ) REQUIRED_PROVENANCE_FIELDS: tuple[str, ...] = ( "environment", "execution_provider_evidence", "warmup_excluded_per_session", "session_start_semantics", ) CONDITION_KEYS: tuple[str, ...] = ( "run_id", "backend", "requested_provider", "observed_provider", "bundle_hash", "tokenizer_hash", "candidate_count", "padded_option_slots", "sequence_length", "ort_version", ) MAX_PADDED_OPTION_SLOTS = 32 MAX_SEQUENCE_LENGTH = 512 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 _nonempty_string(value: Any, label: str) -> str: if not isinstance(value, str) or not value: raise ValueError(f"{label} must be a non-empty string") return value def percentile(values: list[float], fraction: float) -> float | None: if not values: return None if not 0.0 <= fraction <= 1.0: raise ValueError("percentile fraction must be within [0, 1]") ordered = sorted(values) index = (len(ordered) - 1) * fraction lower = math.floor(index) upper = math.ceil(index) if lower == upper: return float(ordered[lower]) return float( ordered[lower] + (ordered[upper] - ordered[lower]) * (index - lower) ) def _provenance_condition_signature(provenance: dict[str, Any]) -> dict[str, Any]: environment = provenance["environment"] evidence = provenance["execution_provider_evidence"] warmup = provenance["warmup_excluded_per_session"] semantics = provenance["session_start_semantics"] return { "environment": environment, "execution_provider_evidence": evidence, "warmup_excluded_per_session": warmup, "session_start_semantics": semantics, } def _validate_row(row: dict[str, Any], line_no: int) -> None: if not isinstance(row, dict): raise TypeError(f"line {line_no}: sample must be a JSON object") provenance = row.get("provenance") if not isinstance(provenance, dict): raise TypeError(f"line {line_no}: provenance object is required") for field in REQUIRED_TOP_FIELDS: if field not in row: raise ValueError(f"line {line_no}: missing required field {field}") for field in REQUIRED_PROVENANCE_FIELDS: if field not in provenance: raise ValueError( f"line {line_no}: provenance missing required field {field}" ) _nonempty_string(row["run_id"], f"line {line_no}.run_id") _positive_integer(row["session_start"], f"line {line_no}.session_start") if isinstance(row["repeat"], bool) or not isinstance(row["repeat"], int) or row["repeat"] < 0: raise ValueError(f"line {line_no}.repeat must be a non-negative integer") _nonempty_string(row["case_id"], f"line {line_no}.case_id") if row["backend"] not in BACKENDS: raise ValueError(f"line {line_no}.backend must be direct or diffusion") if row["requested_provider"] not in PROVIDERS: raise ValueError( f"line {line_no}.requested_provider must be wasm or webgpu" ) observed = row["observed_provider"] if observed is not None and observed not in PROVIDERS: raise ValueError( f"line {line_no}.observed_provider must be wasm, webgpu, or null" ) _sha256(row["bundle_hash"], f"line {line_no}.bundle_hash") _sha256(row["tokenizer_hash"], f"line {line_no}.tokenizer_hash") _sha256(row["input_hash"], f"line {line_no}.input_hash") noise_hash = _sha256( row.get("noise_hash"), f"line {line_no}.noise_hash", allow_none=True ) if row["backend"] == "diffusion" and noise_hash is None: raise ValueError( f"line {line_no}.noise_hash is required for diffusion backend" ) _positive_integer(row["candidate_count"], f"line {line_no}.candidate_count") _positive_integer( row["padded_option_slots"], f"line {line_no}.padded_option_slots" ) _positive_integer(row["sequence_length"], f"line {line_no}.sequence_length") if "seed" not in row: raise ValueError(f"line {line_no}: missing required field seed") seed_value = row["seed"] if seed_value is None: if row["backend"] != "direct" or noise_hash is not None: raise ValueError( f"line {line_no}.seed may only be null for direct backend with " "noise_hash=null (deterministic-direct; seed is not_applicable); " "diffusion or direct-with-noise require a non-negative integer seed" ) else: _nonnegative_integer(seed_value, f"line {line_no}.seed") if row["candidate_count"] > row["padded_option_slots"]: raise ValueError( f"line {line_no}: candidate_count must be <= padded_option_slots" ) if row["padded_option_slots"] > MAX_PADDED_OPTION_SLOTS: raise ValueError( f"line {line_no}: padded_option_slots must be <= {MAX_PADDED_OPTION_SLOTS}" ) if row["sequence_length"] > MAX_SEQUENCE_LENGTH: raise ValueError( f"line {line_no}: sequence_length must be <= {MAX_SEQUENCE_LENGTH}" ) _nonempty_string(row["ort_version"], f"line {line_no}.ort_version") status = row["status"] if status not in {"ok", "error", "timeout"}: raise ValueError( f"line {line_no}.status must be ok, error, or timeout" ) error = row.get("error") if status in {"error", "timeout"}: if error is None: pass elif not isinstance(error, str) or not error: raise ValueError( f"line {line_no}.error must be a non-empty string for status {status}" ) elif status == "ok" and error is not None: raise ValueError( f"line {line_no}.error must be null for successful samples" ) for field in TIMING_FIELDS: timing = _finite_number( row.get(field), f"line {line_no}.{field}", allow_none=True ) if status == "ok" and timing is None: raise ValueError( f"line {line_no}: successful sample requires {field}" ) for field in HEAP_FIELDS: _finite_number(row.get(field), f"line {line_no}.{field}", allow_none=True) environment = provenance["environment"] if not isinstance(environment, dict): raise TypeError(f"line {line_no}.provenance.environment must be an object") _nonempty_string( provenance["execution_provider_evidence"], f"line {line_no}.provenance.execution_provider_evidence", ) _positive_integer( provenance["warmup_excluded_per_session"], f"line {line_no}.provenance.warmup_excluded_per_session", ) _nonempty_string( provenance["session_start_semantics"], f"line {line_no}.provenance.session_start_semantics", ) def _iter_jsonl(path: Path) -> Iterator[tuple[int, dict[str, Any]]]: with path.open("r", encoding="utf-8") as handle: line_no = 0 for raw in handle: line_no += 1 stripped = raw.strip() if not stripped: continue yield line_no, json.loads(stripped) def _condition_tuple(row: dict[str, Any]) -> tuple[Any, ...]: provenance_sig = _provenance_condition_signature(row["provenance"]) return ( tuple(row[key] for key in CONDITION_KEYS), provenance_sig["warmup_excluded_per_session"], provenance_sig["session_start_semantics"], json.dumps(provenance_sig["environment"], sort_keys=True), provenance_sig["execution_provider_evidence"], ) def validate_and_aggregate(input_path: Path) -> dict[str, Any]: if not input_path.exists(): raise FileNotFoundError(f"input file not found: {input_path}") rows: list[dict[str, Any]] = [] seen_identities: set[tuple[str, int, int]] = set() first_condition: tuple[Any, ...] | None = None provenance_reference: dict[str, Any] | None = None session_samples: dict[int, int] = {} success_count = 0 failure_count = 0 total_rows = 0 stage_timings: dict[str, list[float]] = {field: [] for field in TIMING_FIELDS} heap_values: dict[str, list[float]] = {field: [] for field in HEAP_FIELDS} for line_no, row in _iter_jsonl(input_path): _validate_row(row, line_no) total_rows += 1 identity = (row["run_id"], row["session_start"], row["repeat"]) if identity in seen_identities: raise ValueError( f"duplicate (run_id, session_start, repeat) identity at line {line_no}: {identity}" ) seen_identities.add(identity) condition = _condition_tuple(row) if first_condition is None: first_condition = condition provenance_reference = _provenance_condition_signature( row["provenance"] ) elif condition != first_condition: raise ValueError( f"line {line_no}: mixed experimental conditions detected; " "bundle, tokenizer, shape, provider, runtime, environment, " "and config provenance fields must be homogeneous across all rows" ) session_start = row["session_start"] session_samples[session_start] = session_samples.get(session_start, 0) + 1 status = row["status"] if status == "ok": success_count += 1 for field in TIMING_FIELDS: stage_timings[field].append(float(row[field])) for field in HEAP_FIELDS: value = row.get(field) if value is not None: heap_values[field].append(float(value)) else: failure_count += 1 rows.append(row) if total_rows == 0: raise ValueError("input JSONL contains no samples") return _build_summary( rows=rows, input_path=input_path, provenance_reference=provenance_reference, first_condition=first_condition, session_samples=session_samples, success_count=success_count, failure_count=failure_count, total_rows=total_rows, stage_timings=stage_timings, heap_values=heap_values, ) def _build_summary( *, rows: list[dict[str, Any]], input_path: Path, provenance_reference: dict[str, Any] | None, first_condition: tuple[Any, ...] | None, session_samples: dict[int, int], success_count: int, failure_count: int, total_rows: int, stage_timings: dict[str, list[float]], heap_values: dict[str, list[float]], ) -> dict[str, Any]: assert provenance_reference is not None assert first_condition is not None assert rows first_row = rows[0] unique_sessions = len(session_samples) warmup_per_session = provenance_reference["warmup_excluded_per_session"] timed_requests_total = success_count + failure_count complete = ( unique_sessions >= 5 and warmup_per_session >= 10 and timed_requests_total >= 100 ) incomplete_reasons: list[str] = [] if unique_sessions < 5: incomplete_reasons.append( f"only {unique_sessions} independent session starts observed (>= 5 required)" ) if warmup_per_session < 10: incomplete_reasons.append( f"warmup_excluded_per_session={warmup_per_session} (>= 10 required)" ) if timed_requests_total < 100: incomplete_reasons.append( f"only {timed_requests_total} timed requests total (>= 100 required)" ) process_cold_claimable = False requested_provider = first_row["requested_provider"] observed_provider = first_row["observed_provider"] requested_is_proven = False summary_timings: dict[str, dict[str, float | None]] = {} for field in TIMING_FIELDS: values = stage_timings[field] summary_timings[field] = { "p50": percentile(values, 0.5), "p95": percentile(values, 0.95), } heap_maxima: dict[str, float | None] = {} for field in HEAP_FIELDS: values = heap_values[field] heap_maxima[field] = max(values) if values else None per_session_samples: dict[str, int] = { str(session_start): count for session_start, count in sorted(session_samples.items()) } summary: dict[str, Any] = { "schema": "vons.browser-benchmark-summary/v1", "generated_from": str(input_path), "input_sha256": _file_sha256(input_path), "experimental_condition": { "run_id": first_row["run_id"], "backend": first_row["backend"], "requested_provider": requested_provider, "observed_provider": observed_provider, "bundle_hash": first_row["bundle_hash"], "tokenizer_hash": first_row["tokenizer_hash"], "candidate_count": first_row["candidate_count"], "padded_option_slots": first_row["padded_option_slots"], "sequence_length": first_row["sequence_length"], "seed": first_row["seed"], "ort_version": first_row["ort_version"], "provenance": provenance_reference, }, "provider_claim_note": ( "requested_provider is not represented as proven in this v1 summary. " "observed_provider preserves the browser-reported value and " "execution_provider_evidence preserves the raw evidence string, " "but no structured verified execution artifact exists to elevate " "a match to a proven claim; requested_is_proven is always False in v1." ), "process_cold_note": ( "process-level cold start identity is not verifiable from page-only " "browser observations in this v1 schema; session_start_semantics is " "preserved as a raw label but process_cold_claimable is always False " "regardless of semantics string content." ), "sample_counts": { "total_rows": total_rows, "successful": success_count, "failed": failure_count, "unique_session_starts": unique_sessions, "per_session_samples": per_session_samples, }, "stage_percentiles_ms": summary_timings, "heap_maxima_bytes": heap_maxima, "measurement_completeness": { "complete": complete, "required_min_session_starts": 5, "required_min_warmup_excluded_per_session": 10, "required_min_timed_requests_total": 100, "actual_session_starts": unique_sessions, "actual_warmup_excluded_per_session": warmup_per_session, "actual_timed_requests_total": timed_requests_total, "incomplete_reasons": incomplete_reasons, }, "process_cold_claimable": bool(process_cold_claimable), "requested_is_proven": bool(requested_is_proven), } return summary def _iter_report_samples( report: dict[str, Any], report_path: Path ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """Return (top_provenance, normalized_rows) from a v1 browser-benchmark report. Normalizes current report field names to the JSONL schema used by the validator: live_candidates -> candidate_count, allocated_candidates -> padded_option_slots, sequence_length preferred with sequence_tokens as fallback alias. Never invents missing values. Raises ValueError on any required field that cannot be resolved. """ if not isinstance(report, dict): raise TypeError("browser benchmark report must be a JSON object") if report.get("schema") != "vons.browser-benchmark/v1": raise ValueError("unexpected browser benchmark schema") _nonempty_string(report.get("run_id"), f"{report_path.name}.run_id") backend = report.get("backend") if backend not in BACKENDS: raise ValueError(f"{report_path.name}.backend must be direct or diffusion") requested_provider = report.get("requested_provider") if requested_provider not in PROVIDERS: raise ValueError( f"{report_path.name}.requested_provider must be wasm or webgpu" ) environment = report.get("environment") if not isinstance(environment, dict): raise TypeError( f"{report_path.name}.environment is required as provenance input" ) execution_provider_evidence = _nonempty_string( report.get("execution_provider_evidence"), f"{report_path.name}.execution_provider_evidence", ) warmup_excluded_per_session = _positive_integer( report.get("warmup_excluded_per_session"), f"{report_path.name}.warmup_excluded_per_session", ) session_start_semantics_raw = report.get("session_start_semantics") if session_start_semantics_raw is None: raise ValueError( f"{report_path.name}.session_start_semantics is required to build " "per-row provenance; do not invent missing values." ) session_start_semantics = _nonempty_string( session_start_semantics_raw, f"{report_path.name}.session_start_semantics", ) top_provenance = { "environment": environment, "execution_provider_evidence": execution_provider_evidence, "warmup_excluded_per_session": warmup_excluded_per_session, "session_start_semantics": session_start_semantics, } samples = report.get("samples") if not isinstance(samples, list): raise TypeError(f"{report_path.name}.samples must be a list") if not samples: raise ValueError(f"{report_path.name}.samples contains no rows") ort_version = report.get("ort_version") ort_version_is_str = isinstance(ort_version, str) and bool(ort_version) if not ort_version_is_str: runtime_info = report.get("runtime_info") if isinstance(runtime_info, dict): rt_ort = runtime_info.get("ortVersion") if isinstance(rt_ort, str) and rt_ort: ort_version = rt_ort ort_version_is_str = True normalized: list[dict[str, Any]] = [] for idx, sample in enumerate(samples, start=1): if not isinstance(sample, dict): raise TypeError(f"{report_path.name} sample {idx} must be an object") live_candidates = sample.get("live_candidates") if live_candidates is None: raise ValueError( f"{report_path.name} sample {idx} is missing live_candidates " "(source for candidate_count normalization)" ) allocated_candidates = sample.get("allocated_candidates") if allocated_candidates is None: raise ValueError( f"{report_path.name} sample {idx} is missing allocated_candidates " "(source for padded_option_slots normalization)" ) sequence_length = sample.get("sequence_length") if sequence_length is None: sequence_length = sample.get("sequence_tokens") if sequence_length is None: raise ValueError( f"{report_path.name} sample {idx} is missing sequence_length " "(sequence_tokens not present as alias fallback)" ) sample_ort_version = sample.get("ort_version") if not (isinstance(sample_ort_version, str) and sample_ort_version): if ort_version_is_str: sample_ort_version = ort_version else: raise ValueError( f"{report_path.name} sample {idx} is missing ort_version and " "report does not provide a top-level ort_version fallback " "(checked both ort_version and runtime_info.ortVersion)" ) sample_backend = sample.get("backend") noise_hash = sample.get("noise_hash") sample_seed = sample.get("seed") if sample_seed is None and sample_backend == "direct" and noise_hash is None: resolved_seed = None else: if "seed" not in sample: raise ValueError( f"{report_path.name} sample {idx} is missing seed; " "null seed is only permitted for direct backend with " "noise_hash=null (deterministic-direct, not_applicable)" ) resolved_seed = sample_seed error_value = sample.get("error") observed_provider = sample.get("observed_provider") mapped: dict[str, Any] = { "run_id": sample["run_id"], "session_start": sample["session_start"], "repeat": sample["repeat"], "case_id": sample["case_id"], "backend": sample_backend, "requested_provider": sample["requested_provider"], "observed_provider": observed_provider, "bundle_hash": sample["bundle_hash"], "tokenizer_hash": sample["tokenizer_hash"], "input_hash": sample["input_hash"], "noise_hash": noise_hash, "candidate_count": live_candidates, "padded_option_slots": allocated_candidates, "sequence_length": sequence_length, "seed": resolved_seed, "ort_version": sample_ort_version, "status": sample["status"], "error": error_value, "provenance": top_provenance, } for field in TIMING_FIELDS: mapped[field] = sample.get(field) for field in HEAP_FIELDS: if field in sample: mapped[field] = sample.get(field) normalized.append(mapped) return top_provenance, normalized def validate_and_aggregate_report(report_path: Path) -> dict[str, Any]: """Entry point for --report-input: loads an existing v1 browser-benchmark JSON report, normalizes its samples to the JSONL schema, then runs the same validation and aggregation used by validate_and_aggregate. """ if not report_path.exists(): raise FileNotFoundError(f"report file not found: {report_path}") report = json.loads(report_path.read_text(encoding="utf-8")) provenance_reference, normalized_rows = _iter_report_samples(report, report_path) total_rows = 0 seen_identities: set[tuple[str, int, int]] = set() first_condition: tuple[Any, ...] | None = None session_samples: dict[int, int] = {} success_count = 0 failure_count = 0 stage_timings: dict[str, list[float]] = {field: [] for field in TIMING_FIELDS} heap_values: dict[str, list[float]] = {field: [] for field in HEAP_FIELDS} accepted_rows: list[dict[str, Any]] = [] for line_no, row in enumerate(normalized_rows, start=1): _validate_row(row, line_no) total_rows += 1 identity = (row["run_id"], row["session_start"], row["repeat"]) if identity in seen_identities: raise ValueError( f"duplicate (run_id, session_start, repeat) identity in sample {line_no}: {identity}" ) seen_identities.add(identity) condition = _condition_tuple(row) if first_condition is None: first_condition = condition elif condition != first_condition: raise ValueError( f"sample {line_no}: mixed experimental conditions detected in report" ) session_start = row["session_start"] session_samples[session_start] = session_samples.get(session_start, 0) + 1 if row["status"] == "ok": success_count += 1 for field in TIMING_FIELDS: stage_timings[field].append(float(row[field])) for field in HEAP_FIELDS: value = row.get(field) if value is not None: heap_values[field].append(float(value)) else: failure_count += 1 accepted_rows.append(row) if total_rows == 0: raise ValueError("report samples yielded no rows after normalization") return _build_summary( rows=accepted_rows, input_path=report_path, provenance_reference=provenance_reference, first_condition=first_condition, session_samples=session_samples, success_count=success_count, failure_count=failure_count, total_rows=total_rows, stage_timings=stage_timings, heap_values=heap_values, ) 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 main() -> None: parser = argparse.ArgumentParser(description=__doc__) group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--input", type=Path, help="JSONL file of raw browser benchmark samples") group.add_argument("--report-input", type=Path, help="Existing vons.browser-benchmark/v1 JSON report; " "normalizes live_candidates/allocated_candidates/" "sequence_tokens to JSONL schema and carries top-level provenance") parser.add_argument("--output", type=Path, required=True, help="destination path for the summary JSON") args = parser.parse_args() if args.input is not None: summary = validate_and_aggregate(args.input) else: summary = validate_and_aggregate_report(args.report_input) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) if __name__ == "__main__": main()