#!/usr/bin/env python3 """Produce reviewable multi-run reference evidence through the product path. Each repetition runs in a fresh process using the workload's canonical seed. The tool never patches framework RNG functions. A run is invalid when the report or provenance manifest records a different seed, its manifest does not verify, grading fails, or the score-bearing data mode differs from the canonical contract. When a measurement protocol declares complete preconditioning executions, their artifacts are retained and validated but excluded from all evidence aggregates. Evidence is written to a new attempt directory and never overwritten. Every run artifact is SHA-256 indexed, and the final evidence summary receives a separate unauthenticated SHA-256 digest sidecar. These digests detect changes; they do not authenticate who produced the evidence. """ from __future__ import annotations import argparse import hashlib import json import math import os import platform import re import shutil import stat import statistics import subprocess import sys import tempfile import time import zipfile from datetime import datetime, timezone from pathlib import Path, PurePosixPath from typing import Any, Mapping ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) TOOL_NAME = "run_reference_sweep.py" TOOL_VERSION = "4.4.0" TOOL_ID = f"tools/{TOOL_NAME} v{TOOL_VERSION}" SCORE_PUBLIC_DATA_MODES = frozenset( {"real", "real-preprocessed-mlperf-tiny-accuracy-set"} ) PERFORMANCE_PUBLIC_DATA_MODES = frozenset({"real", "checkpoint-backed", "local-prompt"}) PUBLIC_STATUSES = frozenset({"score-bearing", "performance-bearing"}) RESULT_ROLES = frozenset({"score-bearing", "performance-bearing"}) REFERENCE_EVIDENCE_SCHEMA = "mlperf-edu-reference-evidence/0.7" HOST_POWER_STATE_SCHEMA = "mlperf-edu-host-power-state/0.1" POWER_STABILITY_POLICY_SCHEMA = "mlperf-edu-power-stability-policy/0.1" DEFAULT_TIMEOUT_SECONDS = 7200.0 MAX_INTER_EXECUTION_COOLDOWN_SECONDS = 300.0 PUBLIC_PRIMARY_METRIC_CV_LIMIT = 0.05 DEFAULT_OUTPUT_DIR = Path.home() / ".mlperf-edu" / "reference_runs" MAX_LINEAGE_PACKAGE_MEMBERS = 256 MAX_LINEAGE_PACKAGE_BYTES = 2 * 1024**3 MAX_LINEAGE_PACKAGE_INDEX_BYTES = 1 << 20 NANOGPT_LINEAGE_ENV = { "checkpoint": "MLPERF_EDU_NANOGPT_CHECKPOINT", "report": "MLPERF_EDU_NANOGPT_TRAIN_REPORT", "manifest": "MLPERF_EDU_NANOGPT_TRAIN_MANIFEST", } POWER_STABILITY_POLICY = { "schema": POWER_STABILITY_POLICY_SCHEMA, "scope": "each complete preconditioning and measured subprocess execution", "promotion_requires_external_power": True, "promotion_requires_low_power_mode_disabled": True, "power_source_change_invalidates_execution": True, "power_mode_change_invalidates_execution": True, "sleep_or_wake_invalidates_execution": True, "unsupported_monitoring_invalidates_promotion": True, } def _command_output(command: list[str]) -> tuple[str | None, str | None]: """Return stripped command output without allowing a telemetry failure to crash.""" try: process = subprocess.run( command, check=False, capture_output=True, text=True, timeout=10, ) except (OSError, subprocess.SubprocessError) as exc: return None, f"{type(exc).__name__}: {exc}" if process.returncode != 0: detail = _tail(process.stderr or process.stdout, lines=3) return None, f"exit {process.returncode}: {detail or 'no diagnostic'}" return process.stdout.strip(), None def _parse_pmset_battery(text: str) -> dict[str, Any]: source_match = re.search(r"Now drawing from '([^']+)'", text) percent_match = re.search(r"\b(\d{1,3})%;", text) detail_match = re.search(r"\b\d{1,3}%;\s*([^;\n]+)", text) raw_source = source_match.group(1) if source_match else None source = { "AC Power": "external", "Battery Power": "battery", }.get(raw_source, "unknown") percent = int(percent_match.group(1)) if percent_match else None if percent is not None and not 0 <= percent <= 100: percent = None return { "source": source, "source_raw": raw_source, "battery_percent": percent, "battery_status": detail_match.group(1).strip() if detail_match else None, } def _parse_pmset_power_mode(text: str, source_raw: str | None) -> int | None: if source_raw not in {"AC Power", "Battery Power"}: return None current_section = None for raw_line in text.splitlines(): line = raw_line.strip() if line.endswith("Power:"): current_section = line.removesuffix(":") continue if current_section != source_raw: continue match = re.fullmatch(r"(?:powermode|lowpowermode)\s+(-?\d+)", line) if match: return int(match.group(1)) return None def _parse_sysctl_epoch(text: str) -> int | None: match = re.search(r"\bsec\s*=\s*(\d+)", text) return int(match.group(1)) if match else None def _capture_darwin_power_state() -> dict[str, Any]: errors: list[str] = [] battery, error = _command_output(["pmset", "-g", "batt"]) if error: errors.append(f"pmset battery query failed: {error}") parsed = _parse_pmset_battery(battery or "") settings, error = _command_output(["pmset", "-g", "custom"]) if error: errors.append(f"pmset settings query failed: {error}") power_mode = _parse_pmset_power_mode(settings or "", parsed["source_raw"]) sleep_text, error = _command_output(["sysctl", "-n", "kern.sleeptime"]) if error: errors.append(f"sleep-state query failed: {error}") wake_text, error = _command_output(["sysctl", "-n", "kern.waketime"]) if error: errors.append(f"wake-state query failed: {error}") sleep_epoch = _parse_sysctl_epoch(sleep_text or "") wake_epoch = _parse_sysctl_epoch(wake_text or "") supported = ( parsed["source"] != "unknown" and power_mode is not None and sleep_epoch is not None and wake_epoch is not None and not errors ) return { "provider": "macos-pmset-sysctl", "supported": supported, **parsed, "power_mode": power_mode, "low_power_mode": power_mode == 1 if power_mode is not None else None, "last_sleep_epoch": sleep_epoch, "last_wake_epoch": wake_epoch, "suspend_clock_offset_seconds": None, "query_errors": errors, } def _capture_linux_power_state() -> dict[str, Any]: root = Path("/sys/class/power_supply") external_online: list[bool] = [] battery_present = False errors: list[str] = [] if root.is_dir(): for supply in sorted(root.iterdir()): try: supply_type = (supply / "type").read_text().strip() except OSError: continue if supply_type == "Battery": battery_present = True if supply_type not in {"Mains", "USB", "USB_C", "Wireless"}: continue try: external_online.append((supply / "online").read_text().strip() == "1") except OSError as exc: errors.append(f"{supply.name} online query failed: {exc}") if any(external_online): source = "external" elif external_online and battery_present: source = "battery" else: source = "unknown" profile, profile_error = _command_output(["powerprofilesctl", "get"]) if profile_error: profile = None suspend_clock_offset = None if hasattr(time, "CLOCK_BOOTTIME"): try: suspend_clock_offset = ( time.clock_gettime(time.CLOCK_BOOTTIME) - time.monotonic() ) except OSError as exc: errors.append(f"CLOCK_BOOTTIME query failed: {exc}") supported = source != "unknown" and suspend_clock_offset is not None and not errors return { "provider": "linux-sysfs-clock-boottime", "supported": supported, "source": source, "source_raw": source, "battery_percent": None, "battery_status": None, "power_mode": profile, "low_power_mode": profile == "power-saver" if profile is not None else None, "last_sleep_epoch": None, "last_wake_epoch": None, "suspend_clock_offset_seconds": suspend_clock_offset, "query_errors": errors, } def capture_host_power_state() -> dict[str, Any]: """Capture source, power mode, and a suspend/wake marker for one boundary.""" system = platform.system() if system == "Darwin": state = _capture_darwin_power_state() elif system == "Linux": state = _capture_linux_power_state() else: state = { "provider": "unsupported", "supported": False, "source": "unknown", "source_raw": None, "battery_percent": None, "battery_status": None, "power_mode": None, "low_power_mode": None, "last_sleep_epoch": None, "last_wake_epoch": None, "suspend_clock_offset_seconds": None, "query_errors": [f"no power-state provider for {system}"], } return { "schema": HOST_POWER_STATE_SCHEMA, "platform": system, "captured_at": datetime.now(timezone.utc).isoformat(), **state, } def assess_power_stability( before: Mapping[str, Any], after: Mapping[str, Any], *, require_promotion_conditions: bool, ) -> list[str]: """Fail closed on source/mode/sleep changes and on unsafe promotion power.""" reasons: list[str] = [] if ( before.get("schema") != HOST_POWER_STATE_SCHEMA or after.get("schema") != HOST_POWER_STATE_SCHEMA ): reasons.append("host power-state snapshot schema is missing or unsupported") return reasons if before.get("provider") != after.get("provider"): reasons.append("host power-state provider changed during execution") if before.get("source") != after.get("source"): reasons.append("host power source changed during execution") if before.get("power_mode") != after.get("power_mode"): reasons.append("host power mode changed during execution") if before.get("provider") == "linux-sysfs-clock-boottime": before_offset = before.get("suspend_clock_offset_seconds") after_offset = after.get("suspend_clock_offset_seconds") if ( isinstance(before_offset, bool) or not isinstance(before_offset, (int, float)) or isinstance(after_offset, bool) or not isinstance(after_offset, (int, float)) or abs(float(after_offset) - float(before_offset)) > 1.0 ): reasons.append("host suspend clock changed during execution") else: if before.get("last_sleep_epoch") != after.get("last_sleep_epoch"): reasons.append("host entered sleep during execution") if before.get("last_wake_epoch") != after.get("last_wake_epoch"): reasons.append("host wake state changed during execution") if require_promotion_conditions: if before.get("supported") is not True or after.get("supported") is not True: reasons.append( "promotion evidence requires supported host power monitoring" ) if before.get("source") != "external" or after.get("source") != "external": reasons.append( "promotion evidence requires external power throughout execution" ) if ( before.get("low_power_mode") is not False or after.get("low_power_mode") is not False ): reasons.append( "promotion evidence requires Low Power Mode to remain disabled" ) return reasons _CHILD_BOOTSTRAP = r""" import hashlib import json import math import os import shutil import sys import time import traceback from pathlib import Path def sha256_file(path): digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1 << 20), b""): digest.update(chunk) return "sha256:" + digest.hexdigest() def resolve_workload(registry, workload_id, variant): if variant: for workload in registry.values(): if (getattr(workload, "canonical_workload", None) == workload_id and getattr(workload, "variant", None) == variant): return workload raise KeyError(f"no variant {variant!r} for workload {workload_id!r}") if workload_id in registry: return registry[workload_id] raise KeyError(f"workload {workload_id!r} not found") def artifact_index(report, report_path, manifest_path, exports): candidates = {"report": report_path, "provenance": manifest_path} for role, path in exports.items(): candidates[f"report_{role}"] = Path(path) for role, value in (report.get("artifacts") or {}).items(): if not isinstance(value, str) or not value: continue path = Path(value) if not path.is_absolute(): path = report_path.parent / path candidates[str(role)] = path indexed = [] seen = set() for role, path in sorted(candidates.items()): path = path.resolve() if path in seen or not path.is_file(): continue seen.add(path) run_root = report_path.parent.resolve() try: path.relative_to(run_root) except ValueError: source_digest = sha256_file(path) safe_role = "".join( character if character.isalnum() or character in "._-" else "_" for character in str(role) ).strip("._-") or "artifact" safe_name = "".join( character if character.isalnum() or character in "._-" else "_" for character in path.name ).strip("._-") or "artifact" retained_dir = run_root.parent / "retained_artifacts" retained_dir.mkdir(parents=True, exist_ok=True) retained = retained_dir / ( f"{safe_role}-{source_digest.removeprefix('sha256:')}-{safe_name}" ) temporary = retained.with_name(retained.name + ".tmp") if retained.exists(): if ( retained.stat().st_size != path.stat().st_size or sha256_file(retained) != source_digest ): raise ValueError( f"retained artifact collision for {role}: {retained}" ) else: shutil.copyfile(path, temporary) if ( temporary.stat().st_size != path.stat().st_size or sha256_file(temporary) != source_digest ): temporary.unlink(missing_ok=True) raise ValueError(f"retained artifact copy failed for {role}") os.replace(temporary, retained) path = retained.resolve() indexed.append({ "role": role, "path": str(path), "sha256": sha256_file(path), "n_bytes": path.stat().st_size, }) return indexed def main(): args_path = Path(sys.argv[1]) result_path = Path(sys.argv[2]) args = json.loads(args_path.read_text()) root = Path(args["root"]) sys.path.insert(0, str(root / "src")) seed = int(args["seed"]) os.environ.pop("MLPERF_EDU_SEED", None) os.environ["MLPERF_EDU_MAX_SEED"] = str(seed) if args.get("device"): os.environ["MLPERF_EDU_DEVICE"] = str(args["device"]) result = { "execution_index": int(args["execution_index"]), "requested_seed": seed, "workload_id": args["workload_id"], "profile": args["profile"], "execution_ok": False, "evidence_valid": False, } try: from mlperf.edu_cli import ( annotate_execution_device, attach_run_fingerprints, enrich_report_for_display, grade_manifest, metric_key_for_quality, run_comparison_fingerprint_sha256, run_workload, update_measurement_manifest, write_report_exports, ) from mlperf.manifest import verify_provd from mlperf.contracts import evaluate_promotion_contract from mlperf.registry import load_registry registry = load_registry() workload = resolve_workload(registry, args["workload_id"], args.get("variant")) output_dir = Path(args["output_dir"]) output_dir.mkdir(parents=True, exist_ok=False) started = time.perf_counter() report = run_workload( workload, args["profile"], output_dir, mode=args.get("mode"), phase=args.get("phase"), ) wall_seconds = time.perf_counter() - started annotate_execution_device(report) artifacts = report.get("artifacts") or {} report_path = Path(str(artifacts.get("report", ""))).resolve() manifest_path = Path(str(artifacts.get("provenance", ""))).resolve() if not report_path.is_file() or not manifest_path.is_file(): raise FileNotFoundError("runner did not produce both report and provenance artifacts") # Match the normal CLI post-run path before verification and grading. report = dict(report) enrich_report_for_display(report, registry) attach_run_fingerprints(report) promotion_contract = evaluate_promotion_contract(workload, report) report["promotion_contract"] = promotion_contract report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n") update_measurement_manifest(report, report_path, manifest_path) exports = write_report_exports(report, report_path, open_report=False) manifest = json.loads(manifest_path.read_text()) verification = verify_provd(manifest_path, repo_root=root) grade = grade_manifest(manifest_path) quality = report.get("quality") or {} metrics = report.get("metrics") or {} phase_contracts = ( ((workload.raw.get("mode_contracts") or {}).get("inference") or {}).get( "phases", {} ) ) if args.get("mode") == "inference" and phase_contracts: phase_contract = phase_contracts.get(args.get("phase") or "full", {}) measurement_protocol = phase_contract.get("measurement_protocol") or {} registry_scenario = phase_contract.get("scenario") else: measurement_protocol = workload.raw.get("measurement_protocol") or {} registry_scenario = getattr(workload, "scenario", None) primary_metric = measurement_protocol.get("primary_metric") quality_metric = quality.get("metric") or getattr(workload, "quality_metric", None) functional_metric = quality_metric def finite_metric(metric_name, *, preferred_key=None): key = None if preferred_key and preferred_key in metrics: key = str(preferred_key) elif metric_name: key = metric_key_for_quality(metric_name, metrics) value = metrics.get(key) if key else None if ( isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)) ): return key, None return key, float(value) primary_metric_key, primary_metric_value = finite_metric(primary_metric) gate_metric = quality_metric or functional_metric gate_metric_key, gate_metric_value = finite_metric( gate_metric, preferred_key=quality.get("metric_key") ) report_seed = report.get("seed") manifest_seed = ((manifest.get("leaves") or {}).get("rng") or {}).get("seed") data_mode = report.get("data_mode") report_scenario = report.get("scenario") manifest_scenario = manifest.get("scenario") execution_backend = str(report.get("backend") or "") requested_device = args.get("device") report_requested_device = report.get("device_requested") report_executed_device = report.get("device_executed") invalid_reasons = [] if report.get("status") != "passed": invalid_reasons.append(f"report status is {report.get('status')!r}, not 'passed'") if not verification.all_ok: invalid_reasons.append("provenance verification failed") if not grade.get("passed"): invalid_reasons.append("submission grading failed") if report_seed != seed: invalid_reasons.append(f"report recorded seed {report_seed!r}, requested {seed}") if manifest_seed != seed: invalid_reasons.append(f"manifest recorded seed {manifest_seed!r}, requested {seed}") if args["evidence_tier"] in {"public-candidate", "promotion-candidate"}: if not registry_scenario: invalid_reasons.append("registry scenario is missing") if report_scenario != registry_scenario: invalid_reasons.append( f"report scenario {report_scenario!r} does not match registry " f"scenario {registry_scenario!r}" ) if manifest_scenario != registry_scenario: invalid_reasons.append( f"manifest scenario {manifest_scenario!r} does not match registry " f"scenario {registry_scenario!r}" ) if primary_metric_value is None: invalid_reasons.append( f"declared primary metric {primary_metric!r} has no finite numeric report value" ) if gate_metric_value is None: invalid_reasons.append( f"declared quality metric {quality_metric!r} has no finite numeric report value" ) if quality.get("target_met") is not True: invalid_reasons.append( f"{workload.public_status} quality or functional gate did not pass" ) if requested_device and requested_device.lower() not in execution_backend.lower(): invalid_reasons.append( f"report execution backend {execution_backend!r} does not match requested device {requested_device!r}" ) expected_requested_device = str(requested_device or "auto").lower() if report_requested_device != expected_requested_device: invalid_reasons.append( f"report device_requested {report_requested_device!r} does not match requested device {expected_requested_device!r}" ) if not isinstance(report_executed_device, str) or not report_executed_device: invalid_reasons.append("report device_executed is missing") elif requested_device and report_executed_device != requested_device.lower(): invalid_reasons.append( f"report device_executed {report_executed_device!r} does not match requested device {requested_device!r}" ) if args["evidence_tier"] in {"public-candidate", "promotion-candidate"} and data_mode not in args["allowed_data_modes"]: invalid_reasons.append( f"public candidate data_mode {data_mode!r} is not allowed for {workload.public_status} evidence" ) review_contract = report.get("review_contract") or {} if args["evidence_tier"] == "public-candidate" and review_contract.get("status") != "passed": invalid_reasons.append("report review contract did not pass") if ( args["evidence_tier"] == "promotion-candidate" and promotion_contract.get("status") != "passed" ): invalid_reasons.append( "report promotion contract did not pass: " + "; ".join(promotion_contract.get("issues") or []) ) if args["evidence_tier"] == "public-candidate": expected_review = { "metric": primary_metric, "metric_key": primary_metric_key, "metric_value": primary_metric_value, "functional_metric": gate_metric, "functional_metric_key": gate_metric_key, "functional_metric_value": gate_metric_value, } for field, expected in expected_review.items(): if review_contract.get(field) != expected: invalid_reasons.append( f"review contract {field}={review_contract.get(field)!r} " f"does not match raw report metric {expected!r}" ) expected_grade = { "passed": True, "target_met": True, "metric": gate_metric, "value": gate_metric_value, "target": quality.get("target"), } for field, expected in expected_grade.items(): if grade.get(field) != expected: invalid_reasons.append( f"grade {field}={grade.get(field)!r} does not match " f"quality or functional gate {expected!r}" ) fingerprint = report.get("run_fingerprint") or {} comparison_fingerprint_sha256 = fingerprint.get( "comparison_fingerprint_sha256" ) recomputed_comparison_fingerprint_sha256 = ( run_comparison_fingerprint_sha256(fingerprint) if isinstance(fingerprint, dict) else None ) if ( not isinstance(comparison_fingerprint_sha256, str) or len(comparison_fingerprint_sha256) != 64 or any( character not in "0123456789abcdef" for character in comparison_fingerprint_sha256 ) ): invalid_reasons.append( "report comparison_fingerprint_sha256 is missing or malformed" ) elif comparison_fingerprint_sha256 != recomputed_comparison_fingerprint_sha256: invalid_reasons.append( "report comparison_fingerprint_sha256 does not match its canonical " "comparison record" ) hardware = fingerprint.get("hardware") or {} execution = fingerprint.get("execution") or {} result.update({ "execution_ok": True, "evidence_valid": not invalid_reasons, "invalid_reasons": invalid_reasons, "status": report.get("status"), "report_recorded_seed": report_seed, "manifest_recorded_seed": manifest_seed, "primary_metric_declared": primary_metric, "primary_metric_key": primary_metric_key, "primary_metric_value": primary_metric_value, "quality_metric_declared": quality_metric, "quality_metric_key": gate_metric_key, "quality_value": gate_metric_value, "functional_metric_declared": functional_metric, "functional_metric_key": gate_metric_key if functional_metric else None, "functional_metric_value": gate_metric_value if functional_metric else None, "reference_metric_role": "performance", "quality_target_met": quality.get("target_met"), "quality_target": quality.get("target"), "quality_direction": quality.get("direction"), "promotion_contract": promotion_contract, "result_role": promotion_contract.get("result_role"), "comparison_fingerprint_sha256": comparison_fingerprint_sha256, "scenario": report_scenario, "manifest_scenario": manifest_scenario, "registry_scenario": registry_scenario, "wall_seconds": wall_seconds, "backend": execution_backend or None, "device_requested": report_requested_device, "device_executed": report_executed_device, "hardware_backend": hardware.get("backend"), "chip": hardware.get("chip"), "fingerprint_backends": execution.get("backends"), "data_mode": data_mode, "report_path": str(report_path), "manifest_path": str(manifest_path), "manifest_verified": verification.all_ok, "verification_checks": [ {"check": name, "ok": ok, "detail": detail} for name, ok, detail in verification.checks ], "grade": { "passed": grade.get("passed"), "status": grade.get("status"), "metric": grade.get("metric"), "value": grade.get("value"), "target": grade.get("target"), "target_met": grade.get("target_met"), }, "artifacts": artifact_index(report, report_path, manifest_path, exports), }) except Exception as exc: result.update({ "execution_ok": False, "evidence_valid": False, "invalid_reasons": [f"{type(exc).__name__}: {exc}"], "error": f"{type(exc).__name__}: {exc}", "traceback": traceback.format_exc(), }) result_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") return 0 if result.get("evidence_valid") else 1 if __name__ == "__main__": raise SystemExit(main()) """ def parse_seeds(raw: str) -> list[int]: try: seeds = [int(token.strip()) for token in raw.split(",") if token.strip()] except ValueError as exc: raise argparse.ArgumentTypeError( "--seeds must be comma-separated integers" ) from exc if not seeds: raise argparse.ArgumentTypeError("--seeds must include at least one seed") if len(set(seeds)) != len(seeds): raise argparse.ArgumentTypeError("--seeds must not contain duplicates") return seeds def parse_run_count(raw: str) -> int: try: count = int(raw) except ValueError as exc: raise argparse.ArgumentTypeError("--runs must be an integer") from exc if count < 1: raise argparse.ArgumentTypeError("--runs must be at least one") return count def parse_preconditioning_run_count(raw: str) -> int: try: count = int(raw) except ValueError as exc: raise argparse.ArgumentTypeError( "--preconditioning-runs must be an integer" ) from exc if count < 0: raise argparse.ArgumentTypeError("--preconditioning-runs must be nonnegative") return count def canonical_seed(workload: Any) -> int: contract = workload.raw.get("canonical_max_contract") or {} config = contract.get("config") or {} value = config.get("random_seed", 42) if isinstance(value, bool) or not isinstance(value, int): raise ValueError(f"{workload.id} canonical random seed must be an integer") return value def execution_contract( workload: Any, *, mode: str | None, phase: str | None ) -> dict[str, Any]: phases = ((workload.raw.get("mode_contracts") or {}).get("inference") or {}).get( "phases", {} ) if mode == "inference" and phases: resolved_phase = phase or "full" contract = phases.get(resolved_phase) if not isinstance(contract, dict): raise ValueError( f"{workload.id} has no inference phase contract for {resolved_phase!r}" ) return contract canonical = workload.raw.get("canonical_max_contract") or {} canonical_mode = canonical.get("mode") if mode is not None and mode != canonical_mode: raise ValueError( f"{workload.id} canonical max mode is {canonical_mode!r}, received {mode!r}" ) return { "result_role": canonical.get("result_role"), "scenario": workload.scenario, "measurement_protocol": workload.raw.get("measurement_protocol") or {}, "config": canonical.get("config") or {}, "quality": canonical.get("quality") or {}, } def execution_result_role(workload: Any, *, mode: str | None, phase: str | None) -> str: """Return the score/performance role for one execution case.""" role = execution_contract(workload, mode=mode, phase=phase).get("result_role") if role not in RESULT_ROLES: raise ValueError( f"{workload.id} execution contract must declare result_role as one of " f"{sorted(RESULT_ROLES)}, received {role!r}" ) return str(role) def aggregate(values: list[float]) -> dict[str, Any]: clean = [float(value) for value in values if math.isfinite(float(value))] if not clean: return { "count": 0, "median": None, "mean": None, "min": None, "max": None, "stdev": None, } return { "count": len(clean), "median": statistics.median(clean), "mean": statistics.fmean(clean), "min": min(clean), "max": max(clean), "stdev": statistics.stdev(clean) if len(clean) > 1 else 0.0, } def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1 << 20), b""): digest.update(chunk) return "sha256:" + digest.hexdigest() class LineagePackageError(ValueError): """Raised when a NanoGPT training package is unsafe or unverifiable.""" def _safe_archive_member_name(name: str) -> bool: """Return whether *name* is a strict, portable archive-relative path.""" if not name or "\\" in name or "\x00" in name or name.endswith("/"): return False parts = name.split("/") if any(part in {"", ".", ".."} for part in parts): return False path = PurePosixPath(name) if path.is_absolute() or path.as_posix() != name: return False return ":" not in parts[0] def _preflight_lineage_archive(package_path: Path) -> dict[str, Any]: """Reject unsafe ZIP structure before any package verifier extracts it.""" try: with zipfile.ZipFile(package_path) as zf: infos = zf.infolist() names = [info.filename for info in infos] if len(infos) > MAX_LINEAGE_PACKAGE_MEMBERS: raise LineagePackageError( f"lineage package contains {len(infos)} members; " f"limit is {MAX_LINEAGE_PACKAGE_MEMBERS}" ) if len(names) != len(set(names)): raise LineagePackageError( "lineage package contains duplicate member names" ) total_size = 0 for info in infos: if not _safe_archive_member_name(info.filename): raise LineagePackageError( f"unsafe lineage package member path: {info.filename!r}" ) if info.flag_bits & 0x1: raise LineagePackageError( f"encrypted lineage package member is unsupported: {info.filename!r}" ) file_type = stat.S_IFMT((info.external_attr >> 16) & 0xFFFF) if info.is_dir() or file_type not in {0, stat.S_IFREG}: raise LineagePackageError( f"lineage package member is not a regular file: {info.filename!r}" ) total_size += info.file_size if total_size > MAX_LINEAGE_PACKAGE_BYTES: raise LineagePackageError( f"lineage package expands to {total_size} bytes; " f"limit is {MAX_LINEAGE_PACKAGE_BYTES}" ) try: index_info = zf.getinfo("package_index.json") except KeyError as exc: raise LineagePackageError( "lineage package is missing package_index.json" ) from exc if index_info.file_size > MAX_LINEAGE_PACKAGE_INDEX_BYTES: raise LineagePackageError("lineage package index is unexpectedly large") try: index = json.loads(zf.read(index_info)) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise LineagePackageError( "lineage package index is not valid UTF-8 JSON" ) from exc except (OSError, zipfile.BadZipFile) as exc: raise LineagePackageError(f"cannot read lineage package: {exc}") from exc if not isinstance(index, dict) or index.get("schema") != "mlperf-edu-package/0.2": raise LineagePackageError( "lineage package must use schema mlperf-edu-package/0.2" ) if index.get("workload") != "causal-language-modeling": raise LineagePackageError( "lineage package must contain a causal-language-modeling result" ) included = index.get("included_files") if not isinstance(included, list) or not included: raise LineagePackageError("lineage package index has no included files") indexed_names: list[str] = [] info_by_name = {info.filename: info for info in infos} for item in included: if not isinstance(item, dict): raise LineagePackageError("lineage package index entries must be objects") archive_name = item.get("path") if not isinstance(archive_name, str) or not _safe_archive_member_name( archive_name ): raise LineagePackageError( f"lineage package index has an unsafe path: {archive_name!r}" ) indexed_names.append(archive_name) info = info_by_name.get(archive_name) if info is None: raise LineagePackageError( f"lineage package index references a missing member: {archive_name}" ) if item.get("n_bytes") != info.file_size: raise LineagePackageError( f"lineage package index size does not match {archive_name}" ) if len(indexed_names) != len(set(indexed_names)): raise LineagePackageError("lineage package index contains duplicate paths") expected_names = {"package_index.json", *indexed_names} if set(names) != expected_names: extras = sorted(set(names) - expected_names) missing = sorted(expected_names - set(names)) raise LineagePackageError( f"lineage package/index coverage mismatch; extras={extras}, missing={missing}" ) manifest_name = index.get("manifest") or index.get("source_manifest") if ( not isinstance(manifest_name, str) or not _safe_archive_member_name(manifest_name) or manifest_name not in indexed_names ): raise LineagePackageError( "lineage package index does not identify an indexed manifest" ) return index def _verify_package_checks(package_path: Path) -> list[tuple[str, bool, str]]: from mlperf.edu_cli import verify_package_archive return verify_package_archive(package_path, repo_root=ROOT) def _verify_provenance_checks(manifest_path: Path) -> list[tuple[str, bool, str]]: from mlperf.manifest import verify_provd return verify_provd(manifest_path, repo_root=ROOT).checks def validate_nanogpt_lineage_package(package_path: Path) -> dict[str, Any]: """Verify a portable NanoGPT training package before creating an attempt.""" package_path = package_path.expanduser().resolve() if not package_path.is_file(): raise LineagePackageError(f"lineage package not found: {package_path}") package_sha256 = sha256_file(package_path) index = _preflight_lineage_archive(package_path) checks = _verify_package_checks(package_path) failed = [name for name, ok, _detail in checks if not ok] if failed: raise LineagePackageError( f"lineage package failed clean-extraction verification: {failed}" ) if sha256_file(package_path) != package_sha256: raise LineagePackageError("lineage package changed during verification") return { "package_path": package_path, "package_sha256": package_sha256, "index": index, "verification_checks": checks, } def _staged_artifact_path( owner_path: Path, raw_path: Any, stage_root: Path, role: str ) -> Path: if not isinstance(raw_path, str) or not raw_path: raise LineagePackageError(f"lineage manifest does not declare {role}") artifact_path = Path(raw_path) if artifact_path.is_absolute(): raise LineagePackageError(f"lineage manifest {role} path must be relative") resolved = (owner_path.parent / artifact_path).resolve() try: resolved.relative_to(stage_root.resolve()) except ValueError as exc: raise LineagePackageError( f"lineage manifest {role} path escapes the staged package" ) from exc if not resolved.is_file(): raise LineagePackageError(f"staged lineage {role} is missing: {raw_path}") return resolved def _validate_staged_lineage( stage_root: Path, index: dict[str, Any] ) -> dict[str, Path]: """Locate and semantically validate the packaged max-training lineage.""" manifest_name = str(index.get("manifest") or index.get("source_manifest")) manifest_path = (stage_root / manifest_name).resolve() try: manifest_path.relative_to(stage_root.resolve()) except ValueError as exc: raise LineagePackageError( "packaged manifest escapes the staging directory" ) from exc if not manifest_path.is_file(): raise LineagePackageError("packaged NanoGPT training manifest is missing") try: manifest = json.loads(manifest_path.read_text()) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise LineagePackageError( "packaged NanoGPT training manifest is invalid" ) from exc if manifest.get("workload") != "causal-language-modeling": raise LineagePackageError( "packaged provenance is not for causal-language-modeling" ) leaves = manifest.get("leaves") or {} report_path = _staged_artifact_path( manifest_path, (leaves.get("measurement") or {}).get("report_path"), stage_root, "training report", ) checkpoint_path = _staged_artifact_path( manifest_path, (leaves.get("weights") or {}).get("path"), stage_root, "checkpoint", ) try: report = json.loads(report_path.read_text()) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise LineagePackageError( "packaged NanoGPT training report is invalid" ) from exc quality = report.get("quality") or {} if ( report.get("workload") != "causal-language-modeling" or report.get("mode") != "training" or report.get("profile") != "max" or report.get("status") != "passed" or report.get("data_mode") != "real" or quality.get("quality_required") is not True or quality.get("target_met") is not True ): raise LineagePackageError( "lineage package must contain a passing real-data " "causal-language-modeling training max report" ) provenance_checks = _verify_provenance_checks(manifest_path) failed = [name for name, ok, _detail in provenance_checks if not ok] if failed: raise LineagePackageError( f"staged NanoGPT training provenance failed verification: {failed}" ) return { "manifest": manifest_path, "report": report_path, "checkpoint": checkpoint_path, } def stage_nanogpt_lineage_package( validation: dict[str, Any], attempt_dir: Path ) -> dict[str, Any]: """Safely extract verified NanoGPT training lineage inside an attempt.""" package_path = Path(validation["package_path"]) expected_sha256 = str(validation["package_sha256"]) if sha256_file(package_path) != expected_sha256: raise LineagePackageError("lineage package changed after verification") stage_root = attempt_dir / "inputs" / "nanogpt-training" stage_tmp = attempt_dir / "inputs" / ".nanogpt-training.tmp" stage_root.parent.mkdir(parents=True, exist_ok=True) stage_tmp.mkdir(exist_ok=False) try: with zipfile.ZipFile(package_path) as zf: # Re-run structural preflight immediately before extraction to close # the gap between verification and staging. current_index = _preflight_lineage_archive(package_path) if current_index != validation["index"]: raise LineagePackageError( "lineage package index changed after verification" ) for info in zf.infolist(): target = stage_tmp / PurePosixPath(info.filename) target.parent.mkdir(parents=True, exist_ok=True) with zf.open(info) as source, target.open("xb") as destination: shutil.copyfileobj(source, destination, length=1 << 20) for item in current_index["included_files"]: staged = stage_tmp / PurePosixPath(str(item["path"])) if staged.stat().st_size != item["n_bytes"]: raise LineagePackageError( f"staged lineage size does not match index: {item['path']}" ) if sha256_file(staged) != item.get("sha256"): raise LineagePackageError( f"staged lineage digest does not match index: {item['path']}" ) located = _validate_staged_lineage(stage_tmp, current_index) if sha256_file(package_path) != expected_sha256: raise LineagePackageError("lineage package changed during staging") relative_locations = { role: path.relative_to(stage_tmp) for role, path in located.items() } stage_tmp.rename(stage_root) except Exception: shutil.rmtree(stage_tmp, ignore_errors=True) raise paths = {role: stage_root / path for role, path in relative_locations.items()} environment = { NANOGPT_LINEAGE_ENV[role]: str(path.resolve()) for role, path in paths.items() } return { "package_sha256": expected_sha256, "package_schema": current_index["schema"], "source_workload": current_index["workload"], "stage_root": stage_root, "paths": paths, "environment": environment, "verification_check_count": len(validation["verification_checks"]), } def _tail(value: str | bytes | None, lines: int = 20) -> str: if isinstance(value, bytes): value = value.decode(errors="replace") return "\n".join((value or "").strip().splitlines()[-lines:]) def _relative_to_attempt(path_value: str | None, attempt_dir: Path) -> str | None: if not path_value: return None path = Path(path_value).resolve() try: return path.relative_to(attempt_dir.resolve()).as_posix() except ValueError: return str(path) def power_stability_record( before: Mapping[str, Any], after: Mapping[str, Any], *, evidence_tier: str, ) -> dict[str, Any]: require_promotion_conditions = evidence_tier in { "public-candidate", "promotion-candidate", } invalid_reasons = assess_power_stability( before, after, require_promotion_conditions=require_promotion_conditions, ) return { "policy": dict(POWER_STABILITY_POLICY), "promotion_conditions_required": require_promotion_conditions, "before": dict(before), "after": dict(after), "stable": not invalid_reasons, "invalid_reasons": invalid_reasons, } def _apply_power_stability( result: dict[str, Any], before: Mapping[str, Any], after: Mapping[str, Any], *, evidence_tier: str, ) -> dict[str, Any]: record = power_stability_record( before, after, evidence_tier=evidence_tier, ) reasons = list(result.get("invalid_reasons") or []) reasons.extend(f"power stability: {reason}" for reason in record["invalid_reasons"]) result["host_power"] = record result["invalid_reasons"] = reasons result["evidence_valid"] = bool(result.get("evidence_valid")) and record["stable"] return result def run_one_seed( bootstrap_path: Path, *, workload_id: str, variant: str | None, profile: str, seed: int, execution_index: int, mode: str | None, phase: str | None, device: str | None, attempt_dir: Path, timeout_seconds: float, evidence_tier: str, allowed_data_modes: frozenset[str], environment_overrides: dict[str, str] | None = None, cooldown_before_seconds: float = 0.0, run_group: str | None = None, ) -> dict[str, Any]: """Run one seed in a fresh process and return its validation record.""" if run_group not in {None, "preconditioning"}: raise ValueError(f"unsupported reference-sweep run group: {run_group!r}") if ( not math.isfinite(cooldown_before_seconds) or cooldown_before_seconds < 0 or cooldown_before_seconds > MAX_INTER_EXECUTION_COOLDOWN_SECONDS ): raise ValueError( "cooldown_before_seconds must be finite, nonnegative, and no greater " f"than {MAX_INTER_EXECUTION_COOLDOWN_SECONDS:g}" ) run_root = attempt_dir / run_group if run_group else attempt_dir run_dir = run_root / f"run_{execution_index:03d}" with tempfile.TemporaryDirectory( prefix=f"mlperf-edu-run-{execution_index:03d}-" ) as tmp: args_path = Path(tmp) / "args.json" result_path = Path(tmp) / "result.json" child_args = { "root": str(ROOT), "workload_id": workload_id, "variant": variant, "profile": profile, "seed": seed, "execution_index": execution_index, "mode": mode, "phase": phase, "device": device, "output_dir": str(run_dir), "evidence_tier": evidence_tier, "allowed_data_modes": sorted(allowed_data_modes), } args_path.write_text(json.dumps(child_args, indent=2, sort_keys=True) + "\n") env = sweep_environment(seed, device, environment_overrides) command = [ sys.executable, str(bootstrap_path), str(args_path), str(result_path), ] if cooldown_before_seconds > 0: time.sleep(cooldown_before_seconds) power_before = capture_host_power_state() preflight_power = power_stability_record( power_before, power_before, evidence_tier=evidence_tier, ) if not preflight_power["stable"]: return { "execution_index": execution_index, "requested_seed": seed, "execution_ok": False, "evidence_valid": False, "timed_out": False, "subprocess_wall_seconds": 0.0, "invalid_reasons": [ f"power stability: {reason}" for reason in preflight_power["invalid_reasons"] ], "host_power": preflight_power, "stdout_tail": "", "stderr_tail": "", "reproduce": reproduce_record( workload_id, variant, profile, seed, mode, phase, device, bool(environment_overrides), ), } started = time.perf_counter() try: process = subprocess.run( command, cwd=ROOT, env=env, capture_output=True, text=True, timeout=timeout_seconds, ) returncode = process.returncode stdout_tail = _tail(process.stdout) stderr_tail = _tail(process.stderr) except subprocess.TimeoutExpired as exc: result = { "execution_index": execution_index, "requested_seed": seed, "execution_ok": False, "evidence_valid": False, "timed_out": True, "timeout_seconds": timeout_seconds, "subprocess_wall_seconds": time.perf_counter() - started, "invalid_reasons": [f"run exceeded {timeout_seconds:g}-second timeout"], "stdout_tail": _tail(exc.stdout), "stderr_tail": _tail(exc.stderr), "reproduce": reproduce_record( workload_id, variant, profile, seed, mode, phase, device, bool(environment_overrides), ), } return _apply_power_stability( result, power_before, capture_host_power_state(), evidence_tier=evidence_tier, ) if result_path.is_file(): result = json.loads(result_path.read_text()) else: result = { "execution_index": execution_index, "requested_seed": seed, "execution_ok": False, "evidence_valid": False, "invalid_reasons": ["child process produced no result record"], } result["returncode"] = returncode result["subprocess_wall_seconds"] = time.perf_counter() - started result["stdout_tail"] = stdout_tail result["stderr_tail"] = stderr_tail result["reproduce"] = reproduce_record( workload_id, variant, profile, seed, mode, phase, device, bool(environment_overrides), ) result["report_path"] = _relative_to_attempt( result.get("report_path"), attempt_dir ) result["manifest_path"] = _relative_to_attempt( result.get("manifest_path"), attempt_dir ) for artifact in result.get("artifacts") or []: artifact["path"] = _relative_to_attempt(artifact.get("path"), attempt_dir) return _apply_power_stability( result, power_before, capture_host_power_state(), evidence_tier=evidence_tier, ) def reproduce_record( workload_id: str, variant: str | None, profile: str, seed: int, mode: str | None, phase: str | None, device: str | None, uses_nanogpt_lineage_package: bool = False, ) -> dict[str, Any]: command = ["uv", "run", "mlperf", "run", "--workload", workload_id] if variant: command.extend(["--variant", variant]) if mode: command.extend(["--mode", mode]) if phase: command.extend(["--phase", phase]) command.extend(["--profile", profile, "--output-dir", ""]) env = {"MLPERF_EDU_MAX_SEED": str(seed)} if device: env["MLPERF_EDU_DEVICE"] = device if uses_nanogpt_lineage_package: env.update( { NANOGPT_LINEAGE_ENV["checkpoint"]: "", NANOGPT_LINEAGE_ENV["report"]: "", NANOGPT_LINEAGE_ENV["manifest"]: "", } ) return {"command": command, "env": env} def sweep_environment( seed: int, device: str | None, overrides: dict[str, str] | None = None, ) -> dict[str, str]: """Return an isolated environment containing only explicit sweep controls.""" env = { name: value for name, value in os.environ.items() if not name.startswith("MLPERF_EDU_") } env["MLPERF_EDU_MAX_SEED"] = str(seed) if device: env["MLPERF_EDU_DEVICE"] = device else: env.pop("MLPERF_EDU_DEVICE", None) if overrides: unexpected = sorted(set(overrides) - set(NANOGPT_LINEAGE_ENV.values())) if unexpected: raise ValueError( f"unsupported reference sweep environment overrides: {unexpected}" ) env.update(overrides) return env def build_outer_execution_policy( *, public_status: str, evidence_tier: str, seeds: list[int], configured_cooldown_seconds: float, ) -> dict[str, Any]: """Describe the fixed delay between fresh timed public-candidate processes.""" applies = evidence_tier in {"public-candidate", "promotion-candidate"} executions = [ { "execution_index": index, "seed": seed, "fresh_process": True, "cooldown_before_seconds": ( configured_cooldown_seconds if applies and index > 1 else 0.0 ), } for index, seed in enumerate(seeds, start=1) ] return { "scope": "outer-process-executions", "applies": applies, "applicability": ( "all public-candidate score-bearing and performance-bearing workloads" ), "mode": "fixed-delay-between-fresh-processes" if applies else "not-applied", "execution_unit": "one fresh Python subprocess per repetition", "process_execution_count": len(executions), "configured_cooldown_seconds": configured_cooldown_seconds, "maximum_cooldown_seconds": MAX_INTER_EXECUTION_COOLDOWN_SECONDS, "first_execution_has_no_cooldown": True, "timing_scope": ( "The cooldown occurs before subprocess launch and is excluded from " "both subprocess wall time and within-run timing samples." ), "executions": executions, } def build_preconditioning_policy( *, seed: int, execution_count: int, ) -> dict[str, Any]: """Describe complete untimed executions that establish a warm device state.""" if execution_count < 0: raise ValueError("preconditioning execution_count must be nonnegative") executions = [ { "execution_index": index, "seed": seed, "fresh_process": True, "output_group": "preconditioning", } for index in range(1, execution_count + 1) ] return { "scope": "outer-process-preconditioning", "applies": bool(executions), "mode": "complete-canonical-executions" if executions else "not-applied", "execution_unit": "one complete canonical workload in a fresh Python subprocess", "process_execution_count": len(executions), "cooldown_between_executions_seconds": 0.0, "cooldown_before_first_measured_execution_seconds": 0.0, "timing_scope": ( "Preconditioning completes before evidence repetitions. Its workload " "timings and quality results are retained for audit but excluded from " "all evidence aggregates, acceptance statistics, and repeatability." ), "executions": executions, } def declared_preconditioning_runs(measurement_protocol: dict[str, Any]) -> int: """Return the canonical aggregate-excluded preparation count.""" count = measurement_protocol.get("outer_preconditioning_runs") if isinstance(count, bool) or not isinstance(count, int) or count < 0: raise ValueError( "measurement_protocol.outer_preconditioning_runs must be a " "nonnegative integer" ) return count def declared_inter_execution_cooldown_seconds( measurement_protocol: dict[str, Any], ) -> float: """Return the canonical delay between measured outer executions.""" seconds = measurement_protocol.get("outer_inter_execution_cooldown_seconds") if ( isinstance(seconds, bool) or not isinstance(seconds, (int, float)) or not math.isfinite(float(seconds)) or float(seconds) < 0 or float(seconds) > MAX_INTER_EXECUTION_COOLDOWN_SECONDS ): raise ValueError( "measurement_protocol.outer_inter_execution_cooldown_seconds must be " "finite, nonnegative, and no greater than " f"{MAX_INTER_EXECUTION_COOLDOWN_SECONDS:g}" ) return float(seconds) def record_outer_execution( result: dict[str, Any], execution: dict[str, Any], policy: dict[str, Any], ) -> None: """Bind process order and stabilization controls to one run record.""" execution_record = dict(execution) result["outer_process_execution"] = execution_record reproduce = dict(result.get("reproduce") or {}) reproduce["reference_sweep"] = { "outer_process_execution": execution_record, "inter_execution_cooldown": { "cli_option": "--inter-execution-cooldown-seconds", "configured_seconds": policy["configured_cooldown_seconds"], "applied_before_this_execution_seconds": execution_record[ "cooldown_before_seconds" ], "applies": policy["applies"], }, "timing_scope": policy["timing_scope"], } result["reproduce"] = reproduce def record_preconditioning_execution( result: dict[str, Any], execution: dict[str, Any], policy: dict[str, Any], ) -> None: """Bind one retained, aggregate-excluded preparation run to its policy.""" execution_record = dict(execution) result["preconditioning_execution"] = execution_record reproduce = dict(result.get("reproduce") or {}) reproduce["reference_sweep"] = { "preconditioning_execution": execution_record, "aggregate_inclusion": "excluded", "timing_scope": policy["timing_scope"], } result["reproduce"] = reproduce def validate_preconditioning( *, rows: list[dict[str, Any]], policy: dict[str, Any], measured_rows: list[dict[str, Any]], ) -> list[str]: """Fail closed unless retained preparation runs match measured execution context.""" reasons: list[str] = [] executions = policy.get("executions") or [] if len(rows) != len(executions): reasons.append( "preconditioning run count does not match its declared execution policy" ) for position, (row, execution) in enumerate(zip(rows, executions), start=1): if row.get("execution_index") != execution.get("execution_index"): reasons.append( f"preconditioning run {position} has an inconsistent execution index" ) if row.get("preconditioning_execution") != execution: reasons.append( f"preconditioning run {position} does not match its execution policy" ) if row.get("requested_seed") != execution.get("seed"): reasons.append( f"preconditioning run {position} did not use the canonical seed" ) if not row.get("evidence_valid"): details = "; ".join( row.get("invalid_reasons") or ["unknown validation failure"] ) reasons.append(f"preconditioning run {position} is invalid: {details}") fingerprints = { str(row.get("comparison_fingerprint_sha256")) for row in [*rows, *measured_rows] if _valid_sha256_hex(row.get("comparison_fingerprint_sha256")) } expected_fingerprint_count = len(rows) + len(measured_rows) valid_fingerprint_count = sum( _valid_sha256_hex(row.get("comparison_fingerprint_sha256")) for row in [*rows, *measured_rows] ) if rows and ( valid_fingerprint_count != expected_fingerprint_count or len(fingerprints) != 1 ): reasons.append( "preconditioning and measured runs must share one valid comparison " "fingerprint" ) return reasons def build_row(result: dict[str, Any]) -> dict[str, Any]: requested = result.get("requested_seed") report_seed = result.get("report_recorded_seed") manifest_seed = result.get("manifest_recorded_seed") seed_match = report_seed == requested and manifest_seed == requested return { "execution_index": result.get("execution_index"), "requested_seed": requested, "report_recorded_seed": report_seed, "manifest_recorded_seed": manifest_seed, "seed_match": seed_match, "status": result.get("status"), "primary_metric_declared": result.get("primary_metric_declared"), "primary_metric_key": result.get("primary_metric_key"), "primary_metric_value": result.get("primary_metric_value"), "quality_metric_declared": result.get("quality_metric_declared"), "quality_metric_key": result.get("quality_metric_key"), "quality_value": result.get("quality_value"), "functional_metric_declared": result.get("functional_metric_declared"), "functional_metric_key": result.get("functional_metric_key"), "functional_metric_value": result.get("functional_metric_value"), "reference_metric_role": result.get("reference_metric_role"), "quality_target_met": result.get("quality_target_met"), "quality_target": result.get("quality_target"), "quality_direction": result.get("quality_direction"), "promotion_contract": result.get("promotion_contract"), "result_role": result.get("result_role"), "comparison_fingerprint_sha256": result.get("comparison_fingerprint_sha256"), "scenario": result.get("scenario"), "manifest_scenario": result.get("manifest_scenario"), "registry_scenario": result.get("registry_scenario"), "wall_seconds": result.get("wall_seconds"), "subprocess_wall_seconds": result.get("subprocess_wall_seconds"), "backend": result.get("backend"), "device_requested": result.get("device_requested"), "device_executed": result.get("device_executed"), "hardware_backend": result.get("hardware_backend"), "fingerprint_backends": result.get("fingerprint_backends"), "chip": result.get("chip"), "data_mode": result.get("data_mode"), "report_path": result.get("report_path"), "manifest_path": result.get("manifest_path"), "manifest_verified": result.get("manifest_verified", False), "grade": result.get("grade"), "artifacts": result.get("artifacts") or [], "execution_ok": bool(result.get("execution_ok")), "evidence_valid": bool(result.get("evidence_valid")) and seed_match, "timed_out": bool(result.get("timed_out")), "invalid_reasons": list(result.get("invalid_reasons") or []), "host_power": result.get("host_power"), "outer_process_execution": result.get("outer_process_execution"), "preconditioning_execution": result.get("preconditioning_execution"), "reproduce": result.get("reproduce"), "stderr_tail": result.get("stderr_tail"), } def seed_sensitivity( rows: list[dict[str, Any]], *, value_field: str = "quality_value", metric: str | None = None, role: str = "quality", ) -> dict[str, Any]: valid = [ row for row in rows if row.get("evidence_valid") and not isinstance(row.get(value_field), bool) and isinstance(row.get(value_field), (int, float)) ] distinct = {round(float(row[value_field]), 12) for row in valid} if len(valid) < 2: verdict = "inconclusive" note = "Fewer than two valid runs; a seed-sensitivity claim cannot be made." elif len(distinct) < 2: verdict = "identical" note = "Every valid seed produced the same quality value; this is not usable variance evidence." else: verdict = "sensitive" note = f"Observed {len(distinct)} distinct quality values across {len(valid)} valid runs." return { "metric": metric, "role": role, "verdict": verdict, "distinct_quality_values": len(distinct), "valid_runs": len(valid), "note": note, } def aggregate_acceptance( aggregate_value: Any, target: Any, direction: str | None ) -> dict[str, Any]: if not isinstance(aggregate_value, (int, float)) or not isinstance( target, (int, float) ): return {"passed": False, "reason": "numeric aggregate and target are required"} if direction == "lower": passed = float(aggregate_value) <= float(target) operator = "<=" elif direction == "higher": passed = float(aggregate_value) >= float(target) operator = ">=" elif direction == "equal": passed = float(aggregate_value) == float(target) operator = "==" else: return { "passed": False, "reason": f"unsupported quality direction {direction!r}", } return { "passed": passed, "statistic": "median", "value": aggregate_value, "operator": operator, "target": target, } def score_acceptance( rows: list[dict[str, Any]], aggregate_value: Any, target: Any, direction: str | None, *, tolerance: float = 0.0, ) -> dict[str, Any]: """Require both median quality and every score-bearing run to pass.""" result = aggregate_acceptance(aggregate_value, target, direction) per_run_passes: list[bool] = [] if isinstance(target, bool) or not isinstance(target, (int, float)): result.update( { "passed": False, "all_runs_passed": False, "passed_runs": 0, "run_count": len(rows), } ) return result for row in rows: value = row.get("quality_value") numeric = ( not isinstance(value, bool) and isinstance(value, (int, float)) and math.isfinite(float(value)) ) if direction == "lower" and numeric: numeric_passed = float(value) - tolerance <= float(target) elif direction == "higher" and numeric: numeric_passed = float(value) + tolerance >= float(target) elif direction == "equal" and numeric: numeric_passed = math.isclose( float(value), float(target), rel_tol=0.0, abs_tol=tolerance ) else: numeric_passed = False per_run_passes.append( bool(numeric_passed and row.get("quality_target_met") is True) ) all_runs_passed = bool(rows) and all(per_run_passes) result.update( { "passed": result.get("passed") is True and all_runs_passed, "all_runs_passed": all_runs_passed, "passed_runs": sum(per_run_passes), "run_count": len(rows), "tolerance": tolerance, } ) return result def performance_acceptance( rows: list[dict[str, Any]], condition: str | None ) -> dict[str, Any]: """Require the functional serving gate on every reference run. Performance itself is summarized, not thresholded. This prevents a benchmark from defining a circular speed target from the same machine used to measure it. """ passed_runs = [ row for row in rows if row.get("evidence_valid") and row.get("quality_target_met") is True ] return { "passed": len(passed_runs) == len(rows) and bool(rows), "statistic": "all_runs", "value": len(passed_runs), "operator": "==", "target": len(rows), "condition": condition, "note": "Every measured run must pass the functional check; performance values are reported without a machine-derived pass threshold.", } def public_data_modes_for( workload: Any, *, mode: str | None = None, phase: str | None = None ) -> frozenset[str]: if workload.public_status == "experimental": if mode == "inference" and workload.id == "causal-language-modeling": return frozenset({"checkpoint-backed"}) contract = workload.raw.get("canonical_max_contract") or {} data_mode = contract.get("data_mode") return frozenset({str(data_mode)}) if data_mode else frozenset() if workload.public_status == "score-bearing": return SCORE_PUBLIC_DATA_MODES if workload.public_status == "performance-bearing": return PERFORMANCE_PUBLIC_DATA_MODES return frozenset() def source_snapshot() -> dict[str, Any]: def git(*args: str) -> str | None: try: return subprocess.check_output( ["git", *args], cwd=ROOT, text=True, stderr=subprocess.DEVNULL ).strip() except (subprocess.CalledProcessError, FileNotFoundError): return None status = git("status", "--porcelain") try: patch = subprocess.check_output( ["git", "diff", "--binary", "HEAD"], cwd=ROOT, stderr=subprocess.DEVNULL ) patch_sha256 = "sha256:" + hashlib.sha256(patch).hexdigest() except (subprocess.CalledProcessError, FileNotFoundError): patch_sha256 = None return { "git_sha": git("rev-parse", "HEAD"), "git_dirty": bool(status) if status is not None else None, "git_status_sha256": ( "sha256:" + hashlib.sha256(status.encode()).hexdigest() if status is not None else None ), "git_patch_sha256": patch_sha256, "tool_path": f"tools/{TOOL_NAME}", "tool_sha256": sha256_file(Path(__file__).resolve()), "python": sys.version, } def build_basis( *, workload: Any, result_role: str, selected_contract: dict[str, Any], profile: str, rows: list[dict[str, Any]], primary_aggregate: dict[str, Any], primary_metric_name: str | None, quality_aggregate: dict[str, Any] | None, quality_metric_name: str | None, dataset_mode: Any, eligible: bool, evidence_tier: str, ) -> dict[str, Any]: performance = result_role == "performance-bearing" functional = dict(workload.raw.get("functional_check") or {}) if performance: selected_quality = selected_contract.get("quality") or {} functional = { "metric": selected_quality.get("metric"), "metric_key": selected_quality.get("metric_key"), "condition": "Every run must pass the canonical functional gate.", } functional_targets = { json.dumps((row.get("grade") or {}).get("target"), sort_keys=True) for row in rows if (row.get("grade") or {}).get("target") is not None } functional_target = None if len(functional_targets) == 1: functional_target = json.loads(next(iter(functional_targets))) if performance: functional["target"] = functional_target variance_aggregate = primary_aggregate if performance else quality_aggregate or {} variance_metric = primary_metric_name if performance else quality_metric_name quality_targets = { float(row["quality_target"]) for row in rows if not isinstance(row.get("quality_target"), bool) and isinstance(row.get("quality_target"), (int, float)) } quality_directions = { str(row["quality_direction"]) for row in rows if row.get("quality_direction") } return { "result_role": result_role, "eligible_for_public_baseline": eligible and workload.public_status in PUBLIC_STATUSES, "eligible_for_promotion": eligible and evidence_tier == "promotion-candidate", "primary_metric": { "name": primary_metric_name, "role": "performance", "aggregate": primary_aggregate, }, "variance_summary": { "runs": len([row for row in rows if row.get("evidence_valid")]), "statistic": "median", "metric": variance_metric, **variance_aggregate, }, "reference_protocol": { "profile": profile, "seeds": [row.get("requested_seed") for row in rows], "seed_interface": "MLPERF_EDU_MAX_SEED", "dataset_mode": dataset_mode, "observed_data_modes": sorted( {str(row.get("data_mode")) for row in rows if row.get("data_mode")} ), "rerun_policy": "If any run fails or times out, create a new full-sweep attempt; never replace one seed in an existing attempt.", "artifact_policy": "Preserve and SHA-256 index every report, provenance manifest, checkpoint, and runner-declared artifact.", "generated_by": TOOL_ID, }, "quality_target": ( None if performance else { "metric": quality_metric_name, "target": next(iter(quality_targets)) if len(quality_targets) == 1 else None, "direction": next(iter(quality_directions)) if len(quality_directions) == 1 else None, "tolerance": getattr(workload, "quality_tolerance", None), "all_runs_must_pass": True, } ), "functional_check": functional, } def _declared_protocol(workload: Any) -> dict[str, Any]: if workload.public_status == "performance-bearing": protocol = workload.raw.get("performance_reference_protocol") return protocol if isinstance(protocol, dict) else {} protocol = getattr(workload, "quality_reference_protocol", None) return protocol if isinstance(protocol, dict) else {} def _valid_sha256_hex(value: object) -> bool: return ( isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value) ) def primary_metric_repeatability( primary_aggregate: dict[str, Any], protocol: dict[str, Any] ) -> dict[str, Any]: """Recompute timing repeatability from the five-run primary metric.""" mean = primary_aggregate.get("mean") stdev = primary_aggregate.get("stdev") raw_limit = protocol.get("repeatability_limit") limit = ( float(raw_limit) if not isinstance(raw_limit, bool) and isinstance(raw_limit, (int, float)) and math.isfinite(float(raw_limit)) else None ) coefficient_of_variation = ( float(stdev) / float(mean) if not isinstance(mean, bool) and isinstance(mean, (int, float)) and math.isfinite(float(mean)) and float(mean) > 0 and not isinstance(stdev, bool) and isinstance(stdev, (int, float)) and math.isfinite(float(stdev)) else None ) return { "metric": protocol.get("repeatability_metric"), "coefficient_of_variation": coefficient_of_variation, "limit": limit, "passed": coefficient_of_variation is not None and limit is not None and coefficient_of_variation <= limit, } def validate_sweep( *, workload: Any, result_role: str, seeds: list[int], mode: str | None, phase: str | None, rows: list[dict[str, Any]], sensitivity: dict[str, Any], acceptance: dict[str, Any], evidence_tier: str, ) -> list[str]: reasons: list[str] = [] selected_contract = execution_contract(workload, mode=mode, phase=phase) for row in rows: host_power = row.get("host_power") if not isinstance(host_power, dict) or host_power.get("stable") is not True: reasons.append( f"seed {row.get('requested_seed')}: stable host power record is missing" ) elif evidence_tier in {"public-candidate", "promotion-candidate"}: if host_power.get("promotion_conditions_required") is not True: reasons.append( f"seed {row.get('requested_seed')}: promotion power conditions were not required" ) for boundary in ("before", "after"): snapshot = host_power.get(boundary) or {} if snapshot.get("supported") is not True: reasons.append( f"seed {row.get('requested_seed')}: {boundary} power monitoring is unsupported" ) if snapshot.get("source") != "external": reasons.append( f"seed {row.get('requested_seed')}: {boundary} power source is not external" ) if snapshot.get("low_power_mode") is not False: reasons.append( f"seed {row.get('requested_seed')}: {boundary} Low Power Mode is not disabled" ) if not row.get("evidence_valid"): details = "; ".join( row.get("invalid_reasons") or ["unknown validation failure"] ) reasons.append(f"seed {row.get('requested_seed')}: {details}") if row.get("result_role") != result_role: reasons.append( f"seed {row.get('requested_seed')}: promotion result role " f"{row.get('result_role')!r} does not match {result_role!r}" ) expected_primary = selected_contract["measurement_protocol"].get( "primary_metric" ) if row.get("primary_metric_declared") != expected_primary: reasons.append( f"seed {row.get('requested_seed')}: primary metric declaration " f"{row.get('primary_metric_declared')!r} does not match " f"{expected_primary!r}" ) if row.get("reference_metric_role") != "performance": reasons.append( f"seed {row.get('requested_seed')}: primary metric role must be performance" ) primary_value = row.get("primary_metric_value") if ( isinstance(primary_value, bool) or not isinstance(primary_value, (int, float)) or not math.isfinite(float(primary_value)) or float(primary_value) <= 0 ): reasons.append( f"seed {row.get('requested_seed')}: primary metric value is not " "finite and positive" ) if result_role != "performance-bearing": if not row.get("quality_metric_declared"): reasons.append( f"run {row.get('execution_index')}: quality metric declaration is missing" ) quality_value = row.get("quality_value") if ( isinstance(quality_value, bool) or not isinstance(quality_value, (int, float)) or not math.isfinite(float(quality_value)) ): reasons.append( f"seed {row.get('requested_seed')}: quality metric value is not finite" ) else: expected_functional = (selected_contract.get("quality") or {}).get("metric") if row.get("functional_metric_declared") != expected_functional: reasons.append( f"seed {row.get('requested_seed')}: functional metric declaration " "does not match the registry" ) functional_value = row.get("functional_metric_value") if ( isinstance(functional_value, bool) or not isinstance(functional_value, (int, float)) or not math.isfinite(float(functional_value)) ): reasons.append( f"seed {row.get('requested_seed')}: functional metric value is not finite" ) primary_metric_keys = { row.get("primary_metric_key") for row in rows if row.get("primary_metric_key") } if len(primary_metric_keys) != 1: reasons.append( "runs did not resolve to exactly one primary metric key: " f"{sorted(primary_metric_keys)}" ) gate_field = "quality_metric_key" gate_metric_keys = {row.get(gate_field) for row in rows if row.get(gate_field)} if len(gate_metric_keys) != 1: reasons.append( f"runs did not resolve to exactly one {gate_field}: " f"{sorted(gate_metric_keys)}" ) if not acceptance.get("passed"): reasons.append( "quality or functional acceptance failed: " f"{acceptance.get('reason') or acceptance}" ) if evidence_tier in {"public-candidate", "promotion-candidate"}: protocol = execution_contract(workload, mode=mode, phase=phase)[ "measurement_protocol" ] declared_runs = protocol.get("outer_reference_runs") if isinstance(declared_runs, int) and len(seeds) != declared_runs: reasons.append( f"registry requires {declared_runs} reference runs, received {len(seeds)}" ) expected_seed = canonical_seed(workload) if any(seed != expected_seed for seed in seeds): reasons.append( f"promotion evidence must repeat canonical seed {expected_seed}, received {seeds}" ) allowed_modes = public_data_modes_for(workload, mode=mode, phase=phase) invalid_modes = sorted( { str(row.get("data_mode")) for row in rows if row.get("data_mode") not in allowed_modes } ) if invalid_modes: reasons.append( f"public candidates allow {sorted(allowed_modes)} for {result_role}, " f"observed {invalid_modes}" ) comparison_fingerprints = [ row.get("comparison_fingerprint_sha256") for row in rows ] malformed_fingerprints = [ value for value in comparison_fingerprints if not _valid_sha256_hex(value) ] distinct_fingerprints = { str(value) for value in comparison_fingerprints if _valid_sha256_hex(value) } if malformed_fingerprints or len(distinct_fingerprints) != 1: reasons.append( "public-candidate runs must have exactly one identical, valid " "comparison_fingerprint_sha256" ) return reasons def write_evidence_summary( attempt_dir: Path, artifact: dict[str, Any] ) -> tuple[Path, Path, str]: artifact_path = attempt_dir / "evidence_summary.json" payload = (json.dumps(artifact, indent=2, sort_keys=True) + "\n").encode("utf-8") with artifact_path.open("xb") as handle: handle.write(payload) digest = hashlib.sha256(payload).hexdigest() digest_path = artifact_path.with_suffix(".json.sha256") with digest_path.open("x") as handle: handle.write(f"{digest} {artifact_path.name}\n") return artifact_path, digest_path, digest def print_summary(rows: list[dict[str, Any]], artifact_path: Path, valid: bool) -> None: print("seed valid primary metric value data_mode") print("---- ----- --------------------- ---------- -----------------") for row in rows: print( f"{str(row.get('requested_seed')):>4} " f"{str(bool(row.get('evidence_valid'))):<5} " f"{str(row.get('primary_metric_key') or '-'):21.21} " f"{str(row.get('primary_metric_value')):10.10} " f"{str(row.get('data_mode') or '-')}" ) print(f"evidence status: {'VALID' if valid else 'INVALID'}") print(f"evidence summary: {artifact_path}") def _uses_nanogpt_training_lineage(workload: Any, *, mode: str | None) -> bool: return workload.id == "causal-language-modeling" and mode == "inference" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog=TOOL_NAME, description="Run a verified multi-seed reference sweep for one MLPerf EDU workload.", ) parser.add_argument("--workload", required=True) parser.add_argument("--variant") parser.add_argument("--profile", default="max") parser.add_argument("--mode", choices=("training", "inference")) parser.add_argument("--phase", choices=("full", "prefill", "decode")) parser.add_argument("--runs", type=parse_run_count, default=5) parser.add_argument( "--preconditioning-runs", type=parse_preconditioning_run_count, default=None, help=( "development-only override for complete aggregate-excluded preparation " "executions; promotion and public candidates use the count declared by " "the measurement protocol" ), ) parser.add_argument( "--seeds", type=parse_seeds, default=None, help="development-only seed list; promotion evidence repeats the canonical seed", ) parser.add_argument( "--device", choices=("cpu", "mps", "default"), default="default" ) parser.add_argument( "--output-dir", default=str(DEFAULT_OUTPUT_DIR), help=f"create-once evidence root (default: {DEFAULT_OUTPUT_DIR})", ) parser.add_argument( "--timeout-seconds", type=float, default=DEFAULT_TIMEOUT_SECONDS, help=f"per-seed timeout (default: {DEFAULT_TIMEOUT_SECONDS:g})", ) parser.add_argument( "--inter-execution-cooldown-seconds", type=float, default=None, help=( "development-only override for the fixed delay between measured fresh " "processes; promotion and public candidates use the measurement-protocol " f"value (maximum: {MAX_INTER_EXECUTION_COOLDOWN_SECONDS:g})" ), ) parser.add_argument( "--evidence-tier", choices=("auto", "promotion-candidate", "public-candidate", "development"), default="auto", ) parser.add_argument( "--nanogpt-lineage-package", help=( "verified mlperf-edu-package/0.2 archive from a passing real-data " "causal-language-modeling training max run; required for promotion " "or public-candidate NanoGPT inference" ), ) args = parser.parse_args(argv) if not math.isfinite(args.timeout_seconds) or args.timeout_seconds <= 0: parser.error("--timeout-seconds must be a finite positive number") if args.inter_execution_cooldown_seconds is not None and ( not math.isfinite(args.inter_execution_cooldown_seconds) or args.inter_execution_cooldown_seconds < 0 or args.inter_execution_cooldown_seconds > MAX_INTER_EXECUTION_COOLDOWN_SECONDS ): parser.error( "--inter-execution-cooldown-seconds must be finite, nonnegative, " f"and no greater than {MAX_INTER_EXECUTION_COOLDOWN_SECONDS:g}" ) try: from mlperf.edu_cli import load_runner from mlperf.registry import load_registry except Exception as exc: print(f"error: cannot import MLPerf EDU: {exc}", file=sys.stderr) return 2 registry = load_registry() workload = None if args.variant: for candidate in registry.values(): if ( getattr(candidate, "canonical_workload", None) == args.workload and getattr(candidate, "variant", None) == args.variant ): workload = candidate break else: workload = registry.get(args.workload) if workload is None: print( f"error: workload/variant not found: {args.workload}/{args.variant}", file=sys.stderr, ) return 2 try: selected_contract = execution_contract( workload, mode=args.mode, phase=args.phase ) result_role = execution_result_role(workload, mode=args.mode, phase=args.phase) except ValueError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if args.phase and args.mode != "inference": print("error: --phase requires --mode inference", file=sys.stderr) return 2 if load_runner(workload, args.profile) is None: print(f"error: no {args.profile!r} runner for {workload.id}", file=sys.stderr) return 2 evidence_tier = args.evidence_tier if evidence_tier == "auto": evidence_tier = ( "public-candidate" if workload.public_status in PUBLIC_STATUSES else "promotion-candidate" ) try: protocol_preconditioning_runs = declared_preconditioning_runs( selected_contract["measurement_protocol"] ) protocol_cooldown_seconds = declared_inter_execution_cooldown_seconds( selected_contract["measurement_protocol"] ) except ValueError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if ( evidence_tier in {"public-candidate", "promotion-candidate"} and args.preconditioning_runs is not None and args.preconditioning_runs != protocol_preconditioning_runs ): print( "error: promotion and public candidates must use " "measurement_protocol.outer_preconditioning_runs " f"({protocol_preconditioning_runs})", file=sys.stderr, ) return 2 preconditioning_run_count = ( args.preconditioning_runs if args.preconditioning_runs is not None else protocol_preconditioning_runs ) if ( evidence_tier in {"public-candidate", "promotion-candidate"} and args.inter_execution_cooldown_seconds is not None and args.inter_execution_cooldown_seconds != protocol_cooldown_seconds ): print( "error: promotion and public candidates must use " "measurement_protocol.outer_inter_execution_cooldown_seconds " f"({protocol_cooldown_seconds:g})", file=sys.stderr, ) return 2 inter_execution_cooldown_seconds = ( args.inter_execution_cooldown_seconds if args.inter_execution_cooldown_seconds is not None else protocol_cooldown_seconds ) requested_seeds = args.seeds or [canonical_seed(workload)] * args.runs preconditioning_policy = build_preconditioning_policy( seed=canonical_seed(workload), execution_count=preconditioning_run_count, ) outer_execution_policy = build_outer_execution_policy( public_status=workload.public_status, evidence_tier=evidence_tier, seeds=requested_seeds, configured_cooldown_seconds=inter_execution_cooldown_seconds, ) uses_nanogpt_lineage = _uses_nanogpt_training_lineage(workload, mode=args.mode) lineage_required = ( evidence_tier in {"public-candidate", "promotion-candidate"} and uses_nanogpt_lineage ) if args.nanogpt_lineage_package and not uses_nanogpt_lineage: print( "error: --nanogpt-lineage-package is only valid for a workload that " "is causal-language-modeling inference", file=sys.stderr, ) return 2 if lineage_required and not args.nanogpt_lineage_package: print( "error: public-candidate NanoGPT inference requires " "--nanogpt-lineage-package from a passing real-data " "causal-language-modeling training max run", file=sys.stderr, ) return 2 lineage_validation: dict[str, Any] | None = None if args.nanogpt_lineage_package: try: lineage_validation = validate_nanogpt_lineage_package( Path(args.nanogpt_lineage_package) ) except LineagePackageError as exc: print(f"error: invalid NanoGPT lineage package: {exc}", file=sys.stderr) return 2 allowed_data_modes = public_data_modes_for( workload, mode=args.mode, phase=args.phase ) device = None if args.device == "default" else args.device # Bind source state before creating evidence. An explicitly selected output # directory may live inside the checkout; its artifacts are not source edits # and must not make an otherwise clean run reject itself. source = source_snapshot() output_root = Path(args.output_dir).expanduser().resolve() output_root.mkdir(parents=True, exist_ok=True) started = datetime.now(timezone.utc) evidence_id = ( f"{workload.id}_{args.profile}_{started.strftime('%Y%m%dT%H%M%S.%fZ')}" ) attempt_dir = output_root / evidence_id attempt_dir.mkdir(exist_ok=False) lineage_stage: dict[str, Any] | None = None if lineage_validation is not None: try: lineage_stage = stage_nanogpt_lineage_package( lineage_validation, attempt_dir ) except (LineagePackageError, OSError, zipfile.BadZipFile) as exc: shutil.rmtree(attempt_dir, ignore_errors=True) print( f"error: could not stage NanoGPT lineage package: {exc}", file=sys.stderr, ) return 2 print( f"{TOOL_ID}: workload={workload.id} profile={args.profile} " f"mode={args.mode or 'default'} phase={args.phase or 'default'} " f"runs={len(requested_seeds)} canonical_seed={canonical_seed(workload)} " f"tier={evidence_tier} timeout={args.timeout_seconds:g}s " f"preconditioning-runs={preconditioning_run_count} " "inter-execution-cooldown=" f"{inter_execution_cooldown_seconds:g}s " f"applies={outer_execution_policy['applies']}" ) preconditioning_rows: list[dict[str, Any]] = [] rows: list[dict[str, Any]] = [] with tempfile.TemporaryDirectory(prefix="mlperf-edu-sweep-bootstrap-") as tmp: bootstrap_path = Path(tmp) / "child.py" bootstrap_path.write_text(_CHILD_BOOTSTRAP) preconditioning_failed = False for preconditioning_execution in preconditioning_policy["executions"]: seed = preconditioning_execution["seed"] execution_index = preconditioning_execution["execution_index"] print( f"running preconditioning {execution_index} (seed {seed}) ...", flush=True, ) result = run_one_seed( bootstrap_path, workload_id=args.workload, variant=args.variant, profile=args.profile, seed=seed, execution_index=execution_index, mode=args.mode, phase=args.phase, device=device, attempt_dir=attempt_dir, timeout_seconds=args.timeout_seconds, evidence_tier=evidence_tier, allowed_data_modes=allowed_data_modes, environment_overrides=( lineage_stage["environment"] if lineage_stage else None ), run_group="preconditioning", ) record_preconditioning_execution( result, preconditioning_execution, preconditioning_policy ) preconditioning_row = build_row(result) preconditioning_rows.append(preconditioning_row) if not preconditioning_row["evidence_valid"]: preconditioning_failed = True print( "preconditioning failed; measured repetitions are skipped", flush=True, ) break if not preconditioning_failed: for process_execution in outer_execution_policy["executions"]: seed = process_execution["seed"] execution_index = process_execution["execution_index"] print( f"running repetition {execution_index} (seed {seed}) ...", flush=True, ) result = run_one_seed( bootstrap_path, workload_id=args.workload, variant=args.variant, profile=args.profile, seed=seed, execution_index=execution_index, mode=args.mode, phase=args.phase, device=device, attempt_dir=attempt_dir, timeout_seconds=args.timeout_seconds, evidence_tier=evidence_tier, allowed_data_modes=allowed_data_modes, environment_overrides=( lineage_stage["environment"] if lineage_stage else None ), cooldown_before_seconds=process_execution[ "cooldown_before_seconds" ], ) record_outer_execution( result, process_execution, outer_execution_policy ) rows.append(build_row(result)) primary_metric_name = next( ( row.get("primary_metric_declared") for row in rows if row.get("primary_metric_declared") ), selected_contract["measurement_protocol"].get("primary_metric"), ) quality_metric_name = ( next( ( row.get("quality_metric_declared") for row in rows if row.get("quality_metric_declared") ), getattr(workload, "quality_metric", None), ) if result_role != "performance-bearing" else None ) primary_values = [ float(row["primary_metric_value"]) for row in rows if row.get("evidence_valid") and not isinstance(row.get("primary_metric_value"), bool) and isinstance(row.get("primary_metric_value"), (int, float)) ] quality_values = [ float(row["quality_value"]) for row in rows if row.get("evidence_valid") and not isinstance(row.get("quality_value"), bool) and isinstance(row.get("quality_value"), (int, float)) ] wall_values = [ float(row["wall_seconds"]) for row in rows if row.get("execution_ok") and isinstance(row.get("wall_seconds"), (int, float)) ] primary_aggregate = aggregate(primary_values) quality_aggregate = ( aggregate(quality_values) if result_role != "performance-bearing" else None ) wall_aggregate = aggregate(wall_values) sensitivity = ( seed_sensitivity( rows, value_field="primary_metric_value", metric=primary_metric_name, role="performance", ) if result_role == "performance-bearing" else seed_sensitivity( rows, value_field="quality_value", metric=quality_metric_name, role="quality", ) ) if result_role == "performance-bearing": acceptance = performance_acceptance( rows, "Every run must pass the canonical functional gate." ) else: observed_targets = { float(row["quality_target"]) for row in rows if not isinstance(row.get("quality_target"), bool) and isinstance(row.get("quality_target"), (int, float)) } observed_directions = { str(row["quality_direction"]) for row in rows if row.get("quality_direction") } quality_target = ( next(iter(observed_targets)) if len(observed_targets) == 1 else None ) quality_direction = ( next(iter(observed_directions)) if len(observed_directions) == 1 else None ) acceptance = score_acceptance( rows, (quality_aggregate or {}).get("median"), quality_target, quality_direction, tolerance=float(getattr(workload, "quality_tolerance", None) or 0.0), ) protocol = selected_contract["measurement_protocol"] invalid_reasons = validate_sweep( workload=workload, result_role=result_role, seeds=requested_seeds, mode=args.mode, phase=args.phase, rows=rows, sensitivity=sensitivity, acceptance=acceptance, evidence_tier=evidence_tier, ) invalid_reasons.extend( validate_preconditioning( rows=preconditioning_rows, policy=preconditioning_policy, measured_rows=rows, ) ) primary_repeatability = primary_metric_repeatability(primary_aggregate, protocol) if evidence_tier in {"public-candidate", "promotion-candidate"}: repeatability_limit = primary_repeatability.get("limit") if repeatability_limit != PUBLIC_PRIMARY_METRIC_CV_LIMIT: invalid_reasons.append( "public protocol must declare a primary performance coefficient-of-" f"variation limit of {PUBLIC_PRIMARY_METRIC_CV_LIMIT:g}" ) if not primary_repeatability.get("metric"): invalid_reasons.append( "public protocol must name its primary performance repeatability metric" ) if primary_repeatability["passed"] is not True: invalid_reasons.append( "primary performance repeatability exceeds the declared " f"coefficient-of-variation limit {repeatability_limit!r}" ) if ( evidence_tier in {"public-candidate", "promotion-candidate"} and source.get("git_dirty") is not False ): invalid_reasons.append( "public reference evidence must be produced from a clean Git worktree" ) eligible = not invalid_reasons dataset_mode = protocol.get("dataset_mode") finished = datetime.now(timezone.utc) lineage_summary = None if uses_nanogpt_lineage: lineage_summary = { "required": lineage_required, "status": "staged" if lineage_stage else "not-supplied", } if lineage_stage: lineage_summary.update( { "package_schema": lineage_stage["package_schema"], "package_sha256": lineage_stage["package_sha256"], "source_workload": lineage_stage["source_workload"], "verification_check_count": lineage_stage[ "verification_check_count" ], "staged_root": _relative_to_attempt( str(lineage_stage["stage_root"]), attempt_dir ), "source_training_report": _relative_to_attempt( str(lineage_stage["paths"]["report"]), attempt_dir ), "source_training_manifest": _relative_to_attempt( str(lineage_stage["paths"]["manifest"]), attempt_dir ), "source_training_checkpoint": _relative_to_attempt( str(lineage_stage["paths"]["checkpoint"]), attempt_dir ), } ) basis = build_basis( workload=workload, result_role=result_role, selected_contract=selected_contract, profile=args.profile, rows=rows, primary_aggregate=primary_aggregate, primary_metric_name=primary_metric_name, quality_aggregate=quality_aggregate, quality_metric_name=quality_metric_name, dataset_mode=dataset_mode, eligible=eligible, evidence_tier=evidence_tier, ) comparison_fingerprints = { str(row.get("comparison_fingerprint_sha256")) for row in rows if _valid_sha256_hex(row.get("comparison_fingerprint_sha256")) } comparison_fingerprint_sha256 = ( next(iter(comparison_fingerprints)) if len(comparison_fingerprints) == 1 else None ) resolved_variant = getattr(workload, "variant", None) artifact = { "schema": REFERENCE_EVIDENCE_SCHEMA, "evidence_id": evidence_id, "status": "valid" if eligible else "invalid", "eligible_for_promotion": eligible and evidence_tier == "promotion-candidate", "eligible_for_public_baseline": eligible and evidence_tier == "public-candidate" and workload.public_status in PUBLIC_STATUSES, "invalid_reasons": invalid_reasons, "tool": {"id": TOOL_ID, "version": TOOL_VERSION}, "generated_at": started.isoformat(), "finished_at": finished.isoformat(), "duration_seconds": (finished - started).total_seconds(), "write_policy": "create-once attempt directory; this tool never overwrites or edits prior evidence", "digest_policy": "The adjacent SHA-256 sidecar is an unauthenticated integrity digest, not a signature.", "rerun_policy": { "mode": "full-sweep-only", "rule": "If any preconditioning or evidence execution fails or times out, create a new attempt and rerun the complete declared protocol. Never replace an individual execution in an existing attempt.", }, "workload": workload.id, "canonical_workload": getattr(workload, "canonical_workload", None) or workload.id, "variant": resolved_variant, "profile": args.profile, "mode": args.mode, "phase": args.phase, "public_status": workload.public_status, "result_role": result_role, "evidence_tier": evidence_tier, "device_requested": args.device, "timeout_seconds_per_run": args.timeout_seconds, "power_stability_policy": dict(POWER_STABILITY_POLICY), "preconditioning": { **preconditioning_policy, "runs": preconditioning_rows, }, "inter_execution_stabilization": outer_execution_policy, "canonical_seed": canonical_seed(workload), "seeds_requested": requested_seeds, "dataset_mode_declared": dataset_mode, "allowed_public_data_modes": sorted(allowed_data_modes), "reference_metric_role": "performance", "primary_metric": { "name": primary_metric_name, "role": "performance", }, "quality_metric": quality_metric_name, "quality_target": ( None if result_role == "performance-bearing" else (basis.get("quality_target") or {}).get("target") ), "quality_direction": (basis.get("quality_target") or {}).get("direction"), "quality_gate": ( basis["quality_target"] if result_role != "performance-bearing" else None ), "functional_gate": ( basis["functional_check"] if result_role == "performance-bearing" else None ), "comparison_fingerprint_sha256": comparison_fingerprint_sha256, "runs": rows, "aggregate": { "primary_metric": primary_aggregate, "quality": quality_aggregate, "wall_seconds": wall_aggregate, }, "primary_metric_repeatability": primary_repeatability, # Retain the established performance-only field alongside primary timing # repeatability for score-bearing training. "repeatability": ( primary_repeatability if result_role == "performance-bearing" else None ), "seed_sensitivity": sensitivity, "acceptance": acceptance, "basis": basis, "source": source, "nanogpt_training_lineage": lineage_summary, } artifact_path, _, _ = write_evidence_summary(attempt_dir, artifact) print_summary(rows, artifact_path, eligible) return 0 if eligible else 1 if __name__ == "__main__": raise SystemExit(main())