refactor to adopt producer -> typed dataclass -> serialized by consumer schema from refactors

Author: angehu-nv

tools/perf_smoke_test/aggregate.py

@@ -29,6 +29,7 @@
     update_baseline,
     update_baselines_git,
 )
+from contracts import BenchResult  # noqa: E402
 from gate_config import BASELINE_PUSH_RETRIES, load_gate_config  # noqa: E402
 from gate_types import FpsMeanThreshold, OracleVerdict  # noqa: E402
 from gpu_identity import canonical_gpu_model  # noqa: E402
@@ -58,11 +59,11 @@ def _parse_args():
     return parser.parse_args()
 
 
-def _find_bench_results(artifacts_dir: Path) -> list[tuple[Path, dict]]:
+def _find_bench_results(artifacts_dir: Path) -> list[tuple[Path, BenchResult]]:
     found = []
     for path in sorted(artifacts_dir.rglob("perf_smoke_test_result.json")):
         with path.open() as fh:
-            found.append((path.parent, json.load(fh)))
+            found.append((path.parent, BenchResult.from_dict(json.load(fh))))
     return found
 
 
@@ -74,20 +75,20 @@ def _short_sha(value: str | None) -> str:
     return value[:12] if value else "none"
 
 
-def _bench_gpu_model(bench_result: dict, fallback: str) -> str:
-    launch_config = bench_result.get("launch_config") or {}
+def _bench_gpu_model(bench_result: BenchResult, fallback: str) -> str:
+    launch_config = bench_result.launch_config or {}
     gpu_model = canonical_gpu_model(launch_config.get("gpu_model") or launch_config.get("gpu_model_raw"))
     return canonical_gpu_model(fallback) if gpu_model == "unknown_gpu" else gpu_model
 
 
-def _thresholds(bench_result: dict, gpu_model: str, backend: str) -> list[FpsMeanThreshold]:
+def _thresholds(bench_result: BenchResult, gpu_model: str, backend: str) -> list[FpsMeanThreshold]:
     """Resolve configured FPS thresholds, preferring the run's launch_config artifact."""
-    launch_config = bench_result.get("launch_config") or {}
+    launch_config = bench_result.launch_config or {}
     raw = launch_config.get("fps_mean_thresholds")
     if raw is not None:
-        return FpsMeanThreshold.from_list(raw, context=f"{bench_result.get('task_id')}/{backend}")
+        return FpsMeanThreshold.from_list(raw, context=f"{bench_result.task_id}/{backend}")
     try:
-        task = get_task(bench_result["task_id"], backend)
+        task = get_task(bench_result.task_id, backend)
         return task.thresholds_for(gpu_model)
     except Exception:
         return []
@@ -109,10 +110,10 @@ def _build_summary_table(rows: list[tuple]) -> str:
         "|---|---|---|---:|---:|---:|---:|---:|---|---|---|---|---|---|",
     ]
     for result, bench_result in rows:
-        gpu_diag = bench_result.get("gpu_diag") or {}
-        launch_config = bench_result.get("launch_config") or {}
+        gpu_diag = bench_result.gpu_diag or {}
+        launch_config = bench_result.launch_config or {}
         gpu_name = gpu_diag.get("gpu_name") or launch_config.get("gpu_model_raw") or launch_config.get("gpu_model", "")
-        provenance = bench_result.get("provenance") or {}
+        provenance = bench_result.provenance or {}
         software = provenance.get("software") or {}
         runtime = ", ".join(
             part
@@ -125,7 +126,7 @@ def _build_summary_table(rows: list[tuple]) -> str:
         )
         # result.note already carries the config_mismatch string on the
         # config-mismatch HARD_FAILURE path; dedupe so it is not shown twice.
