tests/test_reliability_gates.py

"""Reliability-target gate tests — machine-executable counterpart of the design review checklist (`pytest tests/test_reliability_gates.py -v` or `pytest -m gate -v`); each test class name states the guarantee it protects and fills gaps left by the per-module test files (MoE param split, feature vector parity, schema/predictor agreement, model disk size, notebook FEATURE_COLS parity, training-data disclosure completeness, precision-support enforcement — see each class's own docstring for specifics)."""

from __future__ import annotations

import ast
import json
import math
from pathlib import Path
from typing import get_args

import pandas as pd
import pytest
import yaml

from src.api.schemas import AccuracyTier, Framework, MemoryFitVerdict, RankingObjective, Scenario
from src.data import manifest
from src.data.gpu_spec_db import load_specs
from src.features.build_features import (
    BYTES_PER_PARAM,
    MEMORY_OVERHEAD_FACTOR,
    MODEL_PARAMS,
    TIER_TO_PRECISION,
    VALID_MEMORY_FIT_VERDICTS,
    _FRAMEWORK_PATTERNS,
    _normalize_framework,
    build_training_df,
    cost_per_million_tokens,
    gpu_supports_precision,
    memory_fit_verdict,
)
from src.models.predictor import (
    FEATURE_COLS,
    VALID_FRAMEWORKS,
    VALID_SCENARIOS,
    VALID_TIERS,
    GpuPredictor,
    _build_feature_vector,
    _selected_precision,
)
from src.recommend.recommender import (
    GpuRecommender,
    VALID_RANKING_OBJECTIVES,
    _pareto_frontier,
    _RANKING_FIELDS,
)


# Helpers

def _nan_match(a: float, b: float) -> bool:
    """True when both are NaN, or both are finite and within 1e-3 of each other."""
    a_nan = math.isnan(a)
    b_nan = math.isnan(b)
    if a_nan and b_nan:
        return True
    if a_nan or b_nan:
        return False
    return abs(a - b) < 1e-3


def _make_single_row(
    gpu_alias: str,
    model_name: str,
    scenario: str,
    tier: str,
    raw_framework: str,
    throughput: float = 1000.0,
) -> pd.DataFrame:
    """One synthetic MLPerf row with every field build_training_df expects."""
    return pd.DataFrame([{
        "round":                       "v5.0",
        "division":                    "closed",
        "submitter":                   "test",
        "system_name":                 "test_system",
        "gpu_name":                    gpu_alias,
        "num_gpus":                    1,
        "vram_gb":                     None,
        "framework":                   raw_framework,
        "system_type":                 "datacenter",
        "hw_status":                   "available",
        "benchmark":                   f"{model_name}-{tier}",
        "benchmark_base":              model_name,
        "benchmark_accuracy_tier":     tier,
        "scenario":                    scenario,
        "precision":                   None,
        "tokens_per_sample":           294,
        "throughput_tokens_per_sec":   throughput,
        "throughput_tok_per_sec_per_gpu": throughput,
        "result_valid":                True,
        "throughput_samples_per_sec":  throughput / 294,
        "latency_mean_ms":             None,
        "latency_p99_ms":              None,
        "ttft_mean_ms":                None,
        "ttft_p99_ms":                 None,
        "tpot_mean_ms":                None,
        "tpot_p99_ms":                 None,
        "log_path":                    "fake/path",
    }])


# MoE param split — bandwidth ceiling uses total, compute uses active

@pytest.mark.gate
class TestMoEParamSplitGate:
    """GATE: MoE bandwidth ceiling must use total_params_b, compute ceiling must use compute_params_b — swapping these args in roofline_ceilings() would overestimate mixtral-8x7b compute throughput 3.31x (total=46.7B vs active=14.1B) and underestimate bandwidth by the same factor."""

    @pytest.fixture(scope="class")
    def mi300x_spec(self) -> dict:
        return {s["id"]: s for s in load_specs()}["mi300x"]

    def test_mixtral_total_ne_active_params(self) -> None:
        total_b, active_b = MODEL_PARAMS["mixtral-8x7b"]
        assert total_b != active_b, "mixtral-8x7b must have distinct total and active params"
        assert total_b == pytest.approx(46.7, abs=0.1)
        assert active_b == pytest.approx(14.1, abs=0.1)

    def test_bw_ceiling_uses_total_params(self, mi300x_spec: dict) -> None:
        """Back-calculate which param count produced the BW ceiling in the serving path."""
        total_b, active_b = MODEL_PARAMS["mixtral-8x7b"]
        precision = TIER_TO_PRECISION["99"]   # fp8
        bpp = BYTES_PER_PARAM[precision]
        hbm_bw = mi300x_spec["hbm_bandwidth_tbps"]

        feats, _, _ = _build_feature_vector(
            gpu_spec=mi300x_spec,
            model_name="mixtral-8x7b",
            scenario="Offline",
            accuracy_tier="99",
            framework="vllm",
        )
        bw_ceil = feats[FEATURE_COLS.index("bandwidth_ceiling_tok_per_sec")]

        # bw_ceil = (hbm_bw * 1e12) / (params * 1e9 * bpp)  ⟹  params = …
        implied_params = (hbm_bw * 1e12) / (bw_ceil * 1e9 * bpp)
        assert implied_params == pytest.approx(total_b, rel=1e-3), (
            f"BW ceiling implies {implied_params:.2f}B params in model footprint; "
            f"expected total_params_b={total_b}B.  "
            f"Compute ceiling likely received total params instead of active params."
        )

    def test_compute_ceiling_uses_active_params(self, mi300x_spec: dict) -> None:
        """Back-calculate which param count produced the compute ceiling in the serving path."""
        total_b, active_b = MODEL_PARAMS["mixtral-8x7b"]
        precision = TIER_TO_PRECISION["99"]
        pt = mi300x_spec.get("peak_tflops") or {}
        peak_tflops = pt.get(precision) or pt.get("fp16")

        feats, _, _ = _build_feature_vector(
            gpu_spec=mi300x_spec,
            model_name="mixtral-8x7b",
            scenario="Offline",
            accuracy_tier="99",
            framework="vllm",
        )
        compute_ceil = feats[FEATURE_COLS.index("compute_ceiling_tok_per_sec")]

        # compute_ceil = (peak_tflops * 1e12) / (2 * params * 1e9)  ⟹  params = …
        implied_params = (peak_tflops * 1e12) / (2.0 * compute_ceil * 1e9)
        assert implied_params == pytest.approx(active_b, rel=1e-3), (
            f"Compute ceiling implies {implied_params:.2f}B active params; "
            f"expected compute_params_b={active_b}B.  "
            f"Bandwidth ceiling likely received active params instead of total params."
        )


