diff --git a/.claude/commands/recover-failed-ingest.md b/.claude/commands/recover-failed-ingest.md index 701daf27d8..9db1d6c330 100644 --- a/.claude/commands/recover-failed-ingest.md +++ b/.claude/commands/recover-failed-ingest.md @@ -324,6 +324,12 @@ python3 utils/validate_reusable_sweep_artifacts.py \ --artifacts-dir /tmp/source-artifacts ``` +The validator first collapses reran (flaky) eval duplicates in place — keeping +the latest result per eval identity when a retried eval left duplicate raw dirs +/ aggregate rows — so a legitimate rerun does not fail validation. It only +collapses identities with a clear latest result; genuinely ambiguous duplicates +are still rejected. + The validator does not compare source coverage with `/tmp/recovery-full-config.json`. It rejects duplicate fixed rows, missing run stats, inconsistent agentic artifacts, malformed eval metadata, raw/aggregate diff --git a/.github/workflows/run-sweep.yml b/.github/workflows/run-sweep.yml index 797c693ded..b348d84854 100644 --- a/.github/workflows/run-sweep.yml +++ b/.github/workflows/run-sweep.yml @@ -723,6 +723,8 @@ jobs: find source-artifacts -maxdepth 1 -mindepth 1 -type d -printf ' %f\n' | sort - name: Validate reusable artifacts + # Collapses reran (flaky) eval duplicates to the latest result + # per identity, then validates consistency for ingest. run: | python3 utils/validate_reusable_sweep_artifacts.py \ --artifacts-dir source-artifacts diff --git a/utils/test_validate_reusable_sweep_artifacts.py b/utils/test_validate_reusable_sweep_artifacts.py index 0c2bcad855..716646619c 100644 --- a/utils/test_validate_reusable_sweep_artifacts.py +++ b/utils/test_validate_reusable_sweep_artifacts.py @@ -6,6 +6,7 @@ from validate_reusable_sweep_artifacts import ( agentic_key, + dedupe_reran_evals, main, validate_agentic_artifacts, validate_eval_artifacts, @@ -473,3 +474,153 @@ def test_eval_only_main_does_not_require_benchmark_artifacts( ) assert main() == 0 + + +# ── dedupe_reran_evals ──────────────────────────────────────────────────────── + + +def _dd_meta(conc: int) -> dict: + return { + "is_multinode": True, + "hw": "b300", + "infmax_model_prefix": "minimaxm3", + "framework": "dynamo-vllm", + "precision": "fp4", + "spec_decoding": "none", + "isl": 8192, + "osl": 1024, + "prefill_tp": 2, + "prefill_ep": 2, + "prefill_dp_attention": True, + "prefill_num_workers": 4, + "decode_tp": 8, + "decode_ep": 8, + "decode_dp_attention": True, + "decode_num_workers": 2, + "conc": conc, + } + + +def _dd_agg_row(conc: int, source: str, em_strict: float) -> dict: + row = _dd_meta(conc) + row["model_prefix"] = row.pop("infmax_model_prefix") + row["task"] = "gsm8k" + row["em_strict"] = em_strict + row["source"] = source + return row + + +def _dd_write_aggregate(root: Path, rows: list[dict]) -> Path: + eval_dir = root / "eval_results_all" + eval_dir.mkdir(exist_ok=True) + path = eval_dir / "agg_eval_all.json" + path.write_text(json.dumps(rows, indent=2)) + return path + + +def _dd_write_legacy_raw( + root: Path, name: str, conc: int, timestamp: str | None +) -> None: + artifact_dir = root / name + artifact_dir.mkdir() + (artifact_dir / "meta_env.json").write_text(json.dumps(_dd_meta(conc))) + if timestamp is not None: + (artifact_dir / f"results_{timestamp}.json").write_text("{}") + + +def test_dedupe_keeps_latest_legacy_rerun(tmp_path: Path) -> None: + # Three reruns of one eval plus a result-less attempt, mirroring a flaky + # config retried until it passed. + old, mid, new, empty = ( + "eval_minimaxm3_conc4096_b300-nv_15", + "eval_minimaxm3_conc4096_b300-nv_16", + "eval_minimaxm3_conc4096_b300-nv_12", + "eval_minimaxm3_conc4096_b300-nv_03", + ) + _dd_write_legacy_raw(tmp_path, old, 4096, "2026-06-26T13-00-22.596040") + _dd_write_legacy_raw(tmp_path, mid, 4096, "2026-06-26T19-00-52.356121") + _dd_write_legacy_raw(tmp_path, new, 4096, "2026-06-27T04-28-31.838775") + _dd_write_legacy_raw(tmp_path, empty, 4096, None) + _dd_write_aggregate( + tmp_path, + [ + _dd_agg_row(4096, f"eval_results/{old}/results_2026-06-26T13-00-22.596040.json", 0.83), + _dd_agg_row(4096, f"eval_results/{new}/results_2026-06-27T04-28-31.838775.json", 0.95), + _dd_agg_row(4096, f"eval_results/{mid}/results_2026-06-26T19-00-52.356121.json", 0.78), + ], + ) + + messages = dedupe_reran_evals(tmp_path) + + assert validate_eval_artifacts(tmp_path) == [] + rows = json.loads((tmp_path / "eval_results_all" / "agg_eval_all.json").read_text()) + assert [r["em_strict"] for r in rows] == [0.95] + assert (tmp_path / new).is_dir() + for superseded in (old, mid, empty): + assert not (tmp_path / superseded).exists() + assert any("kept 1 of 3" in message for message in messages) + + +def test_dedupe_leaves_ambiguous_duplicates_for_validation(tmp_path: Path) -> None: + # Duplicate raw identities with no result timestamps cannot be ordered, so + # dedupe must leave them and validation must still reject them. + for name in ("eval_minimaxm3_conc4096_b300-nv_01", "eval_minimaxm3_conc4096_b300-nv_02"): + _dd_write_legacy_raw(tmp_path, name, 4096, None) + _dd_write_aggregate( + tmp_path, + [_dd_agg_row(4096, "eval_results/eval_minimaxm3_conc4096_b300-nv_01/x.json", 0.9)], + ) + + assert dedupe_reran_evals(tmp_path) == [] + assert any("duplicate" in e for e in validate_eval_artifacts(tmp_path)) + + +def test_dedupe_is_noop_for_clean_artifacts(tmp_path: Path) -> None: + name = "eval_minimaxm3_conc4096_b300-nv_01" + _dd_write_legacy_raw(tmp_path, name, 4096, "2026-06-27T04-28-31.838775") + agg_path = _dd_write_aggregate( + tmp_path, + [_dd_agg_row(4096, f"eval_results/{name}/results_2026-06-27T04-28-31.838775.json", 0.95)], + ) + before = agg_path.read_text() + + assert dedupe_reran_evals(tmp_path) == [] + assert agg_path.read_text() == before + assert (tmp_path / name).is_dir() + assert validate_eval_artifacts(tmp_path) == [] + + +def test_dedupe_prunes_superseded_batched_conc(tmp_path: Path) -> None: + # Two batched reruns overlap on conc 32; the newer run wins that conc while + # each run keeps the concurrencies unique to it. + older = tmp_path / "eval_minimaxm3_batch_b300-nv_05" + newer = tmp_path / "eval_minimaxm3_batch_b300-nv_09" + for artifact_dir, concs, stamp in ( + (older, [16, 32], "2026-06-26T10-00-00.000000"), + (newer, [32, 64], "2026-06-26T20-00-00.000000"), + ): + artifact_dir.mkdir() + meta = _dd_meta(0) + meta["eval_concs"] = concs + meta["completed_eval_concs"] = list(concs) + (artifact_dir / "meta_env.json").write_text(json.dumps(meta)) + for conc in concs: + (artifact_dir / f"results_{stamp}_conc{conc}.json").write_text("{}") + _dd_write_aggregate( + tmp_path, + [ + _dd_agg_row(16, f"eval_results/{older.name}/results_2026-06-26T10-00-00.000000_conc16.json", 0.50), + _dd_agg_row(32, f"eval_results/{older.name}/results_2026-06-26T10-00-00.000000_conc32.json", 0.40), + _dd_agg_row(32, f"eval_results/{newer.name}/results_2026-06-26T20-00-00.000000_conc32.json", 0.90), + _dd_agg_row(64, f"eval_results/{newer.name}/results_2026-06-26T20-00-00.000000_conc64.json", 0.70), + ], + ) + + dedupe_reran_evals(tmp_path) + + assert validate_eval_artifacts(tmp_path) == [] + assert json.loads((older / "meta_env.json").read_text())["completed_eval_concs"] == [16] + assert not (older / "results_2026-06-26T10-00-00.000000_conc32.json").exists() + assert (older / "results_2026-06-26T10-00-00.000000_conc16.json").exists() + rows = json.loads((tmp_path / "eval_results_all" / "agg_eval_all.json").read_text()) + assert [r["em_strict"] for r in rows if r["conc"] == 32] == [0.90] diff --git a/utils/validate_reusable_sweep_artifacts.py b/utils/validate_reusable_sweep_artifacts.py index 12f46b4a3a..aed9590e04 100644 --- a/utils/validate_reusable_sweep_artifacts.py +++ b/utils/validate_reusable_sweep_artifacts.py @@ -6,10 +6,12 @@ import argparse import csv import json +import re +import shutil import sys from collections import Counter from pathlib import Path -from typing import Any, Iterable +from typing import Any, Iterable, Optional def as_bool(value: Any) -> bool: @@ -445,6 +447,197 @@ def validate_run_stats(artifacts_dir: Path, required: bool) -> list[str]: return ["missing run-stats artifact for fixed-sequence benchmarks"] +# ── Dedupe reran eval artifacts ─────────────────────────────────────────────── +# +# A flaky eval retried several times leaves multiple raw ``eval_*`` dirs and +# multiple ``eval_results_all`` rows for one logical eval identity, which the +# checks above would otherwise reject. ``dedupe_reran_evals`` collapses those to +# the latest result per identity (by lm-eval result timestamp) so a legitimate +# rerun does not fail validation. It only acts on identities that have a clear +# latest result; genuinely ambiguous duplicates (no result timestamp to order +# them by) are left in place for validation to reject. Eval-only; fixed-sequence +# and agentic artifacts are untouched. + +# lm-eval result files are ``results_.json`` (optionally a ``_concN`` / +# staging suffix). The timestamp uses dashes throughout, so it is fixed-width +# and lexicographically sortable. +_TIMESTAMP_RE = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}(?:\.\d+)?") + +# Batched result files carry their concurrency as a ``_concN`` suffix (kept in +# sync with ``collect_eval_results.CONC_SUFFIX_RE``). +_CONC_SUFFIX_RE = re.compile(r"_conc(\d+)(?:_\d+)?\.json$") + + +def _result_concurrency(name: str) -> Optional[int]: + """Extract a batched eval concurrency from a staged result file name.""" + match = _CONC_SUFFIX_RE.search(name) + return int(match.group(1)) if match else None + + +def _result_timestamp(name: str) -> Optional[str]: + """Extract the sortable lm-eval timestamp from a result file name.""" + match = _TIMESTAMP_RE.search(name) + return match.group(0) if match else None + + +def _raw_dir_contributions( + artifact_dir: Path, +) -> tuple[list[tuple[tuple[Any, ...], Optional[int]]], dict[str, Any], bool]: + """Return (identity, conc) pairs a raw dir contributes, plus its meta. + + Mirrors ``raw_eval_key_rows``: a batched artifact contributes one identity + per ``completed_eval_concs`` entry; a legacy artifact contributes one from + its meta. ``conc`` is the batched concurrency (``None`` for legacy). + """ + meta = load_json(artifact_dir / "meta_env.json") + if not isinstance(meta, dict): + return [], {}, False + batched = isinstance(meta.get("eval_concs"), list) + if batched: + concs = meta.get("completed_eval_concs") + if not isinstance(concs, list): + return [], meta, True + return ( + [(eval_key({**meta, "conc": conc}), as_int(conc)) for conc in concs], + meta, + True, + ) + return [(eval_key(meta), None)], meta, False + + +def _eval_winners(artifacts_dir: Path) -> dict[tuple[Any, ...], str]: + """Pick the raw dir holding the latest result for each eval identity. + + Ranks only by the lm-eval result timestamp, so an identity appears here + only when at least one of its raw dirs carries a real result. Identities + with no timestamped result get no winner and are left untouched. + """ + best: dict[tuple[Any, ...], tuple[str, str]] = {} + for artifact_dir in raw_eval_artifact_dirs(artifacts_dir): + contributions, _, batched = _raw_dir_contributions(artifact_dir) + if not contributions: + continue + for path in artifact_dir.glob("results*.json"): + stamp = _result_timestamp(path.name) + if stamp is None: + continue + conc = _result_concurrency(path.name) + for key, key_conc in contributions: + # A batched result file is tagged with its conc; only let it + # compete for the matching identity. + if batched and conc is not None and key_conc != conc: + continue + candidate = (stamp, artifact_dir.name) + if best.get(key) is None or candidate > best[key]: + best[key] = candidate + return {key: name for key, (_, name) in best.items()} + + +def _dedupe_eval_aggregate( + artifacts_dir: Path, winners: dict[tuple[Any, ...], str] +) -> list[str]: + """Keep one aggregate row per winning identity (its winner dir's row).""" + eval_dir = artifacts_dir / "eval_results_all" + if not eval_dir.is_dir(): + return [] + messages: list[str] = [] + for agg_path in sorted(eval_dir.glob("*.json")): + data = load_json(agg_path) + if not isinstance(data, list): + continue + groups: dict[tuple[Any, ...], list[int]] = {} + keep: set[int] = set() + for idx, row in enumerate(data): + if not isinstance(row, dict): + keep.add(idx) + continue + groups.setdefault(eval_key(row), []).append(idx) + for key, indices in groups.items(): + winner = winners.get(key) + # Only collapse identities with a clear latest result; leave + # ambiguous duplicates for validation to reject. + if winner is None or len(indices) == 1: + keep.update(indices) + continue + keep.add( + next( + ( + idx + for idx in indices + if winner in str(data[idx].get("source") or "") + ), + max( + indices, + key=lambda idx: _result_timestamp( + str(data[idx].get("source") or "") + ) + or "", + ), + ) + ) + if len(keep) != len(data): + kept = [row for idx, row in enumerate(data) if idx in keep] + agg_path.write_text(json.dumps(kept, indent=2)) + messages.append( + f"{agg_path.name}: kept {len(kept)} of {len(data)} eval row(s)" + ) + return messages + + +def _prune_raw_eval_dir( + artifact_dir: Path, winners: dict[tuple[Any, ...], str] +) -> Optional[str]: + """Drop a raw dir's identities that a newer dir supersedes.""" + contributions, meta, batched = _raw_dir_contributions(artifact_dir) + if not contributions: + return None + name = artifact_dir.name + + def superseded(key: tuple[Any, ...]) -> bool: + winner = winners.get(key) + return winner is not None and winner != name + + if not batched: + if superseded(contributions[0][0]): + shutil.rmtree(artifact_dir) + return f"removed superseded raw eval dir {name!r}" + return None + + losing = { + conc for key, conc in contributions if conc is not None and superseded(key) + } + if not losing: + return None + for path in artifact_dir.glob("results*.json"): + if _result_concurrency(path.name) in losing: + path.unlink() + remaining = [ + conc + for conc in meta.get("completed_eval_concs", []) + if as_int(conc) not in losing + ] + if not remaining: + shutil.rmtree(artifact_dir) + return f"removed superseded batched raw eval dir {name!r}" + meta["completed_eval_concs"] = remaining + (artifact_dir / "meta_env.json").write_text(json.dumps(meta)) + dropped = ",".join(str(conc) for conc in sorted(losing)) + return ( + f"pruned superseded conc(s) {dropped} from batched raw eval dir {name!r}" + ) + + +def dedupe_reran_evals(artifacts_dir: Path) -> list[str]: + """Collapse reran eval duplicates in place; return a change log.""" + winners = _eval_winners(artifacts_dir) + messages = _dedupe_eval_aggregate(artifacts_dir, winners) + for artifact_dir in raw_eval_artifact_dirs(artifacts_dir): + message = _prune_raw_eval_dir(artifact_dir, winners) + if message: + messages.append(message) + return messages + + def main() -> int: """CLI entry point.""" parser = argparse.ArgumentParser() @@ -456,6 +649,14 @@ def main() -> int: f"artifacts directory does not exist: {args.artifacts_dir}" ) + # Collapse reran (flaky) eval duplicates to the latest result before + # validating, so a legitimate rerun does not fail the consistency checks. + dedupe_messages = dedupe_reran_evals(args.artifacts_dir) + if dedupe_messages: + print("Collapsed reran eval duplicates (kept latest result per identity):") + for message in dedupe_messages: + print(f" {message}") + fixed_rows = actual_benchmark_key_rows(args.artifacts_dir) agentic_rows = agentic_keys_from_paths( agentic_point_files(args.artifacts_dir)