-        note_parts = list(dict.fromkeys(part for part in (result.note, bench_result.get("config_mismatch")) if part))
+        note_parts = list(dict.fromkeys(part for part in (result.note, bench_result.config_mismatch) if part))
         note_parts.extend(_render_crossed(result.crossed_thresholds))
         lines.append(
             f"| {result.task_id} | {result.backend} | {result.verdict.value}"
@@ -189,11 +190,12 @@ def main() -> int:
     pending_git_updates: list[BaselineUpdateRecord] = []
 
     for artifact_dir, bench_result in items:
-        task_id = bench_result["task_id"]
-        backend = bench_result.get("backend_key") or bench_result.get("backend")
+        task_id = bench_result.task_id
+        backend = bench_result.backend_key or bench_result.backend
         bench_gpu_model = _bench_gpu_model(bench_result, args.gpu_model)
+        # The baseline_manager storage layer works with the serialized (dict) form.
         match_context = match_context_from_bench_result(
-            bench_result,
+            bench_result.to_dict(),
             gpu_model=bench_gpu_model,
             base_sha=args.base_sha,
             target_branch=args.target_branch,
@@ -248,7 +250,7 @@ def main() -> int:
                 task_id=task_id,
                 backend=backend,
                 fps=oracle_result.measured_fps,
-                bench_result=bench_result,
+                bench_result=bench_result.to_dict(),
                 target_branch=args.target_branch,
                 source_branch=args.source_branch,
                 trusted_source=args.trusted_source,

tools/perf_smoke_test/benchmark_result_adapter.py

@@ -14,12 +14,6 @@
 :mod:`oracle` and :mod:`build_bench_result` consume, replacing the legacy
 phase-array parsing.
 
-**Deliberately dropped vs. the legacy phase-array output** (the bundle stores
-only aggregates, so these have no source and were never gating):
-
-- ``raw_fps_median`` / ``raw_fps_p5`` / ``raw_fps_p95`` (percentiles)
-- ``p99_over_median`` / ``outlier_count`` (raw step-time distribution)
-
 ``raw_fps_min`` is *recovered* from the slowest steady-state step
 (``num_envs / iteration_time_s.peak``); ``raw_fps_{mean,std,max}`` come straight
 from ``runtime.total_fps.{mean,std,peak}``.  Warmup exclusion is applied at the
@@ -34,6 +28,8 @@
 from pathlib import Path
 from typing import Any
 
+from contracts import RuntimeSample
+
 # Map the schema's rendering-backend vocabulary to the gate's render-preset
 # tokens used in tasks.json / backend_identity.  ``"none"`` (headless, no camera)
 # maps to ``None``.
@@ -239,19 +235,31 @@ def benchmark_info(bundle: dict) -> dict:
     return {k: v for k, v in info.items() if v is not None}
 
 
-def to_gate_fields(bundle: dict) -> dict:
-    """Project a runtime bundle into the flat fields the result builder writes.
+def project_runtime(bundle: dict) -> RuntimeSample | None:
+    """Project a runtime bundle into a typed :class:`~contracts.RuntimeSample`.
 
-    Returns a dict with ``raw_fps_*`` stats, ``startup_time_s``, ``gpu_diag``,
-    and ``provenance``; keys are omitted when their source is absent.
+    Returns ``None`` when ``bundle`` is not a valid schema-v1 runtime bundle, so the
+    caller can degrade to a HARD_FAILURE (missing benchmark output).
     """
-    out: dict = {}
-    out.update(fps_stats(bundle))
-    startup = startup_seconds(bundle)
-    if startup is not None:
-        out["startup_time_s"] = startup
-    diag = gpu_diag(bundle)
-    if diag:
-        out["gpu_diag"] = diag
-    out["provenance"] = provenance(bundle)
-    return out
+    if not is_runtime_bundle(bundle):
+        return None
+    stats = fps_stats(bundle)
+    info = benchmark_info(bundle)
+    return RuntimeSample(
+        fps_mean=stats.get("raw_fps_mean"),
+        fps_std=stats.get("raw_fps_std"),
+        fps_min=stats.get("raw_fps_min"),
+        fps_max=stats.get("raw_fps_max"),
+        startup_time_s=startup_seconds(bundle),
+        task=info.get("task"),
+        num_envs=info.get("num_envs"),
+        seed=info.get("seed"),
+        num_frames=info.get("num_frames"),
+        warmup_frames=info.get("warmup_frames"),
+        status=info.get("status"),
+        physics_backend=info.get("physics_backend"),
+        render_backend=info.get("render_backend"),
+        presets=info.get("presets") or [],
+        provenance=provenance(bundle),
+        gpu_diag=gpu_diag(bundle),
+    )

tools/perf_smoke_test/build_bench_result.py

@@ -42,12 +42,8 @@
     normalize_physics_backend,
     normalize_render_backend,
 )
-from benchmark_result_adapter import (  # noqa: E402
-    benchmark_info as bundle_benchmark_info,
-    is_runtime_bundle,
-    load_info,
-    to_gate_fields,
-)
+from benchmark_result_adapter import load_info, project_runtime  # noqa: E402
+from contracts import BenchResult  # noqa: E402
 from gate_config import load_gate_config  # noqa: E402
 from gate_types import FailurePhase  # noqa: E402
 from gpu_identity import normalize_gpu_fields  # noqa: E402
@@ -223,8 +219,8 @@ def main() -> int:
         timeout_s=args.timeout_s,
     )
 
-    # Read the runtime bundle (schema v1) and project it into gate fields.
-    info_provenance: dict = {}
+    # Read the runtime bundle (schema v1) and project it into a typed RuntimeSample.
+    sample = None
     benchmark_info: dict = {}
     config_mismatch: str | None = None
     observed_backend = None
@@ -234,19 +230,19 @@ def main() -> int:
     if perf_smoke_test_info_present:
         info_path = artifact_dir / "perf_smoke_test_info.json"
         bundle = load_info(info_path)
-        if bundle is not None and is_runtime_bundle(bundle):
-            info_provenance = to_gate_fields(bundle)
-            benchmark_info = bundle_benchmark_info(bundle)
+        sample = project_runtime(bundle) if bundle is not None else None
+        if sample is not None:
+            benchmark_info = sample.benchmark_info()
             observed_backend = backend_identity_from_benchmark_info(benchmark_info)
             runtime_contract, runtime_contract_hash = build_runtime_contract(
-                provenance=info_provenance.get("provenance"),
-                gpu_diag=info_provenance.get("gpu_diag"),
+                provenance=sample.provenance,
+                gpu_diag=sample.gpu_diag,
                 backend=expected_backend,
                 policy=runtime_policy,
             )
             runtime_info = build_runtime_publish_info(
-                provenance=info_provenance.get("provenance"),
-                gpu_diag=info_provenance.get("gpu_diag"),
+                provenance=sample.provenance,
+                gpu_diag=sample.gpu_diag,
                 policy=runtime_policy,
             )
             config_mismatch = _config_drift(benchmark_info, launch_config)
@@ -257,38 +253,38 @@ def main() -> int:
     if config_mismatch and failure_phase is None:
         failure_phase = FailurePhase.CONFIG_MISMATCH.value
 
-    bench_result = {
-        "task_id": task_id,
-        "backend": backend_key,
-        "physics_backend": physics_backend,
-        "render_backend": render_backend,
-        "backend_key": backend_key,
-        "preset": preset,
-        "attempt": args.attempt,
-        "was_retried": args.was_retried,
-        "exit_code": args.exit_code,
-        "failure_phase": failure_phase,
-        "stdout_tail": log_text[-2000:] if len(log_text) > 2000 else log_text,
-        "wall_time_s": args.wall_time_s,
-        "startup_time_s": info_provenance.get("startup_time_s"),
-        "perf_smoke_test_info_present": perf_smoke_test_info_present,
-        "raw_fps_mean": info_provenance.get("raw_fps_mean"),
-        "raw_fps_std": info_provenance.get("raw_fps_std"),
-        "raw_fps_min": info_provenance.get("raw_fps_min"),
-        "raw_fps_max": info_provenance.get("raw_fps_max"),
-        "benchmark_info": benchmark_info,
-        "observed_backend": observed_backend.to_dict() if observed_backend else None,
-        "config_mismatch": config_mismatch,
-        "runtime_contract": runtime_contract,
-        "runtime_contract_hash": runtime_contract_hash,
-        "runtime_info": runtime_info,
-        "gpu_diag": info_provenance.get("gpu_diag"),
-        "provenance": info_provenance.get("provenance"),
-        "launch_config": launch_config,
-        "launch_config_hash": launch_config.get("launch_config_hash"),
-        "benchmark_contract_hash": launch_config.get("benchmark_contract_hash"),
-        "baseline_epoch": launch_config.get("baseline_epoch", 1),
-        "task_config_snapshot": {
+    bench_result = BenchResult(
+        task_id=task_id,
+        backend=backend_key,
+        physics_backend=physics_backend,
+        render_backend=render_backend,
+        backend_key=backend_key,
+        preset=preset,
+        attempt=args.attempt,
+        was_retried=args.was_retried,
+        exit_code=args.exit_code,
+        failure_phase=failure_phase,
+        stdout_tail=log_text[-2000:] if len(log_text) > 2000 else log_text,
+        wall_time_s=args.wall_time_s,
+        startup_time_s=sample.startup_time_s if sample else None,
+        perf_smoke_test_info_present=perf_smoke_test_info_present,
+        raw_fps_mean=sample.fps_mean if sample else None,
+        raw_fps_std=sample.fps_std if sample else None,
+        raw_fps_min=sample.fps_min if sample else None,
+        raw_fps_max=sample.fps_max if sample else None,
+        benchmark_info=benchmark_info,
+        observed_backend=observed_backend.to_dict() if observed_backend else None,
+        config_mismatch=config_mismatch,
+        runtime_contract=runtime_contract,
+        runtime_contract_hash=runtime_contract_hash,
+        runtime_info=runtime_info,
+        gpu_diag=(sample.gpu_diag or None) if sample else None,
+        provenance=sample.provenance if sample else None,
+        launch_config=launch_config,
+        launch_config_hash=launch_config.get("launch_config_hash"),
+        benchmark_contract_hash=launch_config.get("benchmark_contract_hash"),
+        baseline_epoch=launch_config.get("baseline_epoch", 1),
+        task_config_snapshot={
             "task_id": task_id,
             "backend": backend_key,
             "physics_backend": physics_backend,
@@ -306,10 +302,10 @@ def main() -> int:
             "runtime_contract_hash": runtime_contract_hash,
             "baseline_epoch": launch_config.get("baseline_epoch", 1),
         },
-    }
+    )
 
     out = artifact_dir / "perf_smoke_test_result.json"
-    out.write_text(json.dumps(bench_result, indent=2))
+    out.write_text(json.dumps(bench_result.to_dict(), indent=2))
 
     status = (
         f"failure_phase={failure_phase!r}, perf_smoke_test_info_present={perf_smoke_test_info_present}, "