# Memory-fit verdict enforcement — does_not_fit excluded, tight kept

@pytest.mark.gate
class TestMemoryFitVerdictEnforcementGate:
    """GATE: recommend()'s hard memory-fit exclusion must track the tri-state verdict exactly — every 'does_not_fit' lands in `filtered` (never frontier/dominated) and 'tight'/'fits' are never silently dropped; drift (e.g. reusing the old boolean vram_fits check) would recommend an unrunnable config or wrongly exclude a runnable one."""

    @pytest.fixture(scope="class")
    def recommender(self) -> GpuRecommender:
        return GpuRecommender(GpuPredictor())

    def test_sweep_all_models_verdict_matches_bucket(self, recommender) -> None:
        checked_any = False
        for model_name in MODEL_PARAMS:
            for tier in get_args(AccuracyTier):
                # No budget/throughput constraint is passed anywhere in this sweep, so every filtered entry can only be a memory-fit rejection; budget/throughput filtering is covered separately in test_recommender.py.
                result = recommender.recommend(model_name=model_name, accuracy_tier=tier)
                for f in result["filtered"]:
                    checked_any = True
                    assert f["memory_fit_verdict"] == "does_not_fit", (
                        f"{model_name}/{tier}/{f['gpu_id']} is filtered with verdict "
                        f"{f['memory_fit_verdict']!r}, but this sweep sets no budget/throughput "
                        "constraint, so the only valid filter reason is does_not_fit"
                    )
                for c in result["frontier"] + result["dominated"]:
                    checked_any = True
                    assert c["memory_fit_verdict"] != "does_not_fit", (
                        f"{model_name}/{tier}/{c['gpu_id']} is a candidate but "
                        f"verdict is does_not_fit — should have been excluded"
                    )
        assert checked_any, "no result entries were produced to check"


# Precision-support enforcement — never silently substituted

@pytest.mark.gate
class TestPrecisionSupportEnforcementGate:
    """GATE: predict()/predict_batch()/recommend() must never return a normal-looking result for a (GPU, tier) whose selected precision has no native peak_tflops entry — before this existed, a100_sxm_80gb@99 (fp8, unsupported on Ampere) silently substituted fp16 TFLOPS for the roofline ceiling while still using fp8 bytes-per-param, surfacing as a real "dominated" recommend() candidate with no error or flag."""

    @pytest.fixture(scope="class")
    def predictor(self) -> GpuPredictor:
        return GpuPredictor()

    @pytest.fixture(scope="class")
    def recommender(self, predictor) -> GpuRecommender:
        return GpuRecommender(predictor)

    def test_predict_raises_iff_precision_unsupported(self, predictor) -> None:
        specs = {s["id"]: s for s in load_specs() if s.get("in_model_scope")}
        checked_any = False
        checked_unsupported = False
        for gpu_id, spec in specs.items():
            for tier in get_args(AccuracyTier):
                selected = _selected_precision(spec, tier)
                supported = gpu_supports_precision(spec, selected)
                checked_any = True
                if supported:
                    result = predictor.predict(
                        gpu_id=gpu_id, model_name="gptj", accuracy_tier=tier
                    )
                    assert result["pred_throughput_tok_per_sec"] >= 0.0, (
                        f"{gpu_id}/{tier} (selected={selected}, supported) "
                        "should predict normally"
                    )
                else:
                    checked_unsupported = True
                    with pytest.raises(ValueError, match="does not support precision"):
                        predictor.predict(
                            gpu_id=gpu_id, model_name="gptj", accuracy_tier=tier
                        )
        assert checked_any, "no (gpu, tier) combos were checked"
        assert checked_unsupported, (
            "no unsupported (gpu, tier) combo was exercised — the gate "
            "would pass vacuously if gpu_specs.yaml ever dropped its one "
            "unsupported-precision case (a100_sxm_80gb.peak_tflops.fp8)"
        )

    def test_recommend_never_returns_unsupported_precision_as_candidate(
        self, recommender
    ) -> None:
        specs = {s["id"]: s for s in load_specs() if s.get("in_model_scope")}
        checked_any = False
        checked_unsupported = False
        for model_name in MODEL_PARAMS:
            for tier in get_args(AccuracyTier):
                result = recommender.recommend(model_name=model_name, accuracy_tier=tier)
                for c in result["frontier"] + result["dominated"]:
                    checked_any = True
                    spec = specs[c["gpu_id"]]
                    selected = _selected_precision(spec, tier)
                    assert gpu_supports_precision(spec, selected), (
                        f"{model_name}/{tier}/{c['gpu_id']} is a candidate "
                        f"but does not support its selected precision {selected!r}"
                    )
                for f in result["filtered"]:
                    checked_any = True
                    spec = specs[f["gpu_id"]]
                    selected = _selected_precision(spec, tier)
                    if not gpu_supports_precision(spec, selected):
                        checked_unsupported = True
                        assert selected in f["reject_reason"], (
                            f"{model_name}/{tier}/{f['gpu_id']} filtered for "
                            f"an unsupported precision, but reject_reason "
                            f"{f['reject_reason']!r} doesn't name it"
                        )
        assert checked_any, "no result entries were produced to check"
        assert checked_unsupported, (
            "no unsupported-precision filtered entry was exercised across "
            "the full model x tier sweep"
        )


# Schema/predictor agreement — Literal types match VALID_* frozensets

