diff --git a/plugins/modelopt/skills/specdec-release/SKILL.md b/plugins/modelopt/skills/specdec-release/SKILL.md index d594c308476..8501f9a900e 100644 --- a/plugins/modelopt/skills/specdec-release/SKILL.md +++ b/plugins/modelopt/skills/specdec-release/SKILL.md @@ -1,14 +1,14 @@ --- name: specdec-release -description: Deterministic readiness gates for a speculative-decoding drafter — training/export (gate_train), and serving canary (gate_smoke). Use when the user asks whether a trained drafter (EAGLE3, DFlash, DFlash2, DSpark, Domino, LiLiCorr) is ready for release review, or to gate the artifacts of a draft-training pipeline run. Do NOT use to configure, debug or triage a draft-training run (use speculative-decoding), to quantize (use ptq), or to serve a model (use deployment). +description: Deterministic readiness gates for a speculative-decoding drafter — training/export (gate_train), serving canary (gate_smoke), matched A|B benchmark validity (gate_bench), and a non-blocking greedy-match report (gate_correctness). Use when the user asks whether a trained drafter (EAGLE3, DFlash, DFlash2, DSpark, Domino, LiLiCorr) is ready for release review, or to gate the artifacts of a draft-training pipeline run. Do NOT use to configure, debug or triage a draft-training run (use speculative-decoding), to quantize (use ptq), or to serve a model (use deployment). license: Apache-2.0 --- # Spec-Dec Release Gates that decide whether a drafter is **ready for release review**: trained cleanly, -exported to fit its target, and servable with speculation active. They set no per-model -performance bar; AL is returned as evidence. The +exported to fit its target, servable with speculation active, and measured validly. They +set no per-model performance bar. AL, speedup and TPS are returned as evidence. The checkpoint release criteria (Addendum A: AL and throughput bars) are a later stage and are not implemented here. @@ -23,14 +23,18 @@ login node; they are stdlib-only. **Never read a verdict from a log or a Slurm s | --- | --- | --- | | Train + export | `python "$SKILL_DIR/scripts/gate_train.py" --regression /regression_check.json --export /exported-checkpoint-final --target-config /config.json --algorithm ` | `check_regression.py --json` output, export dir, target config | | Smoke | `python "$SKILL_DIR/scripts/gate_smoke.py" --smoke --draft --target ` | `vllm_smoke_test.sh` `SMOKE_JSON` | +| A\|B bench | `python "$SKILL_DIR/scripts/gate_bench.py" --summary /ab_summary.json --concurrencies 1,8,32 --repeats N --algorithm DFLASH --num-spec-tokens 7 --draft --target ` | `compare_ab.py` summary | +| Correctness | `python "$SKILL_DIR/scripts/gate_correctness.py" --summary /ab_summary.json` | same summary; **report-only**, always passes | A gate never passes a run its producing script failed, classifies the failure itself only when the script cannot (so `gate_smoke` reads raw values; `gate_train` keeps `check_regression.py`'s verdict), and never re-implements a script's comparison. -Artifact files are read +Pass `gate_bench` what the run was **submitted** with, not targets. Artifact files are read from host paths (`/cicd//...`), but `--draft`/`--target` are compared to the paths the -job recorded, so pass them as the job saw them (e.g. `/scratchspace/...`). `gate_smoke` +job recorded, so pass them as the job saw them (e.g. `/scratchspace/...`). An empty +summary is `USER_CONFIG_ERROR`: check `--summary`; if the path is right, the bench job died +before writing any arm, so resubmit as for `BENCH_INCOMPLETE`. `gate_smoke` classifies failures from the raw values and its `stage`; a SMOKE_JSON without `stage` comes from an older script, so re-run the smoke test. @@ -46,7 +50,11 @@ from an older script, so re-run the smoke test. | `CHECKPOINT_NOT_SERVABLE` | smoke | Same as `day0-release`: fix the export or serving flags; do not benchmark | | `SPEC_NOT_ACTIVE` | smoke | AL ≤ 1.05: drafter not wired up or useless. Check `SPEC_METHOD`, token count, drafter path | | `ACCEPTANCE_REGRESSION` | smoke | Serves and speculates, but AL < the recipe's `MIN_ACCEPTANCE_LENGTH`. Triage training with **speculative-decoding**; do not benchmark | -| `SPEC_CONFIG_MISMATCH` | smoke | Served the wrong drafter or target. Fix the submission; re-run | - -Verdict: both gates passed → a servable drafter, ready for a matched benchmark; -training, export or serving failed for good → `INFEASIBLE`. +| `SPEC_CONFIG_MISMATCH` | smoke, bench | Wrong or unrecorded (pass `--block_size`/`--draft_length`) drafter/target/algorithm/token count, or speculation on in arm A / off in arm B. Fix the submission; re-run | +| `BENCH_INCOMPLETE` | bench | Missing rows or arms, appended results, arms with different response counts. Resubmit into a **fresh** `SAVE_ROOT` (≤ 2 automatic retries) | +| `ARMS_UNMATCHED` | bench | Arms or rows differ in target, engine, GPU, TP, dataset, length or temperature, or an arm ran at another concurrency than its directory says. The comparison is invalid; re-run in one job | +| `ANOMALOUS` | bench | Implausible numbers (AL or speedup above tokens + 1, non-positive TPS, unpaired turns). Do not read the numbers; investigate | + +Verdict: every blocking gate passed → `DRAFTER_READY` (not a publish decision); +training, export or serving failed for good → `INFEASIBLE`; the measurement is invalid → +`ANOMALOUS`. diff --git a/plugins/modelopt/skills/specdec-release/scripts/gate_bench.py b/plugins/modelopt/skills/specdec-release/scripts/gate_bench.py new file mode 100644 index 00000000000..add6883cd92 --- /dev/null +++ b/plugins/modelopt/skills/specdec-release/scripts/gate_bench.py @@ -0,0 +1,299 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Spec-dec benchmark gate: is the matched A|B measurement valid? + +Reads the ab_summary.json written by examples/specdec_bench/compare_ab.py (arm A = target +alone, arm B = target + drafter). Like day0-release's gate_run, it decides whether the +numbers can be trusted, not whether they are good: it sets no performance bar. AL, +speedup and TPS are returned as evidence, with the setup they were measured under. + +The expected concurrencies, repeats, algorithm and token count are what the run was +submitted with, not thresholds. Pure decision logic in ``evaluate_bench``. +""" + +from __future__ import annotations + +import argparse +import json +import math +import sys + +# With speculation off every step emits exactly one token; allow rounding only. +SPEC_OFF_MAX_AL = 1.02 +# Mean acceptance length counts the bonus token, so 1.0 means no draft token was accepted. +SPEC_ON_MIN_AL = 1.05 +# Settings both arms, and every row, must share for the comparison to be matched. +MATCHED_KEYS = ( + "model_dir", + "engine", + "engine_version", + "gpu", + "tp_size", + "ep_size", + "dataset", + "dataset_path", + "output_length", + "num_requests", +) +# run.py passes the speculative-token count as --block_size for these, --draft_length otherwise. +_BLOCK_SIZE_ALGOS = {"DFLASH", "DSPARK"} + +_RANKING = ("BENCH_INCOMPLETE", "ARMS_UNMATCHED", "SPEC_CONFIG_MISMATCH", "ANOMALOUS") + + +def _num(value): + if isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value): + return value + return None + + +def _spec_tokens(config): + algo = (config.get("speculative_algorithm") or "").upper() + return config.get("block_size" if algo in _BLOCK_SIZE_ALGOS else "draft_length") + + +def evaluate_bench( + summary, + concurrencies=None, + repeats=1, + algorithm=None, + num_spec_tokens=None, + draft=None, + target=None, +): + """Validate an A|B summary. + + Args: + summary: dict from ab_summary.json. + concurrencies: concurrencies the run was submitted with; each must be present. + repeats: how many repeats were submitted; each concurrency needs this many rows. + algorithm: run.py --speculative_algorithm of arm B (e.g. DFLASH); optional. + num_spec_tokens: speculative tokens arm B was submitted with; optional. + draft: expected drafter path or HF id; optional. + target: expected target model path or HF id; optional. + + Returns: + dict ``{pass, failure_class, detail, notes, evidence, measurement_setup}``. + """ + rows = (summary or {}).get("rows") or [] + if not rows: + return { + "pass": False, + "failure_class": "USER_CONFIG_ERROR", + "detail": "summary has no rows", + "notes": [], + } + + problems, notes = [], [] + if not summary.get("ok"): + problems.append(("BENCH_INCOMPLETE", "comparator reported problems")) + found = {} + for r in rows: + found.setdefault(r.get("concurrency"), set()).add(r.get("repeat")) + problems.extend( + ("BENCH_INCOMPLETE", f"concurrency {c}: {len(found.get(c, ()))} of {repeats} repeats") + for c in concurrencies or [] + if len(found.get(c, ())) < repeats + ) + extra = sorted(set(found) - set(concurrencies or found), key=str) + if extra: + notes.append(f"concurrencies not requested: {extra}") + more = sorted((c for c, reps in found.items() if len(reps) > repeats), key=str) + if more: + notes.append(f"more than {repeats} repeat(s) at concurrencies {more}") + + reference = rows[0].get("config_a") or {} + for r in rows: + where = f"repeat {r.get('repeat')} C={r.get('concurrency')}" + problems.extend(("BENCH_INCOMPLETE", f"{where}: {p}") for p in r.get("problems") or []) + ca, cb = r.get("config_a"), r.get("config_b") + if not ca or not cb: + problems.append(("BENCH_INCOMPLETE", f"{where}: arm configuration missing")) + continue + + n_req = ca.get("num_requests") + responses = (r.get("responses_a"), r.get("responses_b")) + if not responses[0] or responses[0] != responses[1]: + problems.append(("BENCH_INCOMPLETE", f"{where}: responses A/B {responses}")) + elif isinstance(n_req, int) and responses[0] != n_req: + # The dataset caps --num_requests (MT-Bench has 80 questions), and a run cut short + # already shows as a missing result entry, so fewer is not a failure by itself. + notes.append(f"{where}: {responses[0]} responses per arm, {n_req} requested") + for key in MATCHED_KEYS: + values = {ca.get(key), cb.get(key), reference.get(key)} + if len(values) > 1: + problems.append( + ("ARMS_UNMATCHED", f"{where}: {key} differs ({sorted(map(str, values))})") + ) + temperatures = { + r.get("temperature_a"), + r.get("temperature_b"), + rows[0].get("temperature_a"), + } + if len(temperatures) > 1: + problems.append( + ( + "ARMS_UNMATCHED", + f"{where}: temperature differs ({sorted(map(str, temperatures))})", + ) + ) + # The directory name gives the row its concurrency; the arms record what actually ran. + for arm, config in (("A", ca), ("B", cb)): + if config.get("concurrency") != r.get("concurrency"): + problems.append( + ("ARMS_UNMATCHED", f"{where}: arm {arm} ran at C={config.get('concurrency')}") + ) + + if (ca.get("speculative_algorithm") or "").upper() != "NONE": + problems.append(("SPEC_CONFIG_MISMATCH", f"{where}: arm A is not NONE")) + al_a, al_b = _num(r.get("al_a")), _num(r.get("al_b")) + if al_a is not None and al_a > SPEC_OFF_MAX_AL: + problems.append( + ("SPEC_CONFIG_MISMATCH", f"{where}: arm A AL {al_a} > {SPEC_OFF_MAX_AL}") + ) + algo_b = (cb.get("speculative_algorithm") or "").upper() + if algorithm and algo_b != algorithm.upper(): + problems.append( + ("SPEC_CONFIG_MISMATCH", f"{where}: arm B ran {algo_b}, not {algorithm}") + ) + tokens = _spec_tokens(cb) + if tokens is None: + # run.py leaves --block_size unset and the engine picks its own default, so the + # token count is unknown and the AL and speedup ceilings below cannot apply. + problems.append( + ( + "SPEC_CONFIG_MISMATCH", + f"{where}: arm B token count not recorded; " + "pass --block_size/--draft_length explicitly", + ) + ) + elif num_spec_tokens is not None and tokens != num_spec_tokens: + problems.append( + ( + "SPEC_CONFIG_MISMATCH", + f"{where}: arm B used {tokens} speculative tokens, not {num_spec_tokens}", + ) + ) + if draft is not None and cb.get("draft_model_dir") != draft: + problems.append( + ("SPEC_CONFIG_MISMATCH", f"{where}: drafter {cb.get('draft_model_dir')!r}") + ) + if target is not None and cb.get("model_dir") != target: + problems.append(("SPEC_CONFIG_MISMATCH", f"{where}: target {cb.get('model_dir')!r}")) + if al_b is None or al_b <= SPEC_ON_MIN_AL: + problems.append( + ( + "SPEC_CONFIG_MISMATCH", + f"{where}: arm B AL {r.get('al_b')!r} shows no speculation", + ) + ) + + problems.extend( + ("ANOMALOUS", f"{where}: {key} {r.get(key)!r}") + for key in ("tps_a", "tps_b", "speedup") + if _num(r.get(key)) is None or r[key] <= 0 + ) + if isinstance(tokens, int) and al_b is not None and al_b > tokens + 1: + problems.append(("ANOMALOUS", f"{where}: AL {al_b} > {tokens} tokens + 1")) + if isinstance(tokens, int) and _num(r.get("speedup")) and r["speedup"] > tokens + 1: + problems.append(("ANOMALOUS", f"{where}: speedup {r['speedup']} > {tokens} tokens + 1")) + if (r.get("match") or {}).get("unpaired_turns"): + problems.append( + ("ANOMALOUS", f"{where}: {r['match']['unpaired_turns']} unpaired turns") + ) + + cb0 = rows[0].get("config_b") or {} + result = { + "notes": notes, + "evidence": [ + {k: r.get(k) for k in ("repeat", "concurrency", "al_b", "speedup", "tps_a", "tps_b")} + for r in rows + ], + "measurement_setup": { + **{k: cb0.get(k) for k in (*MATCHED_KEYS, "speculative_algorithm", "draft_model_dir")}, + "num_speculative_tokens": _spec_tokens(cb0), + "temperature": rows[0].get("temperature_b"), + "concurrencies": sorted(c for c in found if isinstance(c, int)), + }, + } + if not problems: + return { + "pass": True, + "failure_class": None, + "detail": f"{len(rows)} row(s) valid", + **result, + } + fc = min((c for c, _ in problems), key=_RANKING.index) + return { + "pass": False, + "failure_class": fc, + "detail": "; ".join(f"{c}: {m}" for c, m in problems), + **result, + } + + +def main(argv=None): + """CLI entry point: read ab_summary.json and print the verdict.""" + p = argparse.ArgumentParser(description="Spec-dec A|B benchmark gate") + p.add_argument("--summary", required=True, help="ab_summary.json from compare_ab.py") + p.add_argument( + "--concurrencies", help="comma-separated concurrencies the run was submitted with" + ) + p.add_argument("--repeats", type=int, default=1, help="repeats the run was submitted with") + p.add_argument("--algorithm", help="arm B --speculative_algorithm, e.g. DFLASH") + p.add_argument( + "--num-spec-tokens", type=int, help="speculative tokens arm B was submitted with" + ) + p.add_argument("--draft", help="expected drafter path or HF id") + p.add_argument("--target", help="expected target model path or HF id") + args = p.parse_args(argv) + + try: + with open(args.summary) as f: + summary = json.load(f) + if not isinstance(summary, dict): + raise ValueError(f"{args.summary} does not hold a JSON object") + rows = summary.get("rows") + if rows is not None and not ( + isinstance(rows, list) and all(isinstance(r, dict) for r in rows) + ): + raise ValueError(f"{args.summary}: 'rows' is not a list of objects") + concurrencies = ( + [int(c) for c in args.concurrencies.split(",")] if args.concurrencies else None + ) + except (OSError, ValueError) as e: + error = {"pass": False, "failure_class": "USER_CONFIG_ERROR", "detail": str(e)} + print(json.dumps({**error, "notes": [], "inputs": {"summary": args.summary}}, indent=2)) + return 2 + + result = evaluate_bench( + summary, + concurrencies, + args.repeats, + args.algorithm, + args.num_spec_tokens, + args.draft, + args.target, + ) + result["inputs"] = {"summary": args.summary} + print(json.dumps(result, indent=2)) + if result["failure_class"] == "USER_CONFIG_ERROR": + return 2 + return 0 if result["pass"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/plugins/modelopt/skills/specdec-release/scripts/gate_correctness.py b/plugins/modelopt/skills/specdec-release/scripts/gate_correctness.py new file mode 100644 index 00000000000..f1e13a22a7c --- /dev/null +++ b/plugins/modelopt/skills/specdec-release/scripts/gate_correctness.py @@ -0,0 +1,149 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Spec-dec correctness report: do greedy outputs with and without the drafter match? + +REPORT-ONLY: always returns pass true (exit 0) on readable input. How often greedy outputs +drift apart depends on the target's numerics, the engine's kernels and the output length, +so any blocking floor would be a per-model number. The rates go into the report as evidence +for the release review, next to the target-alone control (compare_ab's ``a_self_match``: A +at concurrency 1 vs A at the next concurrency). That control is the noise floor: a hybrid +target can agree with itself on few turns, and then a low A-vs-B rate is not the drafter's. + +Reads ab_summary.json from examples/specdec_bench/compare_ab.py and uses only rows measured +at concurrency 1 with temperature 0 recorded for both arms (batching alone breaks bitwise +equality above concurrency 1), with no comparator problems and speculation on in arm B. +It does not repeat gate_bench's other checks (matched arms, drafter, plausibility), so read +its rates only alongside a passing gate_bench. Pure logic in ``evaluate_correctness``. +""" + +from __future__ import annotations + +import argparse +import json +import sys + + +def evaluate_correctness(summary): + """Report A-vs-B exact match over the greedy, concurrency-1 rows of an A|B summary. + + Returns: + dict ``{pass, failure_class, blocking, measurable, detail, notes, + min_exact_match_rate, min_first_turn_exact_match_rate, min_self_exact_match_rate, + min_self_first_turn_exact_match_rate, per_repeat}``. + """ + greedy = [ + r + for r in (summary or {}).get("rows") or [] + if r.get("concurrency") == 1 + and r.get("temperature_a") == r.get("temperature_b") == 0 + and isinstance(r.get("match"), dict) + ] + rows = [ + r + for r in greedy + if not r.get("problems") + and isinstance(r.get("config_b"), dict) + and (r["config_b"].get("speculative_algorithm") or "NONE").upper() != "NONE" + ] + notes: list[str] = [] + if not (summary or {}).get("ok"): + notes.append("comparator reported problems; see gate_bench") + if len(rows) < len(greedy): + notes.append( + f"{len(greedy) - len(rows)} greedy row(s) skipped: comparator problems or " + "speculation off in arm B" + ) + result = {"pass": True, "failure_class": None, "blocking": False, "notes": notes} + if not rows: + return { + **result, + "measurable": False, + "detail": "no concurrency-1 row with temperature 0 recorded for both arms", + } + + per_repeat = [{"repeat": r.get("repeat"), **r["match"]} for r in rows] + exact = [m["exact_match_rate"] for m in per_repeat if m.get("exact_match_rate") is not None] + first = [ + m["first_turn_exact_match_rate"] + for m in per_repeat + if m.get("first_turn_exact_match_rate") is not None + ] + if any(m.get("unpaired_turns") for m in per_repeat): + notes.append("some turns had no counterpart in the other arm") + min_exact = min(exact) if exact else None + min_first = min(first) if first else None + controls = [r["a_self_match"] for r in rows if isinstance(r.get("a_self_match"), dict)] + self_exact = [c["exact_match_rate"] for c in controls if c.get("exact_match_rate") is not None] + self_first = [ + c["first_turn_exact_match_rate"] + for c in controls + if c.get("first_turn_exact_match_rate") is not None + ] + min_self = min(self_exact) if self_exact else None + min_self_first = min(self_first) if self_first else None + detail = ( + f"exact match {min_exact} (all turns), {min_first} (first turns); not gated" + if exact + else "no turn pairs to compare" + ) + if self_exact: + detail += f"; target alone across batch shapes {min_self} / {min_self_first}" + elif exact: + notes.append( + "no target-alone control (arm A at a second concurrency): a low rate cannot be " + "told apart from the target's own batch-shape drift" + ) + return { + **result, + "measurable": bool(exact), + "detail": detail, + "min_exact_match_rate": min_exact, + "min_first_turn_exact_match_rate": min_first, + "min_self_exact_match_rate": min_self, + "min_self_first_turn_exact_match_rate": min_self_first, + "per_repeat": per_repeat, + } + + +def main(argv=None): + """CLI entry point: read ab_summary.json and print the report.""" + p = argparse.ArgumentParser(description="Spec-dec correctness report (non-blocking)") + p.add_argument("--summary", required=True, help="ab_summary.json from compare_ab.py") + args = p.parse_args(argv) + + try: + with open(args.summary) as f: + summary = json.load(f) + if not isinstance(summary, dict): + raise ValueError(f"{args.summary} does not hold a JSON object") + rows = summary.get("rows") + if rows is not None and not ( + isinstance(rows, list) and all(isinstance(r, dict) for r in rows) + ): + raise ValueError(f"{args.summary}: 'rows' is not a list of objects") + except (OSError, ValueError) as e: + error = {"pass": False, "failure_class": "USER_CONFIG_ERROR", "detail": str(e)} + print(json.dumps({**error, "notes": [], "inputs": {"summary": args.summary}}, indent=2)) + return 2 + + result = evaluate_correctness(summary) + result["inputs"] = {"summary": args.summary} + print(json.dumps(result, indent=2)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/plugins/modelopt/skills/specdec-release/tests/test_specdec_gates.py b/plugins/modelopt/skills/specdec-release/tests/test_specdec_gates.py index 53145fbf7a1..a683a0b3e02 100644 --- a/plugins/modelopt/skills/specdec-release/tests/test_specdec_gates.py +++ b/plugins/modelopt/skills/specdec-release/tests/test_specdec_gates.py @@ -16,7 +16,8 @@ """Unit tests for the spec-dec readiness gates. These are deterministic — no GPU, cluster, or network. Inputs carry the values of a real -Qwen3-8B DFlash2 run (2000 steps on GB300), trimmed to the fields the gates read. Run with: +Qwen3-8B DFlash2 run (2000 steps on GB300) and of its matched A|B benchmark, trimmed to +the fields the gates read. Run with: python -m pytest "$SKILL_DIR/tests/test_specdec_gates.py" """ @@ -30,6 +31,10 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) +from gate_bench import evaluate_bench +from gate_bench import main as bench_main +from gate_correctness import evaluate_correctness +from gate_correctness import main as correctness_main from gate_smoke import evaluate_smoke from gate_smoke import main as smoke_main from gate_train import evaluate_training @@ -96,6 +101,63 @@ def _smoke(): } +# M0 matched benchmark (MT-Bench, 80 requests, temperature 0, 7 speculative tokens): +# concurrency -> (AL A, AL B, Output TPS A, Output TPS B). +_M0 = { + 1: (1.0, 1.7716790663672861, 257.84536489469593, 369.63355605307476), + 8: (1.0011991946127003, 1.7539989653331933, 1732.5925424359984, 2461.854827003075), + 32: (1.0006437486492314, 1.7754698448542836, 4150.190548956775, 5761.7201506093925), +} + + +def _summary(): + """compare_ab.py ab_summary.json of the M0 benchmark, fields the gates read.""" + rows = [] + for conc, (al_a, al_b, tps_a, tps_b) in _M0.items(): + config = { + "draft_model_dir": DRAFT, + "num_requests": 80, + "model_dir": TARGET, + "engine": "VLLM", + "engine_version": "0.31.0", + "gpu": "NVIDIA GB300", + "tp_size": 1, + "ep_size": 1, + "dataset": "mtbench", + "dataset_path": "/hf-local/HuggingFaceH4/mt_bench_prompts/raw/question.jsonl", + "output_length": 4096, + "concurrency": conc, + } + row = { + "repeat": 0, + "concurrency": conc, + "problems": [], + "al_a": al_a, + "al_b": al_b, + "tps_a": tps_a, + "tps_b": tps_b, + "speedup": tps_b / tps_a, + "responses_a": 80, + "responses_b": 80, + "turns_a": 160, + "turns_b": 160, + "temperature_a": 0.0, + "temperature_b": 0.0, + "config_a": {**config, "speculative_algorithm": "NONE", "block_size": 8}, + "config_b": {**config, "speculative_algorithm": "DFLASH", "block_size": 7}, + } + if conc == 1: + row["match"] = { + "pairs": 160, + "unpaired_turns": 0, + "exact_match_rate": 0.95625, + "first_turn_pairs": 80, + "first_turn_exact_match_rate": 1.0, + } + rows.append(row) + return {"ok": True, "rows": rows} + + def _with(obj, path, value): """A copy of obj with one nested key replaced (value None deletes it).""" obj = copy.deepcopy(obj) @@ -264,6 +326,110 @@ def test_smoke(change, kwargs, failure_class): assert r["pass"] == (failure_class is None) +# ── gate_bench ──────────────────────────────────────────────────────── + +SUBMITTED = { + "concurrencies": [1, 8, 32], + "algorithm": "DFLASH", + "num_spec_tokens": 7, + "draft": DRAFT, +} + + +def test_bench_m0_passes_and_reproduces_m0_numbers(): + r = evaluate_bench(_summary(), **SUBMITTED) + assert r["pass"], r["detail"] + # m0_analysis.json, computed by hand from the same run.py outputs. + evidence = {e["concurrency"]: e for e in r["evidence"]} + assert evidence[1]["speedup"] == pytest.approx(1.4335474139860265) + assert evidence[8]["speedup"] == pytest.approx(1.420908128544606) + assert evidence[32]["speedup"] == pytest.approx(1.3883025568687934) + assert r["measurement_setup"]["gpu"] == "NVIDIA GB300" + assert r["measurement_setup"]["num_speculative_tokens"] == 7 + + # --num_requests above the dataset size: run.py runs all 80 MT-Bench questions. + summary = _summary() + for row in summary["rows"]: + row["config_a"]["num_requests"] = row["config_b"]["num_requests"] = 100 + r = evaluate_bench(summary, **SUBMITTED) + assert r["pass"] and r["notes"], r["detail"] + + +def _mutate_row(field, value, arm_config=None): + def apply(summary): + row = summary["rows"][0] + if arm_config: + row[arm_config][field] = value + else: + row[field] = value + + return apply + + +@pytest.mark.parametrize( + ("mutate", "submitted", "failure_class"), + [ + (lambda s: s["rows"].pop(), {}, "BENCH_INCOMPLETE"), + (lambda s: None, {"repeats": 2}, "BENCH_INCOMPLETE"), + (_mutate_row("responses_b", 79), {}, "BENCH_INCOMPLETE"), + (_mutate_row("gpu", "NVIDIA H100", "config_b"), {}, "ARMS_UNMATCHED"), + (_mutate_row("tp_size", 2, "config_b"), {}, "ARMS_UNMATCHED"), + (lambda s: s["rows"][1].update(temperature_a=0.7, temperature_b=0.7), {}, "ARMS_UNMATCHED"), + (lambda s: s["rows"][2]["config_b"].update(concurrency=1), {}, "ARMS_UNMATCHED"), + (_mutate_row("al_a", 1.3), {}, "SPEC_CONFIG_MISMATCH"), + (_mutate_row("speculative_algorithm", "DFLASH", "config_a"), {}, "SPEC_CONFIG_MISMATCH"), + (lambda s: None, {"num_spec_tokens": 15}, "SPEC_CONFIG_MISMATCH"), + # --block_size left unset: the engine's default ran, and the ceilings can't apply. + ( + _mutate_row("block_size", None, "config_b"), + {"num_spec_tokens": None}, + "SPEC_CONFIG_MISMATCH", + ), + (_mutate_row("al_b", 1.0), {}, "SPEC_CONFIG_MISMATCH"), + (_mutate_row("al_b", 9.0), {}, "ANOMALOUS"), + (_mutate_row("tps_a", 0.0), {}, "ANOMALOUS"), + ], +) +def test_bench_failures(mutate, submitted, failure_class): + summary = _summary() + mutate(summary) + r = evaluate_bench(summary, **{**SUBMITTED, **submitted}) + assert not r["pass"] + assert r["failure_class"] == failure_class, r["detail"] + + +# ── gate_correctness ────────────────────────────────────────────────── + + +def test_correctness_reports_greedy_match_without_gating(): + summary = _summary() + r = evaluate_correctness(summary) + assert r["pass"] and r["measurable"] and not r["blocking"] + assert r["min_exact_match_rate"] == pytest.approx(0.95625) + assert r["min_self_exact_match_rate"] is None and r["notes"] # no control: say so + + # The target-alone control is reported as the noise floor next to the A-vs-B rate. + summary["rows"][0]["a_self_match"] = { + "exact_match_rate": 0.2, + "first_turn_exact_match_rate": 0.25, + } + r = evaluate_correctness(summary) + assert r["min_self_exact_match_rate"] == 0.2 + assert r["min_self_first_turn_exact_match_rate"] == 0.25 + assert "target alone" in r["detail"] and not r["notes"] + + summary["rows"][0]["temperature_b"] = None # unknown temperature: not greedy evidence + r = evaluate_correctness(summary) + assert r["pass"] and not r["measurable"] + + # Rows gate_bench would reject are not evidence either: appended results, or arm B + # comparing the target with itself. + for path, value in ((["rows", 0, "problems"], ["appended"]), + (["rows", 0, "config_b", "speculative_algorithm"], "NONE")): # fmt: skip + r = evaluate_correctness(_with(_summary(), path, value)) + assert r["pass"] and not r["measurable"] and r["notes"] + + # ── CLIs ────────────────────────────────────────────────────────────── @@ -312,3 +478,19 @@ def test_cli_exit_codes(tmp_path, capsys): for not_an_object in ([], "stage", None): bad = _write(tmp_path / "bad.json", not_an_object) assert smoke_main(["--smoke", bad]) == 2 + assert bench_main(["--summary", bad]) == 2 + assert correctness_main(["--summary", bad]) == 2 + + for rows in ({"0": {}}, ["row"]): + bad = _write(tmp_path / "bad.json", {"rows": rows}) + assert bench_main(["--summary", bad]) == 2 + assert correctness_main(["--summary", bad]) == 2 + bad_match = _with(_summary(), ["rows", 0, "match"], []) + assert correctness_main(["--summary", _write(tmp_path / "bad.json", bad_match)]) == 0 + none_concurrency = _with(_summary(), ["rows", 0, "concurrency"], None) + assert bench_main(["--summary", _write(tmp_path / "bad.json", none_concurrency)]) == 1 + + summary = _write(tmp_path / "ab_summary.json", _summary()) + assert bench_main(["--summary", summary, "--concurrencies", "1,8,32"]) == 0 + assert bench_main(["--summary", summary, "--concurrencies", "1,64"]) == 1 + assert bench_main(["--summary", str(tmp_path / "missing.json")]) == 2