feat(laboratory): publish repeated M48S load envelope

src/k1link/web/m48s_fixed_class_detector_lab_api.py

@@ -305,6 +305,7 @@ def _load_result_uncached(candidate: Path) -> dict[str, Any]:
     decision = manifest.get("decision")
     method = manifest.get("method")
     metrics = manifest.get("metrics")
+    load_envelope = metrics.get("load_envelope") if isinstance(metrics, dict) else None
     status = manifest.get("status")
     integrated = status == INTEGRATED_STATUS
     if (
@@ -332,13 +333,18 @@ def _load_result_uncached(candidate: Path) -> dict[str, Any]:
         or (integrated and decision.get("integrated_world_state_gate_passed") is not True)
         or (integrated and decision.get("detector_replacement_authorized") is not False)
         or (
-            metrics.get("load_envelope") is not None
+            load_envelope is not None
             and (
                 decision.get("load_envelope_evaluated") is not True
-                or decision.get("production_rate_repeatability_passed") is not False
-                or decision.get("reserve_12_fps_passed") is not True
-                or decision.get("limit_15_fps_passed") is not True
-                or decision.get("load_envelope_accepted") is not False
+                or not isinstance(load_envelope, dict)
+                or decision.get("production_rate_repeatability_passed")
+                is not load_envelope.get("production_rate_repeatability_passed")
+                or decision.get("reserve_12_fps_passed")
+                is not load_envelope.get("reserve_12_fps_passed")
+                or decision.get("limit_15_fps_passed")
+                is not load_envelope.get("limit_15_fps_passed")
+                or decision.get("load_envelope_accepted")
+                is not load_envelope.get("load_envelope_accepted")
             )
         )
         or decision.get("production_accepted") is not False
@@ -435,10 +441,104 @@ def _valid_runtime_hardening(value: object) -> bool:
 
 
 def _valid_load_envelope(value: object) -> bool:
-    if not isinstance(value, dict) or value.get("schema_version") != (
-        "missioncore.m48s-load-envelope-comparison/v1"
+    if not isinstance(value, dict):
+        return False
+    if value.get("schema_version") == "missioncore.m48s-load-envelope-comparison/v1":
+        return _valid_legacy_load_envelope(value)
+    if value.get("schema_version") != "missioncore.m48s-load-envelope-comparison/v2":
+        return False
+    if (
+        value.get("repeat_count_per_rate") != 3
+        or value.get("production_rate_repeatability_passed") is not True
+        or value.get("reserve_12_fps_passed") is not True
+        or value.get("limit_15_fps_passed") is not True
+        or value.get("load_envelope_accepted") is not True
+        or value.get("compute_capacity_at_least_fps") != 15.0
+        or value.get("bottleneck_interpretation")
+        != "cold-video-decode-isolated-before-admission-no-steady-gpu-saturation"
     ):
         return False
+    scenarios = value.get("scenarios")
+    scenario_ids = [item.get("id") for item in scenarios if isinstance(item, dict)] if (
+        isinstance(scenarios, list)
+    ) else []
+    if scenario_ids != ["production-10fps", "reserve-12fps", "limit-15fps"]:
+        return False
+    required_numeric = {
+        "requested_source_rate_hz",
+        "source_frames_admitted",
+        "delivered_world_states",
+        "superseded_frames",
+        "delivery_ratio",
+        "effective_world_state_fps",
+        "world_state_completion_age_p95_ms",
+        "world_state_completion_age_p99_ms",
+        "world_state_completion_age_maximum_ms",
+        "hot_loop_decode_p95_ms",
+        "hot_loop_decode_maximum_ms",
+        "preadmission_decode_maximum_ms",
+        "source_prefetch_preparation_maximum_ms",
+        "source_pacing_lateness_p95_ms",
+        "source_pacing_lateness_maximum_ms",
+        "detector_p95_ms",
+        "detector_maximum_ms",
+        "gpu_utilization_mean_percent",
+        "gpu_utilization_maximum_percent",
+        "gpu_memory_maximum_mib",
+        "process_peak_rss_mib",
+        "queue_capacity",
+        "additional_inference_passes",
+    }
+    for scenario in scenarios:
+        repetitions = scenario.get("repetitions") if isinstance(scenario, dict) else None
+        if (
+            not isinstance(scenario, dict)
+            or scenario.get("repeat_count") != 3
+            or scenario.get("all_repetitions_passed") is not True
+            or scenario.get("integrity_gate_passed") is not True
+            or scenario.get("operating_target_gate_passed") is not True
+            or scenario.get("additional_inference_passes") != 0
+            or scenario.get("queue_capacity") != 2
+            or not isinstance(scenario.get("queue_high_watermarks"), dict)
+            or any(value != 2 for value in scenario["queue_high_watermarks"].values())
+            or not isinstance(scenario.get("thresholds"), dict)
+            or any(not _nonnegative_number(scenario.get(key)) for key in required_numeric)
+            or not isinstance(repetitions, list)
+            or len(repetitions) != 3
+        ):
+            return False
+        for expected_repetition, repetition in enumerate(repetitions, start=1):
+            if (
+                not isinstance(repetition, dict)
+                or repetition.get("repetition") != expected_repetition
+                or repetition.get("operating_target_gate_passed") is not True
+                or not isinstance(repetition.get("run_id"), str)
+                or not repetition["run_id"]
+                or any(
+                    not _nonnegative_number(repetition.get(key))
+                    for key in {
+                        "delivered_world_states",
+                        "superseded_frames",
+                        "delivery_ratio",
+                        "effective_world_state_fps",
+                        "world_state_completion_age_p95_ms",
+                        "hot_loop_decode_p95_ms",
+                        "hot_loop_decode_maximum_ms",
+                        "preadmission_decode_maximum_ms",
+                        "source_pacing_lateness_maximum_ms",
+                    }
+                )
+                or any(
+                    not isinstance(repetition.get(key), str)
+                    or SHA256.fullmatch(repetition[key]) is None
+                    for key in ("result_sha256", "frame_evidence_sha256")
+                )
+            ):
+                return False
+    return True
+
+
+def _valid_legacy_load_envelope(value: dict[str, object]) -> bool:
     if (
         value.get("production_rate_repeatability_passed") is not False
         or value.get("reserve_12_fps_passed") is not True
@@ -450,10 +550,9 @@ def _valid_load_envelope(value: object) -> bool:
     ):
         return False
     scenarios = value.get("scenarios")
-    scenario_ids = [item.get("id") for item in scenarios if isinstance(item, dict)] if (
-        isinstance(scenarios, list)
-    ) else []
-    if scenario_ids != ["production-10fps", "reserve-12fps", "limit-15fps"]:
+    if not isinstance(scenarios, list) or [
+        item.get("id") for item in scenarios if isinstance(item, dict)
+    ] != ["production-10fps", "reserve-12fps", "limit-15fps"]:
         return False
     expected_targets = (False, True, True)
     required_numeric = {
@@ -485,7 +584,7 @@ def _valid_load_envelope(value: object) -> bool:
             or scenario.get("additional_inference_passes") != 0
             or scenario.get("queue_capacity") != 2
             or not isinstance(scenario.get("queue_high_watermarks"), dict)
-            or any(value != 2 for value in scenario["queue_high_watermarks"].values())
+            or any(depth != 2 for depth in scenario["queue_high_watermarks"].values())
             or not isinstance(scenario.get("thresholds"), dict)
             or any(not _nonnegative_number(scenario.get(key)) for key in required_numeric)
             or not isinstance(scenario.get("frame_evidence_sha256"), str)