@pytest.mark.gate
class TestSchemaPredictorAgreementGate:
    """GATE: Pydantic Literal types in src/api/schemas.py must exactly match the independently-defined VALID_* frozensets in src/models/predictor.py — schemas accepting more causes a 500 at the predictor, schemas accepting less makes a valid predictor call unreachable via the API."""

    def test_scenario_literals_match_valid_scenarios(self) -> None:
        assert set(get_args(Scenario)) == VALID_SCENARIOS, (
            f"schemas.Scenario={set(get_args(Scenario))} "
            f"!= predictor.VALID_SCENARIOS={VALID_SCENARIOS}"
        )

    def test_accuracy_tier_literals_match_valid_tiers(self) -> None:
        assert set(get_args(AccuracyTier)) == VALID_TIERS, (
            f"schemas.AccuracyTier={set(get_args(AccuracyTier))} "
            f"!= predictor.VALID_TIERS={VALID_TIERS}"
        )

    def test_framework_literals_match_valid_frameworks(self) -> None:
        assert set(get_args(Framework)) == VALID_FRAMEWORKS, (
            f"schemas.Framework={set(get_args(Framework))} "
            f"!= predictor.VALID_FRAMEWORKS={VALID_FRAMEWORKS}"
        )


# has_training_data completeness across every recommend() code path

@pytest.mark.gate
class TestHasTrainingDataCompletenessGate:
    """GATE: every dict recommend() returns (frontier/dominated/filtered) must carry 'has_training_data' and 'training_data_tier' regardless of which of the three independent code paths built it (predict_batch-derived, VRAM-fail reject, precision-fail reject) — a future path that forgets either field would pass unit tests then 500 at request time on real traffic; both fields are checked together since they're added/read at the same call sites."""

    @pytest.fixture(scope="class")
    def predictor(self) -> GpuPredictor:
        return GpuPredictor()

    @pytest.fixture(scope="class")
    def recommender(self, predictor: GpuPredictor) -> GpuRecommender:
        return GpuRecommender(predictor)

    def test_zero_row_gpus_flagged_false(self, predictor: GpuPredictor) -> None:
        # a100_sxm_80gb/l4/rtx4090 are in_model_scope with zero training rows — a regression here means an unvalidated prediction is silently presented with the same confidence as a real one.
        for gpu_id in ("a100_sxm_80gb", "l4", "rtx4090"):
            assert predictor.has_training_data(gpu_id) is False, (
                f"{gpu_id} unexpectedly has training data — "
                "if this is intentional (new calibration data added), "
                "update this gate and the disclosure copy in the UI/README."
            )
            assert predictor.training_data_tier(gpu_id) == "none", (
                f"{gpu_id} unexpectedly has a non-'none' training_data_tier."
            )

    def test_amd_gpus_below_the_reliability_floor_flagged(self, predictor: GpuPredictor) -> None:
        # mi300x/mi325x/mi355x have real rows but fewer than the 100-row-per-GPU floor — the exact case a plain boolean silently conflated with "sufficient" until this gate/field existed.
        for gpu_id in ("mi300x", "mi325x", "mi355x"):
            assert predictor.has_training_data(gpu_id) is True, (
                f"{gpu_id} unexpectedly has zero training rows — "
                "if the corpus shrank, update this gate."
            )
            assert predictor.training_data_tier(gpu_id) == "below_floor", (
                f"{gpu_id}'s training_data_tier is no longer 'below_floor' — "
                "if this GPU's row count crossed 100 (e.g. new calibration data), "
                "that's a real improvement — update this gate and the disclosure "
                "copy in the UI/README rather than silently letting it drift."
            )

    def test_every_result_dict_carries_both_fields(self, recommender: GpuRecommender) -> None:
        # Exercises all three branches in one pass: llama3.1-405b guarantees VRAM-fail entries (nothing fits 405B); a $1/hr budget guarantees budget-reject entries plus a populated frontier/dominated.
        combos = [
            dict(model_name="llama3.1-405b", accuracy_tier="99.9"),
            dict(model_name="gptj", accuracy_tier="99", budget_per_gpu_hr=1.0),
        ]
        checked_any = False
        for kwargs in combos:
            result = recommender.recommend(**kwargs)
            for bucket in ("frontier", "dominated", "filtered"):
                for entry in result[bucket]:
                    checked_any = True
                    assert "has_training_data" in entry, (
                        f"{bucket} entry for {entry.get('gpu_id')} "
                        f"(query {kwargs}) is missing has_training_data — "
                        "this would fail Pydantic response validation at "
                        "request time, not at test time."
                    )
                    assert "training_data_tier" in entry, (
                        f"{bucket} entry for {entry.get('gpu_id')} "
                        f"(query {kwargs}) is missing training_data_tier — "
                        "this would fail Pydantic response validation at "
                        "request time, not at test time."
                    )
                    assert entry["training_data_tier"] in ("none", "below_floor", "sufficient"), (
                        f"{bucket} entry for {entry.get('gpu_id')} has an "
                        f"invalid training_data_tier: {entry['training_data_tier']!r}"
                    )
        assert checked_any, (
            "no result entries were produced to check — queries need adjusting"
        )


# Feature vector parity — training path vs serving path

@pytest.mark.gate
class TestFeatureVectorParityGate:
    """GATE: all 20 FEATURE_COLS values match between build_training_df (training path) and _build_feature_vector (serving path) — 13 continuous features here, 6 categorical in test_predictor.py::TestBuildFeatureVector, and mlperf_round_num intentionally excluded (training uses the real round ordinal, serving always uses the latest) and verified separately there."""

    # (gpu_alias, gpu_id, model_name, scenario, tier, raw_framework); gpu_alias must match a known alias in data/gpu_specs.yaml, raw_framework is a realistic MLPerf string that _normalize_framework maps to the family label _build_feature_vector expects.
    CASES = [
        # NVIDIA Hopper, FP8 tier (99 -> fp8, falls back for GPUs without fp8)
        (
            "NVIDIA H200-SXM-141GB",          "h200_sxm",  "llama2-70b",
            "Offline", "99",   "TensorRT 10.2.0",
        ),
        # AMD CDNA3, FP8 tier (99.9 -> fp8 for AMD) — verifies both paths apply the vendor override consistently.
        (
            "AMD Instinct MI300X 192GB HBM3",  "mi300x",    "gptj",
            "Server",  "99.9", "vLLM 0.4.3+rocm614",
        ),
        # AMD CDNA3, BF16 tier (base -> bf16); MoE model (active != total params)
        (
            "AMD Instinct MI325X",             "mi325x",    "mixtral-8x7b",
            "Offline", "99",   "ROCm 7.0",
        ),
        # AMD CDNA4 — only GPU where amd_arch_gen=2 and is_cdna4=1; a CDNA4 arch-ordinal encoding divergence can't be detected by the 3 CDNA3/NVIDIA cases above.
        (
            "AMD Instinct MI355X 288GB HBM3e", "mi355x",    "llama2-70b",
            "Offline", "99",   "ROCm 7.0",
        ),
        # NVIDIA at tier 99.9 selects fp16 TFLOPS (not fp8, no AMD override) — without this case a training/serving divergence in NVIDIA fp16 TFLOPS selection would pass undetected (all other cases are tier 99 or AMD 99.9, both fp8).
        (
            "NVIDIA H100-SXM-80GB",            "h100_sxm", "gptj",
            "Server",  "99.9", "TensorRT 10.4.0",
        ),
    ]

    @pytest.fixture(scope="class")
    def spec_map(self) -> dict:
        return {s["id"]: s for s in load_specs()}

    @pytest.mark.parametrize(
        "gpu_alias, gpu_id, model_name, scenario, tier, raw_fw",
        CASES,
    )
    def test_continuous_features_match(
        self, spec_map, gpu_alias, gpu_id, model_name, scenario, tier, raw_fw,
    ):
        """GATE: every continuous FEATURE_COL has the same value in both paths — a mismatch means the model evaluates features at serving time that it was never trained on."""
        # ── Training path ─────────────────────────────────────────────────
        raw_df = _make_single_row(gpu_alias, model_name, scenario, tier, raw_fw)
        feat_df = build_training_df(raw_df)
        assert len(feat_df) == 1, (
            f"build_training_df dropped the row — check gpu_alias={gpu_alias!r} "
            f"and benchmark_base={model_name!r}"
        )
        train_row = feat_df.iloc[0]

        # ── Serving path ──────────────────────────────────────────────────
        fw_family = _normalize_framework(raw_fw)
        serving_feats, _, _ = _build_feature_vector(
            gpu_spec=spec_map[gpu_id],
            model_name=model_name,
            scenario=scenario,
            accuracy_tier=tier,
            framework=fw_family,
        )

        # ── Compare continuous columns — mlperf_round_num excluded, its training/serving divergence is intentional, see class docstring.
        continuous_cols = [
            c for c in FEATURE_COLS
            if c not in {
                "scenario_offline", "is_base_tier",
                "fw_tensorrt", "fw_vllm", "fw_rocm_other", "is_cdna4",
                "mlperf_round_num",
            }
        ]
        mismatches = []
        for col in continuous_cols:
            idx = FEATURE_COLS.index(col)
            sv = float(serving_feats[idx])
            tv_raw = train_row[col]
            tv = float(tv_raw) if not pd.isna(tv_raw) else float("nan")

            if not _nan_match(sv, tv):
                mismatches.append(
                    f"  {col:<40} serving={sv:.6g}  training={tv:.6g}"
                )

        assert not mismatches, (
            f"feature mismatch for gpu={gpu_id!r} model={model_name!r} "
            f"scenario={scenario!r} tier={tier!r}:\n"
            + "\n".join(mismatches)
        )


# Model artifact total disk size < 50 MB

_MODEL_DIR = Path("data/models")
_DISK_LIMIT_MB = 50


@pytest.mark.gate
class TestModelDiskSizeGate:
    """GATE: data/models/ total size is < 50 MB — the deployment budget for free-tier Hugging Face Spaces (shared storage, slow cold-start I/O)."""

    def test_model_dir_exists(self) -> None:
        assert _MODEL_DIR.is_dir(), (
            f"{_MODEL_DIR} not found — run train_final.py to generate artifacts"
        )

    def test_total_size_under_limit(self) -> None:
        total_bytes = sum(f.stat().st_size for f in _MODEL_DIR.rglob("*") if f.is_file())
        total_mb = total_bytes / (1024 ** 2)
        assert total_mb < _DISK_LIMIT_MB, (
            f"data/models/ is {total_mb:.1f} MB — exceeds {_DISK_LIMIT_MB} MB limit. "
            "Trim model artifacts before deploying to HF Spaces."
        )

    def test_required_artifacts_present(self) -> None:
        required = ["prophet_v1.json", "feature_metadata.json"]
        for name in required:
            assert (_MODEL_DIR / name).exists(), (
                f"required artifact {name} is missing from {_MODEL_DIR}"
            )


# Pricing coverage — all in-scope GPUs have pricing entries

_PRICING_PATH = Path("data/pricing.yaml")


@pytest.mark.gate
class TestPricingCoverageGate:
    """GATE: every in_model_scope=True GPU has a data/pricing.yaml entry — GpuRecommender.__init__()'s ValueError for missing pricing is only visible at startup, and a gap would otherwise cause cost_efficiency=None, raising TypeError in _pareto_frontier's sort and silently breaking all recommendation responses."""

    def test_pricing_file_exists(self) -> None:
        assert _PRICING_PATH.is_file(), (
            f"{_PRICING_PATH} not found — create data/pricing.yaml before deployment"
        )

    def test_all_in_scope_gpus_have_pricing(self) -> None:
        with _PRICING_PATH.open() as f:
            pricing: dict = yaml.safe_load(f).get("pricing", {})
        in_scope_ids = [s["id"] for s in load_specs() if s.get("in_model_scope")]
        assert in_scope_ids, (
            "no in-scope GPUs found — check gpu_specs.yaml in_model_scope flags"
        )
        missing = [gid for gid in in_scope_ids if gid not in pricing]
        assert not missing, (
            f"pricing.yaml missing entries for in-scope GPUs: {missing}. "
            "GpuRecommender.__init__() will raise ValueError at startup."
        )


# AMD FP8 override — three-way consistency gate

@pytest.mark.gate
class TestAmdFp8OverrideGate:
    """GATE: the AMD 99.9-tier FP8 override must apply consistently across build_training_df (training), _build_feature_vector (serving), and _gpu_model_size_gb (VRAM pre-filter) — a regression skews MI355X training efficiency_ratio or wrongly rejects MI300X for llama2-70b@99.9 (140 GB FP16 vs 192 GB VRAM) despite fitting as 70 GB FP8; verified end-to-end via vram_headroom ≈0.635 (FP8) vs ≈0.271 (FP16, drifted)."""

    @pytest.fixture(scope="class")
    def recommender(self) -> GpuRecommender:
        return GpuRecommender(GpuPredictor())

    def test_mi300x_vram_headroom_uses_fp8_not_fp16(self, recommender) -> None:
        # Minimal batch/context override isolates the FP8-vs-FP16 weight comparison this gate targets from the KV-cache term (covered separately by test_recommender.py).
        result = recommender.recommend(
            model_name="llama2-70b", accuracy_tier="99.9",
            batch_size=1, input_tokens=64, output_tokens=1,
        )
        candidates = result["frontier"] + result["dominated"]
        mi300x = next((r for r in candidates if r["gpu_id"] == "mi300x"), None)
        assert mi300x is not None, (
            "MI300X absent from llama2-70b tier-99.9 candidates — "
            "VRAM filter may be applying FP16 model size (140 GB) instead of FP8 (70 GB), "
            "causing MI300X (192 GB) to be incorrectly rejected."
        )
        fp8_model_gb = 70.0  # 70B params × 1 byte/param (FP8)
        expected_headroom = (192.0 - fp8_model_gb * 1.10) / 192.0   # + 10% overhead, ≈ 0.599
        assert mi300x["vram_headroom"] == pytest.approx(expected_headroom, abs=0.01), (
            f"MI300X vram_headroom={mi300x['vram_headroom']:.3f}, "
            f"expected {expected_headroom:.3f} (FP8 70 GB model). "
            "A value near 0.271 indicates FP16 (140 GB) is being used instead."
        )


# Framework normalization — pattern outputs subset of VALID_FRAMEWORKS

@pytest.mark.gate
class TestFrameworkNormalizationGate:
    """GATE: every _FRAMEWORK_PATTERNS label (and the "other" fallback) is in VALID_FRAMEWORKS — a pattern added with a label outside VALID_FRAMEWORKS would enter training data but never be requestable via the API, causing silent 500s (predictor._validate() rejecting an apparently valid input)."""

    def test_all_pattern_outputs_in_valid_frameworks(self) -> None:
        labels_from_patterns = {label for _, label in _FRAMEWORK_PATTERNS}
        unknown_labels = labels_from_patterns - VALID_FRAMEWORKS
        assert not unknown_labels, (
            f"_FRAMEWORK_PATTERNS returns labels not in VALID_FRAMEWORKS: "
            f"{unknown_labels}. Add to predictor.VALID_FRAMEWORKS or remove from patterns."
        )

    def test_fallback_other_in_valid_frameworks(self) -> None:
        fallback = _normalize_framework("some_completely_unknown_framework_xyz")
        assert fallback == "other", (
            f"no-match fallback should be 'other', got {fallback!r}. "
            "The fallback must be in VALID_FRAMEWORKS so unrecognized frameworks "
            "can be requested via the API without a 422 validation error."
        )
        assert "other" in VALID_FRAMEWORKS, (
            "'other' is _normalize_framework's no-match return value "
            "but is not in VALID_FRAMEWORKS — unrecognized frameworks fail predictor._validate()."
        )


# Notebook FEATURE_COLS parity — training/serving vs. analysis notebooks

_NOTEBOOKS_DIR = Path("notebooks")
_FEATURE_COLS_NOTEBOOKS: tuple[str, ...] = (
    "03_model_training.ipynb",
    "04_top1_benchmark.ipynb",
    "05_mr008_ablation.ipynb",
)

# In-place mutations of FEATURE_COLS (or a list it's built from) can't be verified statically, so the value this module resolves may not match what the notebook computes at runtime.
_MUTATING_LIST_METHODS = frozenset({
    "append", "extend", "insert", "remove", "pop", "clear", "sort", "reverse",
})


def _parse_notebook_feature_cols(notebook_path: Path) -> list[str]:
    """Extract the notebook's actual FEATURE_COLS = X + Y + ... expression via AST (without executing the notebook) — order matters, not just the set, since XGBoost's colsample_bytree does positional random feature subsampling."""
    nb = json.loads(notebook_path.read_text())
    code = "\n".join(
        "".join(cell["source"])
        for cell in nb["cells"]
        if cell["cell_type"] == "code"
    )
    tree = ast.parse(code)

    list_literals: dict[str, list[str]] = {}
    feature_cols_expr: ast.expr | None = None
    for node in ast.walk(tree):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if not isinstance(target, ast.Name):
                    continue
                if isinstance(node.value, ast.List):
                    list_literals[target.id] = [
                        elt.value for elt in node.value.elts
                        if isinstance(elt, ast.Constant) and isinstance(elt.value, str)
                    ]
                elif target.id == "FEATURE_COLS":
                    feature_cols_expr = node.value

    if feature_cols_expr is None:
        raise RuntimeError(f"FEATURE_COLS assignment not found in {notebook_path}")

    # Names this resolution depends on, populated by _resolve below; FEATURE_COLS is always included so a bare `FEATURE_COLS += [...]` after the found assignment is still caught.
    referenced_names: set[str] = {"FEATURE_COLS"}

    def _resolve(expr: ast.expr) -> list[str]:
        if isinstance(expr, ast.Name):
            referenced_names.add(expr.id)
            if expr.id not in list_literals:
                raise RuntimeError(f"{expr.id} list literal not found in {notebook_path}")
            return list_literals[expr.id]
        if isinstance(expr, ast.BinOp) and isinstance(expr.op, ast.Add):
            return _resolve(expr.left) + _resolve(expr.right)
        raise RuntimeError(
            f"Unsupported FEATURE_COLS expression in {notebook_path}: {ast.dump(expr)}"
        )

    result = _resolve(feature_cols_expr)

    # The scan above only sees each name's initial literal assignment, not later in-place mutation (`+=`, `.append(...)`, etc) — since that's exactly the drift this gate exists to catch, fail loudly rather than silently trust a possibly-stale value.
    for node in ast.walk(tree):
        if isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name):
            if node.target.id in referenced_names:
                raise RuntimeError(
                    f"{notebook_path} mutates {node.target.id!r} with '+=' — "
                    "static AST parsing can't verify the resulting FEATURE_COLS. "
                    "Rewrite as a single 'FEATURE_COLS = A + B + ...' assignment."
                )
        if (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr in _MUTATING_LIST_METHODS
            and isinstance(node.func.value, ast.Name)
            and node.func.value.id in referenced_names
        ):
            raise RuntimeError(
                f"{notebook_path} calls "
                f"{node.func.value.id}.{node.func.attr}(...) — static AST parsing "
                "can't verify the resulting FEATURE_COLS. Rewrite as a single "
                "'FEATURE_COLS = A + B + ...' assignment."
            )

    return result


@pytest.mark.gate
class TestNotebookFeatureColsParityGate:
    """GATE: every notebook manually reconstructing the feature list must match predictor.FEATURE_COLS exactly, in order — TestFeatureVectorParityGate protects the two source-code copies, but only this gate catches the two notebooks silently drifting out of sync after a FEATURE_COLS change (otherwise only caught by manually re-running them)."""

    @pytest.mark.parametrize("notebook_name", _FEATURE_COLS_NOTEBOOKS)
    def test_notebook_feature_cols_match_predictor(self, notebook_name: str) -> None:
        notebook_path = _NOTEBOOKS_DIR / notebook_name
        assert notebook_path.is_file(), f"{notebook_path} not found"
        notebook_cols = _parse_notebook_feature_cols(notebook_path)
        assert notebook_cols == list(FEATURE_COLS), (
            f"{notebook_name} FEATURE_COLS diverges from predictor.FEATURE_COLS.\n"
            f"  notebook:  {notebook_cols}\n"
            f"  predictor: {list(FEATURE_COLS)}"
        )


# memory_fit_verdict() output subset of VALID_MEMORY_FIT_VERDICTS

@pytest.mark.gate
class TestMemoryFitVerdictLabelsGate:
    """GATE: memory_fit_verdict() only ever returns a value from VALID_MEMORY_FIT_VERDICTS, and schemas.MemoryFitVerdict matches that set exactly — mirrors the framework-label closed-set gate; an unlisted fourth verdict from a future threshold-logic change would otherwise silently 500 at request time via Pydantic validation instead of failing a test."""

    @pytest.mark.parametrize("utilization_target", [0.0, 0.50, 0.90, 0.9001, 0.98, 0.9801, 5.0])
    def test_verdict_always_in_valid_set(self, utilization_target: float) -> None:
        vram_gb = 100.0
        weights_gb = utilization_target * vram_gb / MEMORY_OVERHEAD_FACTOR
        verdict, _, _ = memory_fit_verdict(weights_gb, 0.0, vram_gb)
        assert verdict in VALID_MEMORY_FIT_VERDICTS, (
            f"memory_fit_verdict() returned {verdict!r}, "
            f"not in VALID_MEMORY_FIT_VERDICTS={VALID_MEMORY_FIT_VERDICTS}"
        )

    def test_schema_literal_matches_valid_set(self) -> None:
        assert set(get_args(MemoryFitVerdict)) == VALID_MEMORY_FIT_VERDICTS, (
            f"schemas.MemoryFitVerdict={set(get_args(MemoryFitVerdict))} "
            f"!= build_features.VALID_MEMORY_FIT_VERDICTS={VALID_MEMORY_FIT_VERDICTS}"
        )


# RecommendRequest.ranking_objective subset of VALID_RANKING_OBJECTIVES

@pytest.mark.gate
class TestRankingObjectiveLabelsGate:
    """GATE: schemas.RankingObjective matches recommender.VALID_RANKING_OBJECTIVES exactly, and _RANKING_FIELDS covers every member — mirrors the memory-fit-verdict/framework-label closed-set gates; a future typo in either set would silently diverge (accepted by one, KeyError/422 from the other) with no test failing."""

    def test_schema_literal_matches_valid_set(self) -> None:
        assert set(get_args(RankingObjective)) == VALID_RANKING_OBJECTIVES, (
            f"schemas.RankingObjective={set(get_args(RankingObjective))} "
            f"!= recommender.VALID_RANKING_OBJECTIVES={VALID_RANKING_OBJECTIVES}"
        )

    def test_every_objective_is_a_real_recommender_ranking_field(self) -> None:
        assert set(_RANKING_FIELDS) == VALID_RANKING_OBJECTIVES, (
            f"recommender._RANKING_FIELDS keys={set(_RANKING_FIELDS)} "
            f"!= VALID_RANKING_OBJECTIVES={VALID_RANKING_OBJECTIVES}"
        )

    @pytest.mark.parametrize("objective", sorted(VALID_RANKING_OBJECTIVES))
    def test_every_objective_accepted_by_recommend(self, objective: str) -> None:
        # Full round-trip through the public entry point, not just the internal frozenset — catches a rejection at the recommend() validation layer even if the frozenset/Literal agree.
        predictor = GpuPredictor()
        recommender = GpuRecommender(predictor)
        result = recommender.recommend(model_name="gptj", ranking_objective=objective)
        assert result["workload"]["ranking_objective"] == objective


# Pareto objective vector uses (throughput, price, watts), not the earlier (throughput, cost_efficiency, vram_headroom)

@pytest.mark.gate
class TestParetoObjectiveVectorGate:
    """GATE: _pareto_frontier()'s dominance check must read watts/price_per_gpu_hr directly — a pair differing only in watts must change the outcome, and one differing only in vram_headroom must NOT, proving the spec's (throughput, usd_per_hour, watts) vector, guarding against silently backsliding to the old (throughput, cost_efficiency, vram_headroom) vector."""

    @staticmethod
    def _cand(
        gpu_id: str, throughput: float, price: float, watts: float, vram_headroom: float,
    ) -> dict:
        return {
            "gpu_id": gpu_id,
            "throughput": throughput,
            "price_per_gpu_hr": price,
            "watts": watts,
            "vram_headroom": vram_headroom,
            "cost_efficiency": throughput / price,
            "tokens_per_watt": throughput / watts,
            "cost_per_million_tokens": cost_per_million_tokens(price, throughput),
        }

    def test_watts_affects_dominance(self) -> None:
        # Identical throughput/price/vram_headroom; B draws less power than A.
        a = self._cand("a", 1000.0, 2.0, 1000.0, 0.5)
        b = self._cand("b", 1000.0, 2.0, 100.0, 0.5)
        frontier, dominated = _pareto_frontier([a, b])
        assert [c["gpu_id"] for c in frontier] == ["b"], (
            "lower watts at identical throughput/price must dominate — "
            "watts is not affecting Pareto dominance"
        )
        assert [c["gpu_id"] for c in dominated] == ["a"]

    def test_vram_headroom_does_not_affect_dominance(self) -> None:
        # Identical throughput/price/watts, only vram_headroom differs — the objective vector has no room for a fourth soft axis, so neither may dominate on vram_headroom alone.
        a = self._cand("a", 1000.0, 2.0, 500.0, 0.9)
        b = self._cand("b", 1000.0, 2.0, 500.0, 0.1)
        frontier, dominated = _pareto_frontier([a, b])
        assert {c["gpu_id"] for c in frontier} == {"a", "b"}, (
            "vram_headroom must not affect Pareto dominance — "
            f"got frontier={[c['gpu_id'] for c in frontier]}, "
            f"dominated={[c['gpu_id'] for c in dominated]}"
        )
        assert dominated == []


# Calibration runner/merge script CSV compatibility

_RUNNER_PATH = Path("benchmarks/run_mi300x_calibration.py")

# Fields merge_calibration_rows.py accesses directly from CSV rows (r["field"] or r.get(...)); update whenever the merge script adds a new field access.
_MERGE_REQUIRED_FIELDS: frozenset[str] = frozenset({
    "round_tag",
    "gpu_name",
    "benchmark_base",
    "benchmark_accuracy_tier",
    "scenario",
    "throughput_tok_per_sec",
    "vllm_version",
    # DataFrame subscript access (merge_calibration_rows.py L64) is not a .get() — raises KeyError if the column is absent from the CSV.
    "precision_used",
})


def _parse_runner_csv_fields() -> frozenset[str]:
    """Extract CSV_FIELDS list from runner script via AST — avoids importing the script directly, which would fail without vllm installed."""
    tree = ast.parse(_RUNNER_PATH.read_text())
    for node in ast.walk(tree):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == "CSV_FIELDS":
                    if isinstance(node.value, ast.List):
                        return frozenset(
                            elt.value
                            for elt in node.value.elts
                            if isinstance(elt, ast.Constant) and isinstance(elt.value, str)
                        )
    raise RuntimeError("CSV_FIELDS constant not found in benchmarks/run_mi300x_calibration.py")


@pytest.mark.gate
class TestRunnerMergeCompatibilityGate:
    """GATE: run_mi300x_calibration.py's CSV_FIELDS is a superset of the columns merge_calibration_rows.py reads — drift causes KeyError (or silent None) when processing real AMD Dev Cloud results after the 25-hr credit clock is already spent, so this gate runs locally before the timed session starts."""

    def test_runner_script_exists(self) -> None:
        assert _RUNNER_PATH.is_file(), (
            f"{_RUNNER_PATH} not found — calibration runner script is missing"
        )

    def test_runner_csv_fields_cover_merge_required(self) -> None:
        runner_fields = _parse_runner_csv_fields()
        missing = _MERGE_REQUIRED_FIELDS - runner_fields
        assert not missing, (
            f"merge_calibration_rows.py accesses fields not in runner CSV_FIELDS: "
            f"{missing}. Add these to CSV_FIELDS in run_mi300x_calibration.py, or "
            "remove the access from merge_calibration_rows.py."
        )


@pytest.mark.gate
class TestDataManifestLockFreshnessGate:
    """GATE: data_manifest.lock's committed calibration-CSV entry matches the CSV's real SHA-256 — the only manifest entry checkable in CI this way, since the gitignored MLPerf mirrors need a live local checkout (checked instead by train_final.py's warn-only verify_manifest()); this gate catches the one drift class reproducible from a fresh clone: the lock going stale against its own tracked sibling."""

    def test_manifest_lock_file_exists(self) -> None:
        assert manifest.MANIFEST_PATH.is_file(), (
            f"{manifest.MANIFEST_PATH} not found — run "
            "`python -m src.data.manifest` to generate it."
        )

    def test_calibration_entry_matches_tracked_csv(self) -> None:
        locked = manifest.load_manifest()
        assert locked is not None, "data_manifest.lock failed to load"
        entry = locked["sources"].get("mi300x_calibration")
        assert entry is not None, "data_manifest.lock missing 'mi300x_calibration' entry"
        current_sha256 = manifest._file_sha256(manifest.CALIBRATION_CSV)
        assert entry["sha256"] == current_sha256, (
            "data_manifest.lock's mi300x_calibration sha256 is stale — "
            "the tracked CSV changed since the lock file was last regenerated. "
            "Run `python -m src.data.manifest` and commit the refreshed lock file."
        )


# feature_metadata.json carries a well-formed corpus_sha256

_FEATURE_METADATA_PATH = Path("data/models/feature_metadata.json")
_TRAINING_CORPUS_PATH = Path("data/processed/mlperf_features.parquet")
_SHA256_HEX_CHARS = frozenset("0123456789abcdef")


@pytest.mark.gate
class TestCorpusShaSidecarGate:
    """GATE: feature_metadata.json records a well-formed corpus_sha256, completing the lineage chain served prediction -> feature_metadata.json -> data_manifest.lock -> pinned MLPerf commits; the field-presence/shape check always runs (metadata is committed), but the byte-for-byte match against the live corpus is best-effort since data/processed/ is gitignored and usually absent in CI."""

    def test_feature_metadata_exists(self) -> None:
        assert _FEATURE_METADATA_PATH.is_file(), (
            f"{_FEATURE_METADATA_PATH} not found — run train_final.py "
            "to generate artifacts"
        )

    def test_corpus_sha256_present_and_well_formed(self) -> None:
        meta = json.loads(_FEATURE_METADATA_PATH.read_text())
        corpus_hash = meta.get("corpus_sha256")
        assert corpus_hash, (
            "feature_metadata.json missing 'corpus_sha256' — retrain "
            "with `python -m src.models.train_final` to populate it."
        )
        assert len(corpus_hash) == 64 and set(corpus_hash) <= _SHA256_HEX_CHARS, (
            f"corpus_sha256 {corpus_hash!r} is not a well-formed SHA-256 hex digest"
        )

    def test_corpus_sha256_matches_live_corpus_when_present(self) -> None:
        if not _TRAINING_CORPUS_PATH.is_file():
            pytest.skip(
                f"{_TRAINING_CORPUS_PATH} not present locally (gitignored, "
                "not fetched/built) — nothing to compare against."
            )
        meta = json.loads(_FEATURE_METADATA_PATH.read_text())
        current_hash = manifest.corpus_sha256(_TRAINING_CORPUS_PATH)
        assert meta.get("corpus_sha256") == current_hash, (
            "feature_metadata.json's corpus_sha256 is stale — the training "
            "corpus changed since the model was last trained. Run "
            "`python -m src.models.train_final` and commit the refreshed artifacts."
        )

    def test_manifest_verified_present_and_boolean(self) -> None:
        """corpus_sha256 alone proves only *which* corpus trained the model, not whether its raw sources matched data_manifest.lock at build time — manifest_verified records that check's outcome (train_final.py's verify_manifest(), which only logs a warning and would otherwise leave no trace)."""
        meta = json.loads(_FEATURE_METADATA_PATH.read_text())
        assert "manifest_verified" in meta, (
            "feature_metadata.json missing 'manifest_verified' — retrain "
            "with `python -m src.models.train_final` to populate it."
        )
        assert isinstance(meta["manifest_verified"], bool), (
            f"manifest_verified should be a bool, got {type(meta['manifest_verified'])!r}"
        )

    def test_manifest_sources_coverage_present_and_sane(self) -> None:
        """manifest_verified alone can't distinguish "every source present and matched" from "every source simply absent, trivially not mismatched" — manifest_sources_present/total give the denominator needed to judge what manifest_verified: true is actually worth for this build."""
        meta = json.loads(_FEATURE_METADATA_PATH.read_text())
        assert "manifest_sources_present" in meta and "manifest_sources_total" in meta, (
            "feature_metadata.json missing manifest_sources_present/total — "
            "retrain with `python -m src.models.train_final` to populate it."
        )
        present, total = meta["manifest_sources_present"], meta["manifest_sources_total"]
        assert isinstance(present, int) and isinstance(total, int), (
            "manifest_sources_present/total should be ints"
        )
        assert 0 <= present <= total, (
            f"manifest_sources_present ({present}) must be between 0 and "
            f"manifest_sources_total ({total})"
        )

    def test_validation_metrics_present_and_well_formed(self) -> None:
        """The key must always exist (possibly null, if notebooks/03 hasn't been re-run since the last corpus build); when non-null, its primary summary fields must be sane finite fractions, not NaN/inf/negative placeholders that would silently pass a looser "key exists" check."""
        meta = json.loads(_FEATURE_METADATA_PATH.read_text())
        assert "validation_metrics" in meta, (
            "feature_metadata.json missing 'validation_metrics' key "
            "(should be present, possibly null) — retrain with "
            "`python -m src.models.train_final` to populate it."
        )
        vm = meta["validation_metrics"]
        if vm is None:
            pytest.skip("validation_metrics is null (notebooks/03 not re-run since last build)")

        primary = vm["primary"]
        for key in ("mape", "smape", "median_ape"):
            val = primary[key]
            assert isinstance(val, (int, float)) and math.isfinite(val) and val >= 0.0, (
                f"validation_metrics.primary.{key} = {val!r} is not a sane fraction"
            )
        rho = primary["spearman_rho"]
        assert isinstance(rho, (int, float)) and math.isfinite(rho) and -1.0 <= rho <= 1.0, (
            f"validation_metrics.primary.spearman_rho = {rho!r} out of [-1, 1]"
        )
        assert isinstance(primary["n_folds"], int) and primary["n_folds"] > 0, (
            f"validation_metrics.primary.n_folds = {primary.get('n_folds')!r} "
            "should be a positive int"
        )

    def test_trained_gpu_row_counts_present_and_consistent_with_trained_gpu_ids(self) -> None:
        """training_data_tier() needs a per-GPU row count to distinguish below-the-floor from ample data — a gap the older 'trained_gpu_ids' boolean set couldn't express; the two fields must describe the same set, so every id in trained_gpu_ids has a positive count here and vice versa."""
        meta = json.loads(_FEATURE_METADATA_PATH.read_text())
        assert "trained_gpu_row_counts" in meta, (
            "feature_metadata.json missing 'trained_gpu_row_counts' — "
            "retrain with `python -m src.models.train_final` to populate it."
        )
        counts = meta["trained_gpu_row_counts"]
        trained_ids = set(meta["trained_gpu_ids"])
        assert set(counts.keys()) == trained_ids, (
            f"trained_gpu_row_counts keys {sorted(counts.keys())} != "
            f"trained_gpu_ids {sorted(trained_ids)}"
        )
        for gpu_id, n in counts.items():
            assert isinstance(n, int) and n > 0, (
                f"trained_gpu_row_counts[{gpu_id!r}] = {n!r} should be a positive int"
            )