diff --git a/src/hyperloom/agents/kernel/tests/test_forge_gemm_tuning.py b/src/hyperloom/agents/kernel/tests/test_forge_gemm_tuning.py index 96f76bf651..3f92e9756a 100644 --- a/src/hyperloom/agents/kernel/tests/test_forge_gemm_tuning.py +++ b/src/hyperloom/agents/kernel/tests/test_forge_gemm_tuning.py @@ -36,6 +36,7 @@ def _payload() -> dict: "global_timeout": 456, "tuner": "fmoe_ck", "untuned_csv": "/tmp/in.csv", + "moe_untuned_csv": "/tmp/untuned_fmoe_from_runtime.csv", "shapes_json": "/tmp/shapes.json", "tunableop_input": "/tmp/tunable.txt", "kernel_signature_log": "/tmp/server.log", @@ -57,12 +58,59 @@ def test_build_cmd_maps_all_options(): assert cmd[cmd.index("--quant-type") + 1] == "auto" assert cmd[cmd.index("--mp") + 1] == "8" assert cmd[cmd.index("--tuner") + 1] == "fmoe_ck" + assert cmd[cmd.index("--untuned-csv") + 1] == "/tmp/in.csv" + assert cmd[cmd.index("--kernel-signature-log") + 1] == "/tmp/server.log" + assert cmd[cmd.index("--tp") + 1] == "1" + assert cmd[cmd.index("--conc") + 1] == "256" + assert cmd[cmd.index("--timeout") + 1] == "123" + assert cmd[cmd.index("--global-timeout") + 1] == "456" assert cmd[cmd.index("--tokens") + 1] == "64,128" assert "--skip-gpu-check" in cmd assert "--verbose" in cmd assert "--thorough" in cmd +def test_build_cmd_forwards_the_moe_untuned_csv(): + """The runtime-derived MoE key reaches forge only through this option. + + The orchestrator derives the CSV from the dispatch tuple in the server log; + without the option forge infers the key from the model config instead -- + the exact failure this lane exists to remove, and one that leaves no trace + because the tuning still reports success. + """ + cmd = forge_gemm_tuning._build_cmd(_payload()) + + assert cmd[cmd.index("--moe-untuned-csv") + 1] == "/tmp/untuned_fmoe_from_runtime.csv" + + +def test_build_cmd_omits_the_moe_untuned_csv_when_absent(): + """No runtime key observed: forge must not receive an empty option.""" + payload = _payload() + payload.pop("moe_untuned_csv") + + assert "--moe-untuned-csv" not in forge_gemm_tuning._build_cmd(payload) + + +def test_build_cmd_asserts_every_option_it_can_emit(): + """Meta-guard: an option added to _build_cmd must be asserted in this file. + + This file is the only guard on the agent-tool argv, and it had drifted to + covering 10 of the options it emits -- which is how the MoE CSV option went + unasserted while being the whole point of this lane. Comparing the emitted + flags against a declared set makes the next omission fail here. + """ + emitted = {tok for tok in forge_gemm_tuning._build_cmd(_payload()) if tok.startswith("--")} + declared = { + "--model-path", "--framework", "--precision", "--quant-type", "--gpu-type", + "--tp", "--conc", "--mp", "--output-dir", "--iters", "--warmup", + "--min-improvement-pct", "--timeout", "--global-timeout", "--tuner", + "--untuned-csv", "--moe-untuned-csv", "--shapes-json", "--tunableop-input", + "--kernel-signature-log", "--gpu-ids", "--skip-gpu-check", "--verbose", + "--thorough", "--tokens", "--kb-current-lib", + } + assert emitted <= declared, f"option(s) not declared here: {sorted(emitted - declared)}" + + def test_build_cmd_forwards_provenance_but_no_knowledge_base_options(monkeypatch): """Tuning has no knowledge base; asking it to consult one aborts the run.""" payload = _payload() diff --git a/src/hyperloom/agents/kernel/tools/forge_gemm_tuning.py b/src/hyperloom/agents/kernel/tools/forge_gemm_tuning.py index d8ce9165d5..10d4516ba6 100644 --- a/src/hyperloom/agents/kernel/tools/forge_gemm_tuning.py +++ b/src/hyperloom/agents/kernel/tools/forge_gemm_tuning.py @@ -57,6 +57,7 @@ def _build_cmd(args: dict[str, Any]) -> list[str]: _add_opt(cmd, args, "global_timeout", "--global-timeout") _add_opt(cmd, args, "tuner", "--tuner") _add_opt(cmd, args, "untuned_csv", "--untuned-csv") + _add_opt(cmd, args, "moe_untuned_csv", "--moe-untuned-csv") _add_opt(cmd, args, "shapes_json", "--shapes-json") _add_opt(cmd, args, "tunableop_input", "--tunableop-input") _add_opt(cmd, args, "kernel_signature_log", "--kernel-signature-log") diff --git a/src/hyperloom/common/model_paths.py b/src/hyperloom/common/model_paths.py index d1e6eabe26..32b5afd996 100644 --- a/src/hyperloom/common/model_paths.py +++ b/src/hyperloom/common/model_paths.py @@ -19,7 +19,9 @@ from __future__ import annotations +import os from pathlib import Path +from typing import Any def _identity_leaf(seg: str) -> str: @@ -128,3 +130,60 @@ def resolve_local_model_dir(model: str | Path | None) -> Path | None: if isinstance(hit, str) and Path(hit).is_file(): return Path(hit).parent return None + + +def resolve_serving_model_path(raw: str) -> str: + """Resolve a session model identity to a path suitable for launching servers. + + Precedence mirrors ``run_hyperloom.sbatch``: an existing directory wins, + then ``HL_MODEL_BASE/``, then the HuggingFace hub cache via + :func:`resolve_local_model_dir`. When nothing resolves, the original + string is returned unchanged. + """ + text = str(raw or "").strip() + if not text: + return "" + try: + direct = Path(text).expanduser() + if direct.is_dir(): + return str(direct) + except OSError: + pass + base = os.environ.get("HL_MODEL_BASE", "").strip() + if base: + leaf = text.rstrip("/").split("/")[-1] + if leaf: + candidate = Path(base) / leaf + try: + if candidate.is_dir(): + return str(candidate) + except OSError: + pass + resolved = resolve_local_model_dir(text) + if resolved is not None: + return str(resolved) + return text + + +def resolve_session_model_path( + *, + params: dict[str, Any] | None = None, + state_model_path: str = "", + for_serving: bool = False, +) -> str: + """Unified session model-path precedence for executors and handlers. + + Order: ``params['model_path']`` → ``$MODEL_PATH`` → ``state.model_path``. + When ``for_serving`` is true, :func:`resolve_serving_model_path` is applied + to the chosen raw value. + """ + raw = ( + str((params or {}).get("model_path") or "").strip() + or os.environ.get("MODEL_PATH", "").strip() + or str(state_model_path or "").strip() + ) + if not raw: + return "" + if for_serving: + return resolve_serving_model_path(raw) + return raw diff --git a/src/hyperloom/inference_optimizer/breakdown/collectors/kernels.py b/src/hyperloom/inference_optimizer/breakdown/collectors/kernels.py index 85384ec88b..a5c06f7e98 100644 --- a/src/hyperloom/inference_optimizer/breakdown/collectors/kernels.py +++ b/src/hyperloom/inference_optimizer/breakdown/collectors/kernels.py @@ -1326,10 +1326,15 @@ def collect_gemm_tuning(state: dict[str, Any]) -> dict[str, Any]: if not isinstance(raw, dict): continue engine = _resolve_gemm_engine(raw) + e2e_gain_pct = _to_float(raw.get("e2e_gain_pct")) speedup = _to_float(raw.get("best_speedup")) gain_pct: float | None = None tuned_tput: float | None = None - if speedup is not None: + if e2e_gain_pct is not None: + gain_pct = e2e_gain_pct + if baseline_tput is not None: + tuned_tput = baseline_tput * (1.0 + e2e_gain_pct / 100.0) + elif speedup is not None: gain_pct = (speedup - 1.0) * 100.0 if baseline_tput is not None: tuned_tput = baseline_tput * speedup @@ -1360,6 +1365,7 @@ def collect_gemm_tuning(state: dict[str, Any]) -> dict[str, Any]: "gpu_type": str(raw.get("gpu_type") or gpu_type), "baseline_tput": baseline_tput, "best_speedup": speedup, + "e2e_gain_pct": e2e_gain_pct, "gain_pct": gain_pct, "tuned_tput": tuned_tput, "tuned_file": tuned_file, diff --git a/src/hyperloom/inference_optimizer/cli/__init__.py b/src/hyperloom/inference_optimizer/cli/__init__.py index 55c9f4abfa..78010bcdc1 100644 --- a/src/hyperloom/inference_optimizer/cli/__init__.py +++ b/src/hyperloom/inference_optimizer/cli/__init__.py @@ -1985,7 +1985,9 @@ async def _run_optimize(args: argparse.Namespace) -> int: ) sys.exit(2) # Re-export so subprocess executors inject the resolved model into the Magpie YAML, not its hardcoded model. - os.environ["MODEL_PATH"] = str(args.model) + from hyperloom.common.model_paths import resolve_serving_model_path + + os.environ["MODEL_PATH"] = resolve_serving_model_path(str(args.model)) or str(args.model) # Quantization prelude (one-shot, before any session/baseline work): # if --quantize was passed, quantize the source model now and rewrite diff --git a/src/hyperloom/inference_optimizer/tests/test_coordinator_gemm_promote_units.py b/src/hyperloom/inference_optimizer/tests/test_coordinator_gemm_promote_units.py index a2f26bade6..a1bf59473e 100644 --- a/src/hyperloom/inference_optimizer/tests/test_coordinator_gemm_promote_units.py +++ b/src/hyperloom/inference_optimizer/tests/test_coordinator_gemm_promote_units.py @@ -2012,6 +2012,112 @@ async def test_does_not_e2e_validate_missing_aiter_candidate( "candidate_artifact_missing" ) + @pytest.mark.asyncio + async def test_integrate_bench_fault_not_recorded_as_zero_gain_revert( + self, tmp_path, monkeypatch + ): + """A server that never booted is an integrate fault, not a 0% REVERT.""" + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + phase = KernelPhase(coord) + fmoe_candidate = tmp_path / "fmoe.csv" + dense_candidate = tmp_path / "dense.csv" + fmoe_candidate.write_text("token,model_dim\n1,2\n", encoding="utf-8") + dense_candidate.write_text("M,N,K\n1,2,3\n", encoding="utf-8") + calls: list[dict] = [] + + async def _fake_integrate(payload, *, session_dir): + calls.append(payload) + if payload["kernel_id"] == "gemm_tune_fmoe_ck": + return { + "status": "failed", + "error_class": "bench_exception", + "decision": "REVERT", + "error": "re-baseline did not succeed", + } + return {"status": "ok", "decision": "KEEP", "new_tput": 120.0, "gain_pct": 9.09} + + monkeypatch.setattr(krh_mod, "integrate_handler", _fake_integrate) + monkeypatch.setattr(explore_mod, "_compute_explore_variant_timeout", lambda **_k: 61) + monkeypatch.setattr( + phase, + "_merge_gemm_candidate_with_runtime", + lambda _env_var, env_value: env_value, + ) + + result = { + "backend": "forge", + "tuners_run": [ + { + "status": "ok", + "tuner": "fmoe_ck", + "improved_shapes": 2, + "env_var": "AITER_CONFIG_FMOE", + "env_value": str(fmoe_candidate), + }, + { + "status": "ok", + "tuner": "dense_bf16", + "improved_shapes": 1, + "env_var": "AITER_CONFIG_DENSE", + "env_value": str(dense_candidate), + }, + ], + } + + await phase._validate_gemm_tuning_e2e(result) + + assert len(calls) == 3 + assert result["e2e_results"]["faults"][0]["reason"] == "integrate_fault:bench_exception" + assert result["e2e_results"]["faults"][0]["fault_attempts"] == 2 + assert result["e2e_results"]["reverted"] == [] + assert result["e2e_results"]["kept"][0]["tuner"] == "dense_bf16" + assert result["decision"] == "KEEP" + + @pytest.mark.asyncio + async def test_integrate_fault_retries_once_before_verdict(self, tmp_path, monkeypatch): + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + phase = KernelPhase(coord) + dense_candidate = tmp_path / "dense.csv" + dense_candidate.write_text("M,N,K\n1,2,3\n", encoding="utf-8") + calls: list[dict] = [] + + async def _fake_integrate(payload, *, session_dir): + calls.append(payload) + if len(calls) == 1: + return { + "status": "failed", + "error_class": "bench_exception", + "decision": "REVERT", + "error": "re-baseline did not succeed", + } + return {"status": "ok", "decision": "KEEP", "new_tput": 110.0, "gain_pct": 10.0} + + monkeypatch.setattr(krh_mod, "integrate_handler", _fake_integrate) + monkeypatch.setattr( + phase, + "_merge_gemm_candidate_with_runtime", + lambda _env_var, env_value: env_value, + ) + result = { + "backend": "forge", + "tuners_run": [ + { + "status": "ok", + "tuner": "dense_bf16", + "improved_shapes": 1, + "env_var": "AITER_CONFIG_DENSE", + "env_value": str(dense_candidate), + }, + ], + } + + await phase._validate_gemm_tuning_e2e(result) + + assert len(calls) == 2 + assert result["e2e_results"]["faults"] == [] + assert result["e2e_results"]["kept"][0]["tuner"] == "dense_bf16" + assert result["decision"] == "KEEP" + @pytest.mark.asyncio async def test_a_stopped_run_leaves_its_tuners_unjudged(self, tmp_path, monkeypatch): """A clock that ran out is not a verdict on the tuners it interrupted.""" @@ -2182,7 +2288,7 @@ async def test_handles_no_candidates_without_rewriting_raw_result(self, tmp_path assert coord.shared_state.optimization_stack == [] @pytest.mark.asyncio - async def test_records_integrate_exception_as_revert(self, tmp_path, monkeypatch): + async def test_records_integrate_exception_as_fault(self, tmp_path, monkeypatch): coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") phase = KernelPhase(coord) dense_candidate = tmp_path / "dense.csv" @@ -2191,7 +2297,13 @@ async def test_records_integrate_exception_as_revert(self, tmp_path, monkeypatch async def _raise_integrate(*_args, **_kwargs): raise RuntimeError("integrate failed") - monkeypatch.setattr(krh_mod, "integrate_handler", _raise_integrate) + calls: list[str] = [] + + async def _counting_raise(*_args, **_kwargs): + calls.append("boom") + raise RuntimeError("integrate failed") + + monkeypatch.setattr(krh_mod, "integrate_handler", _counting_raise) monkeypatch.setattr( phase, "_merge_gemm_candidate_with_runtime", @@ -2213,9 +2325,15 @@ async def _raise_integrate(*_args, **_kwargs): await phase._validate_gemm_tuning_e2e(result) - assert result["decision"] == "REVERT" - assert result["micro_decision"] == "candidate_no_e2e_gain" - assert "integrate failed" in result["e2e_results"]["reverted"][0]["reason"] + assert result["status"] == "failed" + assert result["micro_decision"] == "integrate_fault" + assert result["e2e_gain_pct"] is None + fault = result["e2e_results"]["faults"][0] + assert fault["reason"] == "integrate_fault:handler_exception" + assert fault["fault"] is True + assert fault["fault_attempts"] == 2 + assert len(calls) == 2 + assert result["e2e_results"]["reverted"] == [] class TestBf16DenseFallback: @@ -2660,6 +2778,148 @@ async def test_forge_e2e_rewrites_latest_attempt_history(self, tmp_path, monkeyp assert attempts[0]["best_speedup"] == 1.5 assert coord.shared_state.last_gemm_tuning["decision"] == "REVERT" + @pytest.mark.asyncio + async def test_forge_e2e_keep_names_the_artifact_the_stack_recorded( + self, tmp_path, monkeypatch + ): + """The history row and the stack entry must name the same artifact. + + The breakdown decides ``adopted`` by matching those two strings. Forge + reports per-tuner envs and never set ``tuned_file``, so the history row + carried "" and no KEEP could ever match -- measured across 419 real + attempts, none was reported adopted. + """ + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + fake = _make_integrate([{"decision": "KEEP", "new_tput": 130.0, "gain_pct": 30.0}]) + monkeypatch.setattr(krh_mod, "integrate_handler", fake) + + await coord._handle_gemm_tuning_result( + { + "status": "ok", + "decision": "KEEP", + "best_speedup": 1.5, + "backend": "forge", + "engine": "forge", + "requires_e2e_validation": True, + "recommended_env": {"AITER_DENSE": "/dense.json"}, + "extra_envs": {"AITER_DENSE": "/dense.json"}, + "tuners_run": [ + { + "status": "ok", + "improved_shapes": 3, + "tuner": "dense_gemm", + "env_var": "AITER_DENSE", + "env_value": "/dense.json", + } + ], + } + ) + + stack = coord.shared_state.optimization_stack + assert stack, "a KEEP must land on the stack" + assert stack[-1]["action"] == "gemm_tuning" + attempts = coord.shared_state.gemm_tuning_attempts + assert attempts[0]["decision"] == "KEEP" + assert attempts[0]["tuned_file"], "history row must name the artifact" + assert attempts[0]["tuned_file"] == stack[-1]["tuned_file"] + + @pytest.mark.asyncio + async def test_a_second_round_claims_its_own_artifact(self, tmp_path, monkeypatch): + """Re-tuning the same tuner must not inherit the earlier round's path. + + ``_lift_to_current_best`` skips the stack append when + ``(action, variant_name)`` already matches, and a GEMM variant is named + ``_`` -- so after a second macro cycle re-tunes the same + tuner, the newest stack entry still describes round one. Taking the + artifact from there would make the second attempt claim the first one's + file, and with it the first one's gain. + """ + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + # Keep the candidate env value verbatim so each round's path is distinct + # and the assertion is about provenance, not about merging. + monkeypatch.setattr( + KernelPhase, + "_merge_gemm_candidate_with_runtime", + lambda _self, _env_var, env_value: env_value, + ) + + def _result(env_value: str) -> dict: + return { + "status": "ok", + "decision": "KEEP", + "best_speedup": 1.5, + "backend": "forge", + "engine": "forge", + "requires_e2e_validation": True, + "recommended_env": {"AITER_DENSE": env_value}, + "extra_envs": {"AITER_DENSE": env_value}, + "tuners_run": [ + { + "status": "ok", + "improved_shapes": 3, + "tuner": "dense_gemm", + "env_var": "AITER_DENSE", + "env_value": env_value, + } + ], + } + + monkeypatch.setattr( + krh_mod, + "integrate_handler", + _make_integrate([{"decision": "KEEP", "new_tput": 130.0, "gain_pct": 30.0}]), + ) + await coord._handle_gemm_tuning_result(_result("/round1.json")) + + first_file = coord.shared_state.gemm_tuning_attempts[-1]["tuned_file"] + assert first_file, "round one must name its artifact" + stack_len = len(coord.shared_state.optimization_stack) + + monkeypatch.setattr( + krh_mod, + "integrate_handler", + _make_integrate([{"decision": "KEEP", "new_tput": 160.0, "gain_pct": 23.1}]), + ) + await coord._handle_gemm_tuning_result(_result("/round2.json")) + + # Same (action, variant_name): the append is skipped by design. + assert len(coord.shared_state.optimization_stack) == stack_len + second_file = coord.shared_state.gemm_tuning_attempts[-1]["tuned_file"] + assert second_file and second_file != first_file + + @pytest.mark.asyncio + async def test_forge_e2e_revert_does_not_claim_an_artifact( + self, tmp_path, monkeypatch + ): + """A REVERT has nothing on the stack, so it must not name one.""" + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + fake = _make_integrate([{"decision": "REVERT", "new_tput": 90.0, "gain_pct": -10.0}]) + monkeypatch.setattr(krh_mod, "integrate_handler", fake) + + await coord._handle_gemm_tuning_result( + { + "status": "ok", + "decision": "KEEP", + "backend": "forge", + "engine": "forge", + "requires_e2e_validation": True, + "recommended_env": {"AITER_DENSE": "/dense.json"}, + "extra_envs": {"AITER_DENSE": "/dense.json"}, + "tuners_run": [ + { + "status": "ok", + "improved_shapes": 3, + "tuner": "dense_gemm", + "env_var": "AITER_DENSE", + "env_value": "/dense.json", + } + ], + } + ) + + assert coord.shared_state.optimization_stack == [] + assert not coord.shared_state.gemm_tuning_attempts[0].get("tuned_file") + @pytest.mark.asyncio async def test_forge_no_improvement_but_ck_eligible_routes_to_validator(self, tmp_path, monkeypatch): # a8w8 tuner reported no_improvement but the CK block-scale switch is @@ -3103,7 +3363,7 @@ async def test_all_revert_resets_and_marks_no_gain(self, tmp_path, monkeypatch): assert result["requires_e2e_validation"] is False @pytest.mark.asyncio - async def test_integrate_exception_reverts_tuner(self, tmp_path, monkeypatch): + async def test_integrate_exception_records_fault_not_revert(self, tmp_path, monkeypatch): coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") async def _boom(payload, *, session_dir): @@ -3128,10 +3388,13 @@ async def _boom(payload, *, session_dir): } await coord._validate_gemm_tuning_e2e(result) - assert result["decision"] == "REVERT" - reverted = result["e2e_results"]["reverted"] - assert len(reverted) == 1 - assert reverted[0]["reason"].startswith("RuntimeError") + assert result["status"] == "failed" + assert result["micro_decision"] == "integrate_fault" + assert result["e2e_gain_pct"] is None + faults = result["e2e_results"]["faults"] + assert len(faults) == 1 + assert faults[0]["reason"] == "integrate_fault:handler_exception" + assert result["e2e_results"]["reverted"] == [] @pytest.mark.asyncio async def test_timeout_fallback_when_explore_helper_raises(self, tmp_path, monkeypatch): diff --git a/src/hyperloom/inference_optimizer/tests/test_forge_gemm_durable_persist.py b/src/hyperloom/inference_optimizer/tests/test_forge_gemm_durable_persist.py index 64b6f7cd36..7d4d896f52 100644 --- a/src/hyperloom/inference_optimizer/tests/test_forge_gemm_durable_persist.py +++ b/src/hyperloom/inference_optimizer/tests/test_forge_gemm_durable_persist.py @@ -27,6 +27,15 @@ def _fake_aiter(monkeypatch, tmp_path: Path) -> Path: return aiter_pkg +def _durable(aiter_pkg: Path, name: str) -> Path: + """Where a persisted CSV lands: below model_configs/, not inside it. + + aiter auto-merges everything its non-recursive ``model_configs/*.csv`` glob + finds when the env var is unset, so the copy has to sit one level down. + """ + return aiter_pkg / "configs" / "model_configs" / "hyperloom" / name + + def test_persist_copies_into_aiter_config_and_snapshots(tmp_path, monkeypatch): aiter_pkg = _fake_aiter(monkeypatch, tmp_path) ws = tmp_path / "ws" @@ -39,18 +48,53 @@ def test_persist_copies_into_aiter_config_and_snapshots(tmp_path, monkeypatch): extra, model_path="/models/Qwen3-14B-FP8", session_dir=ws ) - dst = aiter_pkg / "configs" / "model_configs" / "a8w8_blockscale_tuned_gemm_qwen3-14b-fp8.csv" - assert dst.is_file() # copied where aiter reads it + dst = _durable(aiter_pkg, "a8w8_blockscale_tuned_gemm_qwen3-14b-fp8.csv") + assert dst.is_file() # copied where the env var can reach it assert out["AITER_CONFIG_GEMM_A8W8_BLOCKSCALE"] == str(dst) # env repointed to durable path assert snap and Path(snap).is_dir() # durable snapshot dir assert (Path(snap) / "manifest.json").is_file() - assert (Path(snap) / "files" / "configs" / "model_configs" / dst.name).is_file() + assert ( + Path(snap) / "files" / "configs" / "model_configs" / "hyperloom" / dst.name + ).is_file() # snapshot must live under the DURABLE optimization_stack/src (survives run # cleanup), NOT the ephemeral runs/gemm_tuning workspace (#2 recipe-portable). assert "optimization_stack" in Path(snap).parts and "src" in Path(snap).parts assert "runs" not in Path(snap).parts +def test_persist_keeps_the_copy_out_of_aiters_auto_merge_scan(tmp_path, monkeypatch): + """The copy must be invisible to aiter's env-less table scan. + + ``jit/core.py::get_config_file`` globs ``model_configs/*{table}*.csv`` and + merges everything it finds whenever the env var is unset -- which is the + common case for a plain server start. A candidate persisted before E2E has + ruled on it would reach every later server that way, so a REVERT would read + as reverted while the table stayed silently in effect. Observed for real: a + V4-Flash run merged dsv3's table, so the scan does not even discriminate by + model. + + One level down is enough: the glob is not recursive, and the env var still + points at the file. + """ + aiter_pkg = _fake_aiter(monkeypatch, tmp_path) + ws = tmp_path / "ws" + ws.mkdir() + src = ws / "tuned.csv" + src.write_text("gfx,cu_num,M,N,K,splitK\ngfx950,256,64,5120,5120,2\n", encoding="utf-8") + + out, _snap = rh._persist_forge_gemm_csv_durably( + {"AITER_CONFIG_GEMM_BF16": str(src)}, + model_path="/models/Qwen3-14B-FP8", + session_dir=ws, + ) + + dst = Path(out["AITER_CONFIG_GEMM_BF16"]) + assert dst.is_file(), "the copy still has to exist for the env var to reach" + model_configs = aiter_pkg / "configs" / "model_configs" + assert list(model_configs.glob("*bf16_tuned_gemm*.csv")) == [] + assert dst.parent != model_configs + + def test_persist_missing_source_csv_is_noop(tmp_path, monkeypatch): _fake_aiter(monkeypatch, tmp_path) extra = {"AITER_CONFIG_GEMM_A8W8_BLOCKSCALE": str(tmp_path / "nope.csv")} @@ -95,7 +139,49 @@ def _boom(**kwargs): extra, model_path="/models/Qwen3-14B-FP8", session_dir=ws ) - dst = aiter_pkg / "configs" / "model_configs" / "a8w8_blockscale_tuned_gemm_qwen3-14b-fp8.csv" + dst = _durable(aiter_pkg, "a8w8_blockscale_tuned_gemm_qwen3-14b-fp8.csv") assert dst.is_file() # copy committed despite the snapshot failure assert out["AITER_CONFIG_GEMM_A8W8_BLOCKSCALE"] == str(dst) # repoint SURVIVES assert snap == "" # snapshot dir empty (it failed), but durability is kept + + +def test_persist_fmoe_csv_uses_tuned_fmoe_stem(tmp_path, monkeypatch): + aiter_pkg = _fake_aiter(monkeypatch, tmp_path) + ws = tmp_path / "ws" + ws.mkdir() + src = ws / "tuned_fmoe.csv" + src.write_text("cu_num,token,model_dim,inter_dim,quantType\n304,16,4096,512,14\n", encoding="utf-8") + + extra = {"AITER_CONFIG_FMOE": str(src)} + out, snap = rh._persist_forge_gemm_csv_durably( + extra, model_path="/models/DeepSeek-V4-Flash", session_dir=ws + ) + + dst = _durable(aiter_pkg, "tuned_fmoe_deepseek-v4-flash.csv") + assert dst.is_file() + assert out["AITER_CONFIG_FMOE"] == str(dst) + assert snap and Path(snap).is_dir() + + +def test_persist_copies_dense_and_fmoe_together(tmp_path, monkeypatch): + aiter_pkg = _fake_aiter(monkeypatch, tmp_path) + ws = tmp_path / "ws" + ws.mkdir() + dense = ws / "dense.csv" + dense.write_text("gfx,M,N,K,splitK\ngfx950,64,5120,5120,2\n", encoding="utf-8") + fmoe = ws / "fmoe.csv" + fmoe.write_text("cu_num,token\n304,16\n", encoding="utf-8") + + extra = { + "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE": str(dense), + "AITER_CONFIG_FMOE": str(fmoe), + } + out, snap = rh._persist_forge_gemm_csv_durably( + extra, model_path="/models/Qwen3-14B-FP8", session_dir=ws + ) + + assert out["AITER_CONFIG_GEMM_A8W8_BLOCKSCALE"].endswith( + "a8w8_blockscale_tuned_gemm_qwen3-14b-fp8.csv" + ) + assert out["AITER_CONFIG_FMOE"].endswith("tuned_fmoe_qwen3-14b-fp8.csv") + assert snap and (Path(snap) / "manifest.json").is_file() diff --git a/src/hyperloom/inference_optimizer/tests/test_geak_breakdown_unit.py b/src/hyperloom/inference_optimizer/tests/test_geak_breakdown_unit.py index 8e07ffe953..d4609a8a86 100644 --- a/src/hyperloom/inference_optimizer/tests/test_geak_breakdown_unit.py +++ b/src/hyperloom/inference_optimizer/tests/test_geak_breakdown_unit.py @@ -679,6 +679,139 @@ async def test_sweep_via_geak_requires_existing_bench_script(tmp_path: Path) -> assert result["error_class"] == "missing_bench_script" +def test_collect_gemm_tuning_prefers_e2e_gain_over_micro_speedup() -> None: + from hyperloom.inference_optimizer.breakdown.collectors.kernels import collect_gemm_tuning + + out = collect_gemm_tuning( + { + "baseline_tput": 1000.0, + "gemm_tuning_attempts": [ + { + "engine": "forge", + "status": "complete", + "decision": "KEEP", + "best_speedup": 1.5, + "e2e_gain_pct": 9.26, + "e2e_validated": True, + "tuned_file": "/tmp/tuned.csv", + } + ], + } + ) + + run = out["runs"][0] + assert run["gain_pct"] == pytest.approx(9.26) + assert run["tuned_tput"] == pytest.approx(1092.6) + assert run["best_speedup"] == pytest.approx(1.5) + + +def _gemm_state_with_keep(*, attempt_tuned_file: str) -> dict: + """A session whose forge run was kept and lifted onto the stack.""" + return { + "baseline_tput": 1000.0, + "cumulative_gain_validated_stack_len": 1, + "optimization_stack": [ + { + "action": "gemm_tuning", + "tuned_file": "/ws/merged_tuned_fmoe.csv", + "gain_pct": 9.26, + } + ], + "gemm_tuning_attempts": [ + { + "engine": "forge", + "status": "complete", + "decision": "KEEP", + "e2e_gain_pct": 9.26, + "e2e_validated": True, + "tuned_file": attempt_tuned_file, + } + ], + } + + +def test_collect_gemm_tuning_marks_a_kept_run_adopted() -> None: + """A forge run whose artifact reached the stack must read as adopted. + + Across 419 real forge attempts this was never true: the attempt row carried + no ``tuned_file`` at all, so the stack lookup matched on the empty string + and every KEEP -- including ones measuring +49% -- was reported as not + adopted. + """ + from hyperloom.inference_optimizer.breakdown.collectors.kernels import collect_gemm_tuning + + out = collect_gemm_tuning( + _gemm_state_with_keep(attempt_tuned_file="/ws/merged_tuned_fmoe.csv") + ) + + run = out["runs"][0] + assert run["adopted"] is True + assert run["gain_pct"] == pytest.approx(9.26) + + +def test_collect_gemm_tuning_leaves_an_unlifted_run_unadopted() -> None: + """Fail-open on the empty case must not make every run look adopted.""" + from hyperloom.inference_optimizer.breakdown.collectors.kernels import collect_gemm_tuning + + out = collect_gemm_tuning(_gemm_state_with_keep(attempt_tuned_file="")) + + assert out["runs"][0]["adopted"] is False + + +class TestCandidateTunedFile: + """The artifact a KEEP adopted, named from the candidate's own env. + + One KEEP is described by three different path strings -- the durable copy in + aiter's config dir, the tuner-workspace original, and the E2E merge product + -- so the attempt row cannot re-derive the one the stack holds. The way out + is not to read the stack back either: reading it back picks up whatever entry + is newest, and ``_lift_to_current_best`` skips the append when + ``(action, variant_name)`` already matches, which a second macro cycle + re-tuning the same tuner does. The attempt would then claim the previous + round's artifact and its gain. Both sides take the value from this one + function instead, so they are the same string by construction. + """ + + def test_prefers_the_candidate_env_var(self) -> None: + """The candidate's own key wins over whatever the env happens to list + first -- a stacked env carries the earlier KEEPs' vars too.""" + from hyperloom.orchestrator.phases.kernel import _candidate_tuned_file + + # Deliberately not first: falling back to insertion order would pick + # the wrong artifact and still look right if the target led the dict. + env = { + "AITER_CONFIG_GEMM_BF16": "/ws/earlier_keep.csv", + "AITER_CONFIG_FMOE": "/ws/merged_tuned_fmoe.csv", + } + assert _candidate_tuned_file(env, "AITER_CONFIG_FMOE") == "/ws/merged_tuned_fmoe.csv" + + def test_falls_back_to_the_only_value_present(self) -> None: + """A candidate whose env_var is not the key its env carries.""" + from hyperloom.orchestrator.phases.kernel import _candidate_tuned_file + + env = {"AITER_CONFIG_GEMM_A8W8": "/ws/tuned_a8w8.csv"} + assert _candidate_tuned_file(env, "AITER_CONFIG_FMOE") == "/ws/tuned_a8w8.csv" + + def test_empty_env_yields_no_claim(self) -> None: + from hyperloom.orchestrator.phases.kernel import _candidate_tuned_file + + assert _candidate_tuned_file({}, "AITER_CONFIG_FMOE") == "" + + def test_tolerates_malformed_input(self) -> None: + from hyperloom.orchestrator.phases.kernel import _candidate_tuned_file + + assert _candidate_tuned_file({"AITER_CONFIG_FMOE": None}, "AITER_CONFIG_FMOE") == "" + assert _candidate_tuned_file({"AITER_CONFIG_FMOE": ""}, "AITER_CONFIG_FMOE") == "" + assert _candidate_tuned_file(None, "AITER_CONFIG_FMOE") == "" + assert _candidate_tuned_file({"k": 42}, "k") == "42" + + def test_the_stack_reader_is_gone(self) -> None: + """Reading the newest stack entry is what allowed the false claim.""" + from hyperloom.orchestrator.phases import kernel as kernel_phase + + assert not hasattr(kernel_phase, "_adopted_tuned_file") + + def test_collect_geak_backfill_fires_on_no_gain(tmp_path: Path) -> None: # A run stamped ``no_gain`` on the COLD basis can still hold a measured hot # win and genuine KEEP rows in the journey. Attribution must not be dropped. diff --git a/src/hyperloom/inference_optimizer/tests/test_gemm_bf16_aiter_routing.py b/src/hyperloom/inference_optimizer/tests/test_gemm_bf16_aiter_routing.py index 32d6da4ccb..ac97bb6016 100644 --- a/src/hyperloom/inference_optimizer/tests/test_gemm_bf16_aiter_routing.py +++ b/src/hyperloom/inference_optimizer/tests/test_gemm_bf16_aiter_routing.py @@ -40,15 +40,38 @@ def _log(tmp_path, text: str) -> str: AITER_FUSED_MOE = ( "(Worker_TP0 pid=1) [aiter] [fused_moe] using 2stage default for " - "(256, 8192, 3072, 1536, 128, 4, 'ActivationType.Swiglu', 'torch.bfloat16', " + "('gfx950', 256, 256, 8192, 1536, 128, 4, 'ActivationType.Swiglu', 'torch.bfloat16', " "'torch.float8_e4m3fn', 'torch.float4_e2m1fn_x2', 'QuantType.per_1x32', True, False)" ) AITER_FUSED_MOE_BF16_FP4 = ( "(Worker_TP0 pid=1) [aiter] [fused_moe] using 2stage default for " - "(256, 8192, 3072, 3072, 128, 4, 'ActivationType.Swiglu', 'torch.bfloat16', " + "('gfx950', 256, 256, 8192, 3072, 128, 4, 'ActivationType.Swiglu', 'torch.bfloat16', " "'torch.bfloat16', 'torch.float4_e2m1fn_x2', 'QuantType.per_1x32', True, False)" ) +# Verbatim lines from a production server.log, one per wording aiter emits. +# Kept literal because the previous hand-written fixtures dropped the leading +# gfx field, which let a regex that could never match a real log pass its tests. +REAL_2STAGE_DEFAULT = ( + "[aiter] [fused_moe] using 2stage default for ('gfx950', 256, 256, 4096, 512, 256, 6, " + "'ActivationType.Silu', 'torch.bfloat16', 'torch.float8_e4m3fn', " + "'torch.float4_e2m1fn_x2', 'QuantType.per_1x32', True, False)" +) +REAL_NO_TUNED_FLYDSL = ( + "[aiter] [fused_moe] no tuned FlyDSL config for ('gfx950', 256, 256, 4096, 512, 256, 6, " + "'ActivationType.Silu', 'torch.bfloat16', 'torch.float8_e4m3fn', " + "'torch.float4_e2m1fn_x2', 'QuantType.per_1x32', True, False), using heuristic " + "FlyDSL fallback (kn1='flydsl_moe1_afp8_wfp4_bf16_t32x128x256_w2_gui', " + "kn2='flydsl_moe2_afp8_wfp4_bf16_t32x128x256_atomic_bnt2')" +) +REAL_2STAGE_WITH_KERNEL_NAMES = ( + "(Worker_TP7 pid=26394) [aiter] [fused_moe] using 2stage " + "(kernelName1='flydsl_moe1_afp8_wfp4_bf16_t64x128x256_w4_bnt0_gui_fp8', " + "kernelName2='opus_moe2_afp8_wfp4_fp8_t64x256x256_sbm64_rbn3584') for " + "('gfx950', 256, 8192, 7168, 512, 384, 6, 'ActivationType.Silu', 'torch.bfloat16', " + "'torch.float8_e4m3fn', 'torch.float4_e2m1fn_x2', 'QuantType.per_1x32', True, False)" +) + class TestAiterServingEvidence: def test_detects_bf16_dense(self, tmp_path): @@ -69,10 +92,19 @@ def test_missing_log(self, tmp_path): assert krh._aiter_serving_evidence(str(tmp_path / "absent.log")) == set() -def _moe_tuple(q_a: str, q_w: str, q_type: str = "QuantType.per_1x32") -> str: +def _moe_tuple( + q_a: str, + q_w: str, + q_type: str = "QuantType.per_1x32", + *, + inter_dim: int = 3072, + expert: int = 128, + topk: int = 4, +) -> str: return ( "(Worker_TP0 pid=1) [aiter] [fused_moe] using 2stage default for " - f"(256, 8192, 3072, 3072, 128, 4, 'ActivationType.Swiglu', 'torch.bfloat16', " + f"('gfx950', 256, 8192, 3072, {inter_dim}, {expert}, {topk}, " + f"'ActivationType.Swiglu', 'torch.bfloat16', " f"'{q_a}', '{q_w}', '{q_type}', True, False)" ) @@ -95,13 +127,28 @@ def test_unquantised_bf16_moe_is_supported(self, tmp_path): log = _log(tmp_path, _moe_tuple("torch.bfloat16", "torch.bfloat16", "QuantType.No")) assert krh._aiter_ck_moe_tuner_supports(log) - def test_any_unsupported_combo_blocks_the_model(self, tmp_path): - """gpt-oss logs both combos; the unsupported one has to win.""" + def test_a_mixed_log_stays_tunable_because_rows_are_filtered(self, tmp_path): + """One checkpoint dispatches several pairs; the tunable ones still count. + + Measured in production: the same model logs both a BF16-activation and an + FP8-activation problem. Blocking the whole model on the unsupported one + would forfeit the tunable half, so the untunable rows are dropped when the + tuning input is written instead. + """ log = _log( tmp_path, _moe_tuple("torch.float8_e4m3fn", "torch.float4_e2m1fn_x2") + "\n" - + _moe_tuple("torch.bfloat16", "torch.float4_e2m1fn_x2"), + + _moe_tuple("torch.bfloat16", "torch.float4_e2m1fn_x2", expert=257, topk=5), + ) + assert krh._aiter_ck_moe_tuner_supports(log) + + def test_an_entirely_unsupported_log_is_rejected(self, tmp_path): + log = _log( + tmp_path, + _moe_tuple("torch.bfloat16", "torch.float4_e2m1fn_x2") + + "\n" + + _moe_tuple("torch.bfloat16", "torch.float4_e2m1fn_x2", expert=257, topk=5), ) assert not krh._aiter_ck_moe_tuner_supports(log) @@ -110,6 +157,149 @@ def test_moe_evidence_without_a_parseable_tuple_defers_to_forge(self, tmp_path): assert krh._aiter_ck_moe_tuner_supports(log) +class TestDispatchKeyExtraction: + """The regex must match every wording aiter actually emits. + + A hand-written fixture previously omitted the leading gfx field, so a regex + that could not match a single real log line passed its tests while silently + disabling the dtype gate in production. + """ + + def test_matches_the_plain_default_wording(self, tmp_path): + keys = krh._aiter_fused_moe_dispatch_keys(_log(tmp_path, REAL_2STAGE_DEFAULT)) + assert len(keys) == 1 + assert keys[0]["inter_dim"] == "512" + assert keys[0]["q_dtype_a"] == "torch.float8_e4m3fn" + assert keys[0]["q_dtype_w"] == "torch.float4_e2m1fn_x2" + assert keys[0]["expert"] == "256" + assert keys[0]["topk"] == "6" + + def test_matches_the_flydsl_fallback_wording(self, tmp_path): + keys = krh._aiter_fused_moe_dispatch_keys(_log(tmp_path, REAL_NO_TUNED_FLYDSL)) + assert len(keys) == 1 + assert keys[0]["inter_dim"] == "512" + + def test_matches_the_wording_that_interposes_kernel_names(self, tmp_path): + """This form puts its own parenthesised group before the tuple.""" + keys = krh._aiter_fused_moe_dispatch_keys( + _log(tmp_path, REAL_2STAGE_WITH_KERNEL_NAMES) + ) + assert len(keys) == 1 + assert keys[0]["model_dim"] == "7168" + assert keys[0]["expert"] == "384" + + def test_dedupes_on_everything_but_the_token_count(self, tmp_path): + a = REAL_2STAGE_DEFAULT + b = REAL_2STAGE_DEFAULT.replace("256, 256, 4096", "256, 512, 4096") + keys = krh._aiter_fused_moe_dispatch_keys(_log(tmp_path, a + "\n" + b)) + assert len(keys) == 1 + + def test_keeps_distinct_problems_from_one_model(self, tmp_path): + """The EP path inflates expert/topk by one; that is a separate problem.""" + a = _moe_tuple("torch.float8_e4m3fn", "torch.float4_e2m1fn_x2") + b = _moe_tuple("torch.float8_e4m3fn", "torch.float4_e2m1fn_x2", expert=129, topk=5) + keys = krh._aiter_fused_moe_dispatch_keys(_log(tmp_path, a + "\n" + b)) + assert len(keys) == 2 + + def test_missing_log(self, tmp_path): + assert krh._aiter_fused_moe_dispatch_keys("") == [] + assert krh._aiter_fused_moe_dispatch_keys(str(tmp_path / "absent.log")) == [] + + +class TestDtypePairSupport: + """Mirrors the four kernel families in aiter's CK MoE codegen.""" + + def test_supported_pairs(self): + for act, weight in ( + ("torch.bfloat16", "torch.bfloat16"), + ("torch.float16", "torch.float16"), + ("torch.float8_e4m3fn", "torch.float8_e4m3fn"), + ("torch.float8_e4m3fn", "torch.float4_e2m1fn_x2"), + ("torch.float8_e4m3fnuz", "torch.float4_e2m1fn_x2"), + ("torch.float4_e2m1fn_x2", "torch.float4_e2m1fn_x2"), + ): + assert krh._aiter_moe_dtype_pair_supported(act, weight), (act, weight) + + def test_bf16_activation_with_fp4_weight_is_the_known_rejection(self): + assert not krh._aiter_moe_dtype_pair_supported( + "torch.bfloat16", "torch.float4_e2m1fn_x2" + ) + + def test_int8_activation_does_not_qualify_for_the_a8w4_family(self): + """The a8w4 branch requires an FP8 activation specifically.""" + assert not krh._aiter_moe_dtype_pair_supported( + "torch.int8", "torch.float4_e2m1fn_x2" + ) + + +class TestWriteFmoeUntunedCsvFromLog: + def test_writes_one_row_per_problem_and_token(self, tmp_path): + path, report = krh._write_fmoe_untuned_csv_from_log( + _log(tmp_path, REAL_2STAGE_DEFAULT), [4, 512], tmp_path / "ws" + ) + rows = [ + line for line in open(path, encoding="utf-8").read().splitlines() if line + ] + assert rows[0].split(",") == [ + "token", "model_dim", "inter_dim", "expert", "topk", "act_type", "dtype", + "q_dtype_a", "q_dtype_w", "q_type", "use_g1u1", "doweight_stage1", + ] + assert len(rows) == 3 # header + 2 tokens + assert rows[1] == ( + "4,4096,512,256,6,ActivationType.Silu,torch.bfloat16," + "torch.float8_e4m3fn,torch.float4_e2m1fn_x2,QuantType.per_1x32,1,0" + ) + assert report["observed"] == 1 + assert report["tunable"] == 1 + assert report["dropped_unsupported"] == [] + + def test_drops_pairs_the_tuner_would_reject(self, tmp_path): + """One unsupported row aborts the whole aiter tuner run, so filter first.""" + log = _log( + tmp_path, + _moe_tuple("torch.float8_e4m3fn", "torch.float4_e2m1fn_x2") + + "\n" + + _moe_tuple("torch.bfloat16", "torch.float4_e2m1fn_x2", expert=129), + ) + path, report = krh._write_fmoe_untuned_csv_from_log(log, [8], tmp_path / "ws") + + body = open(path, encoding="utf-8").read() + assert "torch.bfloat16,torch.float4_e2m1fn_x2" not in body + assert report["observed"] == 2 + assert report["tunable"] == 1 + assert report["dropped_unsupported"] == ["torch.bfloat16/torch.float4_e2m1fn_x2"] + + def test_no_tunable_problem_yields_no_csv(self, tmp_path): + log = _log(tmp_path, _moe_tuple("torch.bfloat16", "torch.float4_e2m1fn_x2")) + path, report = krh._write_fmoe_untuned_csv_from_log(log, [8], tmp_path / "ws") + + assert path == "" + assert report["observed"] == 1 + assert report["tunable"] == 0 + + def test_unwritable_workspace_costs_only_the_moe_input(self, tmp_path, monkeypatch): + """A full disk must not take the dense tuners down with the MoE one.""" + log = _log(tmp_path, _moe_tuple("torch.float8_e4m3fn", "torch.float4_e2m1fn_x2")) + + def _boom(*_args, **_kwargs): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(krh.Path, "write_text", _boom) + + path, report = krh._write_fmoe_untuned_csv_from_log(log, [8], tmp_path / "ws") + + assert path == "" + assert report["tunable"] == 1 + assert "No space left on device" in report["write_error"] + + def test_no_moe_evidence_yields_no_csv(self, tmp_path): + path, report = krh._write_fmoe_untuned_csv_from_log( + _log(tmp_path, "INFO server started\n"), [8], tmp_path / "ws" + ) + assert path == "" + assert report["observed"] == 0 + + class TestResolveVllmAiterRouting: def test_dense_bf16_model(self, tmp_path): flags = krh._resolve_vllm_aiter_routing( diff --git a/src/hyperloom/inference_optimizer/tests/test_gemm_shape_coverage.py b/src/hyperloom/inference_optimizer/tests/test_gemm_shape_coverage.py index 7c598526d3..c25d00504e 100644 --- a/src/hyperloom/inference_optimizer/tests/test_gemm_shape_coverage.py +++ b/src/hyperloom/inference_optimizer/tests/test_gemm_shape_coverage.py @@ -1,12 +1,19 @@ # SPDX-FileCopyrightText: 2025 Advanced Micro Devices, Inc. # SPDX-License-Identifier: MIT -"""Unit tests for aiter tuned-GEMM shape alignment and coverage reporting.""" +"""Unit tests for aiter tuned-GEMM shape alignment and coverage reporting. + +Also covers the fail-open guards around that reporting: its verdict can block a +KEEP, so every way it can fail to reach one has to degrade to "undetermined" +rather than to "the artifact did not apply". +""" from __future__ import annotations import json +import pytest + from hyperloom.orchestrator.kernel.gemm_shape_coverage import ( aiter_lookup_keys, aiter_padded_m_coarse, @@ -285,6 +292,19 @@ def test_reads_shape_keys(self, tmp_path): def test_missing_file_is_empty(self, tmp_path): assert tuned_csv_shapes(tmp_path / "nope.csv") == set() + def test_an_fmoe_csv_yields_no_dense_shapes(self, tmp_path): + """An MoE table has no M,N,K columns; reading one as dense would invent + shapes and report coverage against a schema it never described.""" + path = tmp_path / "tuned_fmoe.csv" + path.write_text( + "token,model_dim,inter_dim,expert,topk,act_type,dtype," + "q_dtype_a,q_dtype_w,q_type,use_g1u1,doweight_stage1,kernelName\n" + "256,4096,512,256,6,ActivationType.Silu,torch.bfloat16," + "torch.float8_e4m3fn,torch.float4_e2m1fn_x2,QuantType.per_1x32,1,0,kernel_a\n", + encoding="utf-8", + ) + assert tuned_csv_shapes(path) == set() + def test_coverage_flags_an_unreachable_artifact(self, tmp_path): """Reproduces the observed failure: raw-M rows, drifted runtime M.""" path = self._csv(tmp_path, [(1076, 5120, 17408), (4142, 5120, 5120)]) @@ -310,3 +330,226 @@ def test_no_requested_shapes(self): report = tuned_config_coverage([(1, 2, 3)], []) assert report["requested"] == 0 assert report["coverage_pct"] is None + + +class TestCoverageGateDoesNotBlockOnMissingEvidence: + """The coverage report can block a KEEP, so it must never guess. + + A report of 0% is a claim the runtime could not reach the tuned rows. When + the CSV yields no keys at all, we have not established that -- we have + failed to read our own artifact. Reporting it as 0% lets an unreadable file + revert a candidate whose throughput genuinely improved, which is the exact + conflation this change set exists to remove. + """ + + ENVS = {"AITER_CONFIG_GEMM": ""} + LOOKUP_LINE = ( + "[aiter] shape is M:1082, N:5120, K:17408, not found tuned config in " + "/x/candidate.csv, will use default config!" + ) + HEADER = "gfx,cu_num,M,N,K,libtype,kernelId,splitK,us,kernelName,tflops,bw,errRatio" + + def _phase(self, tmp_path): + from types import SimpleNamespace + + run_dir = tmp_path / "runs" / "integrate" / "integrate-gemm_tune_aiter_dense" + run_dir.mkdir(parents=True) + (run_dir / "server.log").write_text(self.LOOKUP_LINE + "\n", encoding="utf-8") + return SimpleNamespace(session_dir=tmp_path) + + def _call(self, phase, csv_path): + """Exercise the body directly, so a bound-method slip cannot fake a pass.""" + from hyperloom.orchestrator.phases.kernel import KernelPhase + + return KernelPhase._gemm_tuned_config_coverage_impl( + phase, "aiter_dense", {"AITER_CONFIG_GEMM": str(csv_path)} + ) + + def test_unreadable_csv_is_undetermined_not_zero_coverage(self, tmp_path): + phase = self._phase(tmp_path) + empty = tmp_path / "candidate.csv" + empty.write_text("", encoding="utf-8") + + assert self._call(phase, empty) is None + + def test_missing_csv_is_undetermined(self, tmp_path): + phase = self._phase(tmp_path) + + assert self._call(phase, tmp_path / "absent.csv") is None + + def test_csv_without_shape_columns_is_undetermined(self, tmp_path): + phase = self._phase(tmp_path) + odd = tmp_path / "candidate.csv" + odd.write_text("a,b,c\n1,2,3\n", encoding="utf-8") + + assert self._call(phase, odd) is None + + def test_readable_csv_with_wrong_keys_still_reports_zero(self, tmp_path): + """Fail-open on unreadable input must not weaken the real check.""" + phase = self._phase(tmp_path) + wrong = tmp_path / "candidate.csv" + wrong.write_text( + f"{self.HEADER}\ngfx950,256,4096,5120,5120,ck,0,0,1.0,name,1,1,0\n", + encoding="utf-8", + ) + + report = self._call(phase, wrong) + assert report is not None + assert report["artifact_applied"] is False + assert report["coverage_pct"] == 0.0 + + def test_matching_csv_reports_applied(self, tmp_path): + phase = self._phase(tmp_path) + good = tmp_path / "candidate.csv" + good.write_text( + f"{self.HEADER}\ngfx950,256,1088,5120,17408,ck,0,0,1.0,name,1,1,0\n", + encoding="utf-8", + ) + + report = self._call(phase, good) + assert report is not None + assert report["artifact_applied"] is True + assert report["coverage_pct"] == 100.0 + + def test_unexpected_failure_is_undetermined(self, tmp_path): + """The wrapper swallows anything the body throws (it can block a KEEP).""" + from types import SimpleNamespace + + from hyperloom.orchestrator.phases.kernel import KernelPhase + + def _boom(*_args, **_kwargs): + raise RuntimeError("coverage exploded") + + phase = SimpleNamespace( + session_dir=tmp_path, + _gemm_tuned_config_coverage_impl=_boom, + ) + + assert ( + KernelPhase._gemm_tuned_config_coverage(phase, "fmoe_ck", self.ENVS) + is None + ) + + +class TestSafeMtime: + def test_missing_path_sorts_last_instead_of_raising(self, tmp_path): + from hyperloom.orchestrator.phases.kernel import _safe_mtime + + assert _safe_mtime(tmp_path / "gone.log") == 0.0 + + def test_existing_path_returns_its_mtime(self, tmp_path): + from hyperloom.orchestrator.phases.kernel import _safe_mtime + + path = tmp_path / "server.log" + path.write_text("x", encoding="utf-8") + assert _safe_mtime(path) == path.stat().st_mtime + + +class TestE2EValidationFailsOpen: + """E2E validation owns the coverage check, so its own failure cannot escape. + + Both entrypoints into gemm tuning guard only the tuning call, not the + validation that follows it. An exception escaping here takes the KERNEL + phase down over a candidate that simply went unmeasured. + """ + + def _phase(self, tmp_path, validate): + from types import SimpleNamespace + + recorded: list[dict] = [] + saved: list[object] = [] + state = SimpleNamespace( + record_gemm_tuning=recorded.append, + save=saved.append, + macro_cycle=0, + ) + return SimpleNamespace( + session_dir=tmp_path, + shared_state=state, + _sync_profile_state_after_gemm_roofline=lambda _r: None, + _validate_gemm_tuning_e2e=validate, + ), recorded + + @pytest.mark.asyncio + async def test_exception_is_recorded_as_a_fault_not_raised(self, tmp_path): + from hyperloom.orchestrator.phases.kernel import KernelPhase + + async def _boom(_result): + raise RuntimeError("e2e exploded") + + phase, recorded = self._phase(tmp_path, _boom) + result: dict = {"backend": "forge"} + + await KernelPhase._handle_gemm_tuning_result(phase, result) + + assert recorded == [result] + fault = result["e2e_results"]["faults"][0] + assert fault["error_class"] == "e2e_validation_exception" + assert "RuntimeError: e2e exploded" in fault["error"] + + @pytest.mark.asyncio + async def test_the_unmeasured_envelope_is_neutralised(self, tmp_path): + """An arm that raised was never measured, so it must not read as a KEEP. + + Recording the fault while leaving the bridge's KEEP envelope in place + would let Orchestration bundle an integrate against it. + """ + from hyperloom.orchestrator.phases.kernel import KernelPhase + + async def _boom(_result): + raise RuntimeError("e2e exploded") + + phase, _ = self._phase(tmp_path, _boom) + result: dict = { + "backend": "forge", + "decision": "KEEP", + "requires_e2e_validation": True, + "recommended_env": {"AITER_CONFIG_FMOE": "/ws/tuned_fmoe.csv"}, + "extra_envs": {"AITER_CONFIG_FMOE": "/ws/tuned_fmoe.csv"}, + } + + await KernelPhase._handle_gemm_tuning_result(phase, result) + + assert result["decision"] == "REVERT" + assert result["requires_e2e_validation"] is False + assert result["e2e_validated"] is False + assert not result["recommended_env"] + assert not result["extra_envs"] + # The reason still has to be legible, not just absent. + assert result["e2e_results"]["faults"][0]["error_class"] == "e2e_validation_exception" + + @pytest.mark.asyncio + async def test_existing_faults_are_preserved(self, tmp_path): + from hyperloom.orchestrator.phases.kernel import KernelPhase + + async def _boom(_result): + raise ValueError("second failure") + + phase, _ = self._phase(tmp_path, _boom) + result: dict = { + "backend": "forge", + "e2e_results": {"faults": [{"tuner": "fmoe_ck", "error_class": "server_died"}]}, + } + + await KernelPhase._handle_gemm_tuning_result(phase, result) + + faults = result["e2e_results"]["faults"] + assert [f["error_class"] for f in faults] == [ + "server_died", + "e2e_validation_exception", + ] + + @pytest.mark.asyncio + async def test_success_path_adds_no_fault(self, tmp_path): + from hyperloom.orchestrator.phases.kernel import KernelPhase + + async def _ok(_result): + return None + + phase, recorded = self._phase(tmp_path, _ok) + result: dict = {"backend": "forge"} + + await KernelPhase._handle_gemm_tuning_result(phase, result) + + assert recorded == [result] + assert "e2e_results" not in result diff --git a/src/hyperloom/inference_optimizer/tests/test_kernel_request_handlers_units.py b/src/hyperloom/inference_optimizer/tests/test_kernel_request_handlers_units.py index c0c8c35116..f3734a2daa 100644 --- a/src/hyperloom/inference_optimizer/tests/test_kernel_request_handlers_units.py +++ b/src/hyperloom/inference_optimizer/tests/test_kernel_request_handlers_units.py @@ -1380,8 +1380,10 @@ async def _unexpected_subprocess(_cmd, *, timeout_sec): result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) - assert result["status"] == "failed" + # Skipped, not failed: forge never started, so it has no verdict. + assert result["status"] == "skipped" assert result["error_class"] == "model_path_unavailable" + assert result["skip_reason"] assert subprocess_called is False @pytest.mark.asyncio @@ -1410,7 +1412,7 @@ async def _unexpected_subprocess(_cmd, *, timeout_sec): result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) assert missing_model_dir.is_absolute() - assert result["status"] == "failed" + assert result["status"] == "skipped" assert result["error_class"] == "model_path_unavailable" assert subprocess_called is False @@ -1446,6 +1448,390 @@ async def _fake_subprocess(cmd, *, timeout_sec): assert result["status"] == "failed" assert result["backend"] == "forge" + # ---- forge wording -> coordinator decision ------------------------------ + # forge reports seven micro_decision wordings; the bridge handled four, and + # the three it missed read in the breakdown like a genuine no_improvement. + + @pytest.mark.asyncio + async def test_a_partial_wording_is_reverted_and_named(self, tmp_path, monkeypatch): + """A barren ``partial_failure`` is a REVERT that still names itself.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps({"status": "ok", "micro_decision": "partial_failure"}) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, sentinel, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) + + assert result["decision"] == "REVERT" + assert result["error_class"] == "forge_partial_failure" + + @pytest.mark.asyncio + async def test_an_empty_run_is_reverted_and_says_so(self, tmp_path, monkeypatch): + """Writing zero rows is not the same outcome as finding nothing.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps({"status": "ok", "micro_decision": "empty_output"}) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, sentinel, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) + + assert result["decision"] == "REVERT" + assert result["error_class"], "an empty run must name itself" + assert "empty" in result["error_class"] + + @pytest.mark.asyncio + async def test_a_genuine_no_improvement_stays_unadorned(self, tmp_path, monkeypatch): + """The distinction only works if the ordinary case stays ordinary.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps({"status": "ok", "micro_decision": "no_improvement"}) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, sentinel, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) + + assert result["decision"] == "REVERT" + assert not result.get("error_class") + + @pytest.mark.asyncio + async def test_a_tuner_error_class_reaches_the_envelope(self, tmp_path, monkeypatch): + """A crash a tuner named must be visible where the breakdown reads.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps( + { + "status": "failed", + "micro_decision": "failed", + "tuners_run": [ + { + "tuner": "fmoe_ck", + "status": "failed", + "error_class": "codegen_unsupported_dtype", + "error": "Unsupported data type combination", + } + ], + } + ) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec): + return 1, sentinel, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) + + assert result["error_class"] == "codegen_unsupported_dtype" + assert "Unsupported data type" in str(result.get("error") or "") + + @pytest.mark.asyncio + async def test_a_candidate_keeps_both_the_env_and_a_sibling_crash( + self, tmp_path, monkeypatch + ): + """One tuner crashed, another delivered: forge reports ``candidate``, so + the env is measured and the crash is still named. Promotability keys on + ``status``, so a named crash must not demote the run.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + monkeypatch.setattr( + krh, "_persist_forge_gemm_csv_durably", lambda envs, **_kw: (dict(envs), "") + ) + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps( + { + "status": "ok", + "micro_decision": "candidate", + "recommended_env": {"AITER_CONFIG_FMOE": "/ws/tuned_fmoe.csv"}, + "tuners_run": [ + {"tuner": "a8w8", "status": "failed", "error_class": "codegen_crash"}, + {"tuner": "fmoe_ck", "status": "ok"}, + ], + } + ) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, sentinel, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) + + assert result["decision"] == "KEEP" + assert result["requires_e2e_validation"] is True + assert result["error_class"] == "codegen_crash" + # status decides promotability; a named crash must not demote the run. + assert result["status"] != "failed" + + @pytest.mark.asyncio + async def test_a_malformed_tuners_run_does_not_break_the_run( + self, tmp_path, monkeypatch + ): + """``tuners_run`` is forge's JSON and may be any shape; lifting a reason + out of it must not turn a run that happened into a reported crash.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + + for malformed in (5, "not-a-list", {"tuner": "fmoe_ck"}, [None, 7]): + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps( + { + "status": "ok", + "micro_decision": "no_improvement", + "tuners_run": malformed, + } + ) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec, _s=sentinel): + return 0, _s, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning( + {"task_id": f"malformed-{type(malformed).__name__}"}, + session_dir=tmp_path, + ) + + # The verdict still lands, and no exception class leaks in as a cause. + assert result["decision"] == "REVERT", malformed + assert result.get("error_class") != "TypeError", malformed + + @pytest.mark.asyncio + async def test_an_absent_micro_decision_is_left_alone(self, tmp_path, monkeypatch): + """No wording at all is not a verdict; the bridge must not invent one.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + sentinel = ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps({"status": "ok"}) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, sentinel, "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + + result = await krh._run_forge_gemm_tuning({}, session_dir=tmp_path) + + assert "decision" not in result + + # ---- MoE runtime key: log -> CSV -> payload -> forge argv --------------- + # Both ends were covered (the CSV writer, and KernelForge's preference for a + # caller-supplied CSV); the handoff between them was not, and deleting any + # link in it left the suite green. + + #: A real dispatch line, gfx field included. Fixtures that dropped the gfx + #: field once let a regex that could never match production pass its tests. + _REAL_MOE_DISPATCH = ( + "(Worker_TP0 pid=1) [aiter] [fused_moe] using 2stage default for " + "('gfx950', 256, 256, 4096, 512, 256, 6, 'ActivationType.Silu', " + "'torch.bfloat16', 'torch.float8_e4m3fn', 'torch.float4_e2m1fn_x2', " + "'QuantType.per_1x32', True, False)" + ) + + @staticmethod + def _moe_state(tmp_path): + model_dir = tmp_path / "moe-model" + model_dir.mkdir(exist_ok=True) + SharedState( + precision="fp8", + framework="sglang", + model_path=str(model_dir), + gpu_type="mi355x", + tp=1, + conc=64, + ).save(tmp_path) + return model_dir + + @staticmethod + def _sentinel() -> str: + return ( + "FORGE_GEMM_TUNE_RESULT_BEGIN\n" + + json.dumps({"status": "ok", "micro_decision": "skipped"}) + + "\nFORGE_GEMM_TUNE_RESULT_END\n" + ) + + @pytest.mark.asyncio + async def test_moe_key_travels_from_the_log_into_the_forge_payload( + self, tmp_path, monkeypatch + ): + """The values must come from the log, not from the config: a + config-derived key is what aiter would never look up.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + log = tmp_path / "server.log" + log.write_text(self._REAL_MOE_DISPATCH + "\n", encoding="utf-8") + + argv: dict[str, list[str]] = {} + + async def _fake_subprocess(cmd, *, timeout_sec): + argv["cmd"] = list(cmd) + return 0, self._sentinel(), "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + payload = {"task_id": "moe-key", "kernel_signature_log": str(log)} + + await krh._run_forge_gemm_tuning(payload, session_dir=tmp_path) + + workspace = krh._gemm_tuning_workspace(payload, session_dir=tmp_path) + written = json.loads( + (workspace / "forge_gemm_tuning_input.json").read_text(encoding="utf-8") + ) + csv_path = Path(written["moe_untuned_csv"]) + assert csv_path.is_file(), "the payload must name a CSV that exists" + + rows = csv_path.read_text(encoding="utf-8").strip().splitlines() + header = rows[0].split(",") + values = dict(zip(header, rows[1].split(","))) + # Straight off the log line, not inferred from the model config. + assert values["inter_dim"] == "512" + assert values["model_dim"] == "4096" + assert values["expert"] == "256" + assert values["topk"] == "6" + assert values["q_dtype_a"] == "torch.float8_e4m3fn" + assert values["q_dtype_w"] == "torch.float4_e2m1fn_x2" + assert values["q_type"] == "QuantType.per_1x32" + + def test_the_token_column_comes_from_the_workload(self, tmp_path): + """``tokens`` arrives as forge's comma-separated string, not a list. + + Every prior case passed a list, so the tests agreed with the annotation + instead of with the only production caller. + """ + log = tmp_path / "server.log" + log.write_text(self._REAL_MOE_DISPATCH + "\n", encoding="utf-8") + + for tokens, expected in ( + ("1,32,64", ["1", "32", "64"]), + ("64", ["64"]), + ([1, 32, 64], ["1", "32", "64"]), + ("", ["1"]), + (" 16 , 16 ,bad,-8, 0 ", ["16"]), + ): + csv_path, _report = krh._write_fmoe_untuned_csv_from_log( + str(log), tokens, tmp_path / f"ws_{str(tokens)[:12].strip()}" + ) + assert csv_path, f"no CSV for tokens={tokens!r}" + rows = Path(csv_path).read_text(encoding="utf-8").strip().splitlines() + got = [r.split(",")[0] for r in rows[1:]] + assert got == expected, f"tokens={tokens!r} -> {got}" + + @pytest.mark.asyncio + async def test_the_moe_csv_reaches_the_forge_argv(self, tmp_path, monkeypatch): + """Deriving the CSV is useless if the option never reaches forge.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + log = tmp_path / "server.log" + log.write_text(self._REAL_MOE_DISPATCH + "\n", encoding="utf-8") + + argv: dict[str, list[str]] = {} + + async def _fake_subprocess(cmd, *, timeout_sec): + argv["cmd"] = list(cmd) + return 0, self._sentinel(), "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + payload = {"task_id": "moe-argv", "kernel_signature_log": str(log)} + + await krh._run_forge_gemm_tuning(payload, session_dir=tmp_path) + + # The handler hands the tool an input JSON; the tool builds forge's argv + # from it. Assert the field the tool reads is the CSV that was derived. + workspace = krh._gemm_tuning_workspace(payload, session_dir=tmp_path) + written = json.loads( + (workspace / "forge_gemm_tuning_input.json").read_text(encoding="utf-8") + ) + assert written["moe_untuned_csv"].endswith("untuned_fmoe_from_runtime.csv") + assert str(workspace) in written["moe_untuned_csv"] + + @pytest.mark.asyncio + async def test_a_caller_supplied_moe_csv_wins(self, tmp_path, monkeypatch): + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + log = tmp_path / "server.log" + log.write_text(self._REAL_MOE_DISPATCH + "\n", encoding="utf-8") + supplied = tmp_path / "operator_moe.csv" + supplied.write_text(krh._FMOE_UNTUNED_CSV_HEADER + "\n", encoding="utf-8") + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, self._sentinel(), "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + payload = { + "task_id": "moe-supplied", + "kernel_signature_log": str(log), + "moe_untuned_csv": str(supplied), + } + + await krh._run_forge_gemm_tuning(payload, session_dir=tmp_path) + + workspace = krh._gemm_tuning_workspace(payload, session_dir=tmp_path) + written = json.loads( + (workspace / "forge_gemm_tuning_input.json").read_text(encoding="utf-8") + ) + assert written["moe_untuned_csv"] == str(supplied) + + @pytest.mark.asyncio + async def test_a_stale_moe_csv_path_falls_back_to_the_log( + self, tmp_path, monkeypatch + ): + """A path that no longer exists must not be forwarded to forge.""" + self._moe_state(tmp_path) + monkeypatch.setattr(krh, "_forge_gemm_tune_available", lambda: True) + log = tmp_path / "server.log" + log.write_text(self._REAL_MOE_DISPATCH + "\n", encoding="utf-8") + + async def _fake_subprocess(cmd, *, timeout_sec): + return 0, self._sentinel(), "" + + monkeypatch.setattr(krh, "_run_subprocess", _fake_subprocess) + payload = { + "task_id": "moe-stale", + "kernel_signature_log": str(log), + "moe_untuned_csv": str(tmp_path / "gone.csv"), + } + + await krh._run_forge_gemm_tuning(payload, session_dir=tmp_path) + + workspace = krh._gemm_tuning_workspace(payload, session_dir=tmp_path) + written = json.loads( + (workspace / "forge_gemm_tuning_input.json").read_text(encoding="utf-8") + ) + assert written["moe_untuned_csv"] != str(tmp_path / "gone.csv") + assert Path(written["moe_untuned_csv"]).is_file() + @pytest.mark.asyncio async def test_vllm_block_fp8_prefers_traced_shapes_over_profile_capture(self, tmp_path, monkeypatch): """vLLM block-FP8 must tune the device-side traced shapes. diff --git a/src/hyperloom/inference_optimizer/tests/test_model_path_resolver.py b/src/hyperloom/inference_optimizer/tests/test_model_path_resolver.py index b2900b7da1..50087aabc3 100644 --- a/src/hyperloom/inference_optimizer/tests/test_model_path_resolver.py +++ b/src/hyperloom/inference_optimizer/tests/test_model_path_resolver.py @@ -107,3 +107,25 @@ def test_load_model_config_dict_local_dir_unchanged(tmp_path): data = _load_model_config_dict(str(d)) assert isinstance(data, dict) assert data.get("model_type") == "mixtral" + + +def test_resolve_serving_model_path_prefers_hl_model_base(tmp_path, monkeypatch): + from hyperloom.common.model_paths import resolve_serving_model_path + + local = tmp_path / "DeepSeek-V4-Pro" + local.mkdir() + (local / "config.json").write_text("{}", encoding="utf-8") + monkeypatch.setenv("HL_MODEL_BASE", str(tmp_path)) + + resolved = resolve_serving_model_path("amd/DeepSeek-V4-Pro") + assert resolved == str(local) + + +def test_resolve_session_model_path_honors_params_then_env_then_state(monkeypatch): + from hyperloom.common.model_paths import resolve_session_model_path + + monkeypatch.setenv("MODEL_PATH", "/env/model") + assert resolve_session_model_path(params={"model_path": "/params/model"}) == "/params/model" + assert resolve_session_model_path(state_model_path="/state/model") == "/env/model" + monkeypatch.delenv("MODEL_PATH", raising=False) + assert resolve_session_model_path(state_model_path="/state/model") == "/state/model" diff --git a/src/hyperloom/inference_optimizer/tests/test_nogit_patch_unit.py b/src/hyperloom/inference_optimizer/tests/test_nogit_patch_unit.py index 85289f8099..644d8ba7e8 100644 --- a/src/hyperloom/inference_optimizer/tests/test_nogit_patch_unit.py +++ b/src/hyperloom/inference_optimizer/tests/test_nogit_patch_unit.py @@ -8,6 +8,7 @@ from __future__ import annotations +import shutil import subprocess from pathlib import Path @@ -985,3 +986,90 @@ def _raise_ctx(*a, **k): assert ok is False assert isinstance(feedback, af.ApplyFeedback) assert feedback.source_context == "" + + +# _sanitize_git_index_lines — placeholder git index headers + +ZERO_INDEX_MODIFY_DIFF = """\ +diff --git a/vllm/fp8.py b/vllm/fp8.py +index 0000000..1111111 100644 +--- a/vllm/fp8.py ++++ b/vllm/fp8.py +@@ -1,2 +1,3 @@ + # fp8 module + original = True ++patched = True +""" + +ZERO_INDEX_CREATE_DIFF = """\ +diff --git a/vllm/new.py b/vllm/new.py +new file mode 100644 +index 0000000..1111111 +--- /dev/null ++++ b/vllm/new.py +@@ -0,0 +1 @@ ++created = True +""" + + +def test_sanitize_drops_index_contradicting_modify_header(): + """An all-zero old blob on a modification hunk contradicts ``---`` → dropped.""" + out, dropped = ng._sanitize_git_index_lines(ZERO_INDEX_MODIFY_DIFF) + assert dropped == 1 + assert out == ZERO_INDEX_MODIFY_DIFF.replace("index 0000000..1111111 100644\n", "") + + +def test_sanitize_keeps_index_on_genuine_creation(): + """A creation hunk (``--- /dev/null``) legitimately has an all-zero old blob.""" + out, dropped = ng._sanitize_git_index_lines(ZERO_INDEX_CREATE_DIFF) + assert dropped == 0 + assert out == ZERO_INDEX_CREATE_DIFF + + +def test_sanitize_keeps_real_blob_hashes(): + """A plausible old blob hash is never touched.""" + text = ZERO_INDEX_MODIFY_DIFF.replace("0000000..1111111", "83db48f..bf269f4") + out, dropped = ng._sanitize_git_index_lines(text) + assert dropped == 0 + assert out == text + + +def test_sanitize_only_touches_the_contradicting_block(): + """In a multi-file patch the creation block keeps its index line.""" + out, dropped = ng._sanitize_git_index_lines(ZERO_INDEX_CREATE_DIFF + ZERO_INDEX_MODIFY_DIFF) + assert dropped == 1 + assert "index 0000000..1111111\n--- /dev/null" in out + assert "index 0000000..1111111 100644" not in out + + +def test_sanitize_no_op_returns_input_unchanged(): + """Nothing to drop → the original object is handed back.""" + out, dropped = ng._sanitize_git_index_lines(SIMPLE_DIFF) + assert dropped == 0 + assert out is SIMPLE_DIFF + + +@pytest.mark.skipif(shutil.which("patch") is None, reason="patch CLI unavailable") +def test_apply_no_git_tolerates_placeholder_index_header(tmp_path): + """A modification hunk carrying a placeholder all-zero index still applies. + + GNU ``patch`` reads the zero old blob as a creation and refuses the hunk + because the target already exists. Regression guard for the warm-replay + nogit cases, which only fail where the ``patch`` CLI is actually present. + """ + root = tmp_path / "tree" + target = root / "vllm" / "fp8.py" + target.parent.mkdir(parents=True) + original = "# fp8 module\noriginal = True\n" + target.write_text(original, encoding="utf-8") + patch_file = tmp_path / "patches" / "000_p.diff" + patch_file.parent.mkdir(parents=True) + patch_file.write_text(ZERO_INDEX_MODIFY_DIFF, encoding="utf-8") + + ok, err, backups, _feedback = ng._apply_patch_no_git(root, patch_file, tmp_path / "bak") + + assert ok is True, err + assert "patched = True" in target.read_text() + + ng._revert_patches_no_git(backups) + assert target.read_text() == original diff --git a/src/hyperloom/inference_optimizer/tests/test_warm_patch_apply.py b/src/hyperloom/inference_optimizer/tests/test_warm_patch_apply.py index b820de18c4..2c1b6ce33f 100644 --- a/src/hyperloom/inference_optimizer/tests/test_warm_patch_apply.py +++ b/src/hyperloom/inference_optimizer/tests/test_warm_patch_apply.py @@ -2,7 +2,9 @@ from __future__ import annotations +import shutil import subprocess +import sys from types import SimpleNamespace from unittest.mock import patch @@ -14,6 +16,7 @@ _create_patch_snapshot, _resolve_recipe_patch_target, _revert_patches, + _revert_warm_patch_state, ) @@ -23,6 +26,12 @@ def fake_repo(tmp_path): repo = tmp_path / "inferencex" repo.mkdir() subprocess.run(["git", "init"], cwd=str(repo), capture_output=True, check=True) + subprocess.run( + ["git", "config", "core.autocrlf", "false"], + cwd=str(repo), + capture_output=True, + check=True, + ) subprocess.run( ["git", "config", "user.email", "test@test.com"], cwd=str(repo), @@ -67,6 +76,11 @@ def output_dir(tmp_path): """ +def _require_patch_cli() -> None: + if not shutil.which("patch"): + pytest.skip("patch CLI unavailable") + + def test_apply_single_patch(fake_repo, output_dir): """Successfully apply a single patch.""" params = { @@ -120,6 +134,10 @@ def test_required_recipe_patch_fails_when_active_framework_root_is_missing( output_dir, monkeypatch, ): + monkeypatch.setattr( + "hyperloom.orchestrator.actions.executors.integrate_patch._resolve_framework_root", + lambda *_args, **_kwargs: None, + ) monkeypatch.setattr( "hyperloom.orchestrator.actions.executors.baseline.resolve_session_framework_root", lambda: "", @@ -387,6 +405,12 @@ def test_required_patch_uses_three_way_after_checks_fail( def _run(command, **_kwargs): calls.append(command) if "rev-parse" in command: + if "--is-inside-work-tree" in command: + return SimpleNamespace( + returncode=0, + stdout="true\n", + stderr="", + ) return SimpleNamespace( returncode=0, stdout=b"0123456789abcdef\n", @@ -540,12 +564,18 @@ def test_snapshot_revert_rejects_head_mismatch( assert result["errors"][0].startswith("head_mismatch:") -def test_required_patch_refuses_repo_without_head(tmp_path, output_dir): +def test_required_timeline_refuses_a_repo_with_no_head(tmp_path, output_dir): + """prelude promotes this tree against a pre_sha it cannot get here. + + Applying via nogit made the run look prepared and then fail downstream with + validated_recipe_checkout_incomplete, leaving a half-patched tree behind. + Refusing up front is the outcome the caller can act on. + """ repo = tmp_path / "unborn" repo.mkdir() subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True) target = repo / "vllm" / "fp8.py" - target.parent.mkdir() + target.parent.mkdir(parents=True) target.write_text("# fp8 module\noriginal = True\n") result = _apply_warm_patches( @@ -559,6 +589,90 @@ def test_required_patch_refuses_repo_without_head(tmp_path, output_dir): assert result["status"] == "failed" assert result["failure"] == "missing_git_head" + assert "original = True" in target.read_text(), "must not leave a patched tree" + + +def test_required_timeline_refuses_a_non_git_install_tree(tmp_path, output_dir): + """Same contract for an install tree that was never a repo.""" + install_root = tmp_path / "dist-packages" + target = install_root / "vllm" / "fp8.py" + target.parent.mkdir(parents=True) + target.write_text("# fp8 module\noriginal = True\n") + + result = _apply_warm_patches( + { + "patches": [{"patch_file": "vllm/fp8.py", "patch_content": VALID_PATCH}], + "required_patch_timeline": True, + }, + str(install_root), + output_dir, + ) + + assert result["status"] == "failed" + assert result["failure"] == "missing_git_head" + assert "original = True" in target.read_text() + + +def test_nogit_still_serves_the_legacy_list(tmp_path, output_dir): + """Nothing downstream of a legacy patch needs a sha, so nogit stays.""" + _require_patch_cli() + install_root = tmp_path / "dist-packages" + target = install_root / "vllm" / "fp8.py" + target.parent.mkdir(parents=True) + target.write_text("# fp8 module\noriginal = True\n") + + applied = _apply_warm_patches( + {"patches": [{"patch_file": "vllm/fp8.py", "patch_content": VALID_PATCH}]}, + str(install_root), + output_dir, + ) + + assert [p["status"] for p in applied] == ["applied_nogit"] + assert "patched = True" in target.read_text() + assert (output_dir / "warm_patches" / "patch_backups").is_dir() + + +def test_nogit_apply_hands_teardown_the_backups_it_needs(tmp_path, output_dir): + """A nogit apply has no sha, so its backups are the only way back.""" + _require_patch_cli() + install_root = tmp_path / "dist-packages" + target = install_root / "vllm" / "fp8.py" + target.parent.mkdir(parents=True) + target.write_text("# fp8 module\noriginal = True\n") + params = {"patches": [{"patch_file": "vllm/fp8.py", "patch_content": VALID_PATCH}]} + + applied = _apply_warm_patches(params, str(install_root), output_dir) + + assert [p["status"] for p in applied] == ["applied_nogit"] + assert not params.get("_warm_patch_snapshot_manifest"), "nogit has no git snapshot" + assert params["_warm_patch_nogit_backups"], "teardown would have nothing to undo" + + +def test_teardown_undoes_a_nogit_apply(tmp_path): + """Keying the revert on pre_sha alone leaked nogit patches into later tasks + that reuse the same checkout.""" + target = tmp_path / "vllm" / "fp8.py" + target.parent.mkdir(parents=True) + target.write_text("# fp8 module\noriginal = True\n") + backup = tmp_path / "backups" / "p__vllm__fp8.py__0000.bak" + backup.parent.mkdir(parents=True) + shutil.copy2(target, backup) + target.write_text("# fp8 module\noriginal = True\npatched = True\n") + + result = _revert_warm_patch_state( + str(tmp_path), + pre_sha="", + nogit_backups=[ + { + "target": str(target), + "existed": True, + "backup_path": str(backup), + "revert_action": "restore", + } + ], + ) + + assert result == {"ok": True, "errors": [], "channel": "nogit"} assert "patched = True" not in target.read_text() @@ -733,10 +847,20 @@ def test_rollback_does_not_erase_already_present_patch( assert target.read_text() == "# fp8 module\noriginal = True\npatched = True\n" +@pytest.mark.skipif( + sys.platform == "win32", + reason="git apply --3way merge baseline is validated on Linux CI/pod", +) def test_real_git_three_way_merge_succeeds(tmp_path, output_dir): repo = tmp_path / "threeway" repo.mkdir() subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True) + subprocess.run( + ["git", "config", "core.autocrlf", "false"], + cwd=repo, + check=True, + capture_output=True, + ) subprocess.run( ["git", "config", "user.email", "test@test.com"], cwd=repo, diff --git a/src/hyperloom/orchestrator/actions/executors/_nogit_patch.py b/src/hyperloom/orchestrator/actions/executors/_nogit_patch.py index 584008b4f9..067219f486 100644 --- a/src/hyperloom/orchestrator/actions/executors/_nogit_patch.py +++ b/src/hyperloom/orchestrator/actions/executors/_nogit_patch.py @@ -18,6 +18,17 @@ * :data:`_PATCH_DEV_NULL` — the sentinel ``/dev/null`` path in diff headers. * :func:`_strip_path_prefix` — drop leading path components like ``git apply -p``. * :func:`_is_within` — containment check (both paths pre-resolved). +* :func:`_sanitize_git_index_lines` — drop ``index`` headers that contradict ``---``. + +Placeholder git index headers +----------------------------- +Unlike ``git apply``, GNU ``patch`` honours the ``index ..`` line and +reads an all-zero *old* blob hash as a file creation. Specialists write +placeholder hashes, so a modification hunk can arrive as +``index 0000000..1111111`` alongside ``--- a/path``, and ``patch`` then refuses +it with ``... which already exists!``. :func:`_sanitize_git_index_lines` drops +such contradicting lines before the CLI sees the patch; genuine creations +(``--- /dev/null``) keep theirs. Backup naming ------------- @@ -71,6 +82,73 @@ # Characters unsafe in filenames (replaced with ``_`` in rel_flat). _UNSAFE_NAME_RE = re.compile(r"[/\\:<>\"?*|]") +# A git ``index ..`` header whose *old* blob hash is all zeros. +_ZERO_OLD_INDEX_RE = re.compile(r"^index 0+\.\.") + + +def _old_path_after_index(lines: list[str], start: int) -> str | None: + """Return the ``--- `` path token of the file block containing ``lines[start]``. + + Scans forward from an ``index`` line to that block's ``--- `` header, + stopping at the next ``diff --git`` header or the first hunk marker so a + later block's header is never attributed to this one. + + Args: + lines: The patch text split into lines. + start: Index of the ``index`` line to resolve. + + Returns: + The raw pre-image path token, or ``None`` when the block has no + ``--- `` header. + """ + for line in lines[start + 1 :]: + if line.startswith("--- "): + return line[4:].strip().split("\t")[0] + if line.startswith("diff --git ") or line.startswith("@@"): + return None + return None + + +def _sanitize_git_index_lines(patch_text: str) -> tuple[str, int]: + """Drop git ``index`` lines whose all-zero old blob contradicts the ``---`` header. + + GNU ``patch`` reads an all-zero *old* blob hash as "this hunk creates the + file" and then refuses the hunk with ``The next patch would create the file + X, which already exists!`` -- even though the accompanying ``--- a/X`` + header says X is being *modified*. ``git apply`` ignores the index line + entirely, so such a patch applies through the git channel and fails only + here, which makes the failure look like a bad patch rather than a header + disagreement. + + Specialists emit placeholder index lines rather than real blob hashes, so + the contradiction is common enough to absorb rather than reject. The + ``---``/``+++`` headers are the authoritative unified-diff surface and GNU + ``patch`` does not need the index line, so a contradicting one is dropped. + + A genuine creation hunk carries ``--- /dev/null`` and keeps its index line, + so real file creations are unaffected. + + Args: + patch_text: The unified-diff text to sanitize. + + Returns: + A ``(sanitized_text, dropped_count)`` pair. When nothing contradicts, + ``dropped_count`` is ``0`` and the text is returned unmodified. + """ + lines = patch_text.splitlines(keepends=True) + kept: list[str] = [] + dropped = 0 + for idx, line in enumerate(lines): + if _ZERO_OLD_INDEX_RE.match(line): + old_path = _old_path_after_index(lines, idx) + if old_path is not None and old_path != _PATCH_DEV_NULL: + dropped += 1 + continue + kept.append(line) + if not dropped: + return patch_text, 0 + return "".join(kept), dropped + def _strip_path_prefix(path: str, level: int) -> str: """Drop ``level`` leading path components (mimics ``git apply -p``). @@ -219,6 +297,31 @@ def _apply_patch_no_git( """ from ._apply_feedback import ApplyFeedback, read_patch_source_context + try: + patch_text = patch_path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + err_msg = f"cannot read patch file: {exc}" + return ( + False, + err_msg, + [], + ApplyFeedback(patch=str(patch_path), channel="nogit", tried_levels=[], stderr=err_msg), + ) + + # Feed the CLI a copy with contradicting index headers removed; keep the + # original path in feedback so advisories point at what the author wrote. + patch_input = patch_path + sanitized_text, dropped_index_lines = _sanitize_git_index_lines(patch_text) + if dropped_index_lines: + backup_root.mkdir(parents=True, exist_ok=True) + patch_input = backup_root / f"{patch_path.stem}.sanitized.diff" + patch_input.write_text(sanitized_text, encoding="utf-8") + log.info( + "nogit patch: dropped %d placeholder git index line(s) from %s that contradicted the --- header", + dropped_index_lines, + patch_path.name, + ) + # Detect strip level via dry-run; accumulate stderr per level for feedback. detected_level: int | None = None dry_run_stderrs: list[str] = [] @@ -227,7 +330,7 @@ def _apply_patch_no_git( tried_levels.append(lvl) try: cp = subprocess.run( - ["patch", f"-p{lvl}", "--dry-run", "-i", str(patch_path)], + ["patch", f"-p{lvl}", "--dry-run", "-i", str(patch_input)], capture_output=True, text=True, timeout=60, @@ -268,7 +371,7 @@ def _apply_patch_no_git( # case the apply is a satisfied no-op -- report success with no backups # (the patch that really made those edits owns the backups needed for a # correct revert). - if _reverse_applies_cleanly(framework_root, patch_path): + if _reverse_applies_cleanly(framework_root, patch_input): log.info( "nogit patch: %s is already fully applied (clean reverse dry-run); treating as a no-op", patch_path.name, @@ -277,7 +380,6 @@ def _apply_patch_no_git( combined_stderr = "\n".join(dry_run_stderrs) err_msg = f"patch --dry-run failed at all strip levels for {patch_path.name}" try: - patch_text = patch_path.read_text(encoding="utf-8", errors="replace") source_ctx = read_patch_source_context(patch_text, framework_root, radius=50) except Exception: # noqa: BLE001 source_ctx = "" @@ -304,12 +406,6 @@ def _fail(err_message: str, recs: list[dict[str, Any]]) -> "tuple[bool, str, lis ), ) - # Resolve target files to back up before mutation. - try: - patch_text = patch_path.read_text(encoding="utf-8", errors="replace") - except OSError as exc: - return _fail(f"cannot read patch file: {exc}", []) - framework_root_resolved = framework_root.resolve() backup_root.mkdir(parents=True, exist_ok=True) backups: list[dict[str, Any]] = [] @@ -440,7 +536,7 @@ def _backup_existing(abs_path: Path, rel: Path, action: str) -> tuple[dict[str, rej_dir.mkdir(parents=True, exist_ok=True) try: cp2 = subprocess.run( - ["patch", f"-p{detected_level}", "--reject-file=-", "-i", str(patch_path)], + ["patch", f"-p{detected_level}", "--reject-file=-", "-i", str(patch_input)], capture_output=True, text=True, timeout=120, @@ -463,7 +559,6 @@ def _backup_existing(abs_path: Path, rel: Path, action: str) -> tuple[dict[str, apply_stderr = cp2.stderr.strip() or cp2.stdout.strip() source_ctx = "" try: - patch_text = patch_path.read_text(encoding="utf-8", errors="replace") source_ctx = read_patch_source_context(patch_text, framework_root, radius=50) except Exception: # noqa: BLE001 pass @@ -558,5 +653,6 @@ def _revert_patches_no_git(backups: list[dict[str, Any]]) -> None: "_is_git_tree", "_is_within", "_revert_patches_no_git", + "_sanitize_git_index_lines", "_strip_path_prefix", ] diff --git a/src/hyperloom/orchestrator/actions/executors/baseline.py b/src/hyperloom/orchestrator/actions/executors/baseline.py index 2a297dac8b..09f0659477 100644 --- a/src/hyperloom/orchestrator/actions/executors/baseline.py +++ b/src/hyperloom/orchestrator/actions/executors/baseline.py @@ -32,6 +32,7 @@ from hyperloom.common.env import is_truthy from hyperloom.common.env_safety import redact_secret_values, scrub_benchmark_process_env from hyperloom.common.git_safety import safe_directory_args +from hyperloom.common.model_paths import resolve_session_model_path from hyperloom.inference_optimizer.session.session_paths import runs_dir from ...framework.paths import resolve_session_framework_root from ...loop.sub_agent_runner import RunnerContext @@ -1338,13 +1339,65 @@ def _verify_three_way_clean( return True, "" +def _patch_paths_from_warm_params(params: dict[str, Any]) -> list[Path]: + """Collect diff-header targets from warm-replay patch payloads.""" + from ...specialists.patch_safety import patch_file_targets + + patch_paths: list[Path] = [] + seen: set[str] = set() + for patch in params.get("patches") or []: + if not isinstance(patch, dict): + continue + content = str(patch.get("patch_content") or "") + patch_ref = str(patch.get("patch_ref") or "") + if not content and patch_ref: + try: + content = Path(patch_ref).read_text(encoding="utf-8", errors="replace") + except OSError: + content = "" + if not content: + continue + for old_raw, new_raw in patch_file_targets(content): + raw = new_raw if new_raw and new_raw not in {"/dev/null", ""} else old_raw + if not raw or raw == "/dev/null" or raw in seen: + continue + seen.add(raw) + patch_paths.append(Path(raw)) + return patch_paths + + def _resolve_recipe_patch_target(params: dict[str, Any]) -> str: - """Return the active framework root for Explore/Framework Recipe patches.""" + """Return the framework root whose tree holds the warm-replay patch targets.""" if not params.get("patches"): return "" + from .integrate_patch import _resolve_framework_root + + patch_paths = _patch_paths_from_warm_params(params) + root = _resolve_framework_root(None, patch_paths=patch_paths or None) + if root is not None: + return str(root) return resolve_session_framework_root() +def _revert_warm_patch_state( + target_repo: str, + *, + pre_sha: str = "", + snapshot_manifest: Any = None, + nogit_backups: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Restore warm-replay patch mutations via git snapshot or nogit backups.""" + if nogit_backups: + from ._nogit_patch import _revert_patches_no_git + + try: + _revert_patches_no_git(list(nogit_backups)) + except Exception as exc: # noqa: BLE001 + return {"ok": False, "errors": [repr(exc)], "channel": "nogit"} + return {"ok": True, "errors": [], "channel": "nogit"} + return _revert_patches(target_repo, pre_sha, snapshot_manifest) + + def _apply_warm_patches( params: dict[str, Any], target_repo: str, @@ -1356,7 +1409,8 @@ def _apply_warm_patches( Reads ``params["patches"]`` (list of dicts with patch_file/patch_content/ patch_ref) and ``params["blocked_patches"]`` (blocklist). Applies each patch - via ``git apply`` in the target repo, skipping blocklisted patches. + via ``git apply`` when the target is a git work-tree, otherwise via the + shared nogit ``patch`` CLI path used by integrate_patch. Legacy patch lists return the list of successfully applied patch metadata dicts (best-effort skip semantics). Current-contract timelines set @@ -1387,7 +1441,16 @@ def _apply_warm_patches( statuses: list[dict[str, Any]] = [] patch_log_dir = output_dir / "warm_patches" patch_log_dir.mkdir(parents=True, exist_ok=True) - pre_sha = _git_head_sha(target_repo) + from ._nogit_patch import _apply_patch_no_git, _is_git_tree + + target_path = Path(target_repo) + git_tree = _is_git_tree(target_path) + pre_sha = _git_head_sha(target_repo) if git_tree else "" + # prelude promotes a required timeline's tree only against a pre_sha and a + # git snapshot manifest. nogit produces neither, so serving this path from it + # turned a successful replay into validated_recipe_checkout_incomplete -- + # worse than the fast failure it replaced. Refuse up front, as before; nogit + # serves the legacy list, where nothing downstream needs a sha. if required_timeline and not pre_sha: return { "required": True, @@ -1400,6 +1463,8 @@ def _apply_warm_patches( "target_repo": target_repo, "rolled_back": False, } + use_nogit = not git_tree or not pre_sha + nogit_backups: list[dict[str, Any]] = [] from ...specialists.patch_safety import is_unified_diff, patch_escapes_tree resolved_contents: dict[int, str] = {} @@ -1460,7 +1525,7 @@ def _apply_warm_patches( resolved_contents[idx] = content snapshot_contents.append(content) snapshot_manifest: dict[str, Any] | None = None - if snapshot_contents: + if snapshot_contents and not use_nogit: try: snapshot_manifest = _create_patch_snapshot( target_repo, @@ -1606,85 +1671,99 @@ def _apply_warm_patches( method = "" try: - checked = subprocess.run( - ["git", "apply", "--check", str(patch_path)], - cwd=target_repo, - capture_output=True, - timeout=30, - check=False, - ) - if checked.returncode == 0: - subprocess.run( - ["git", "apply", str(patch_path)], - cwd=target_repo, - capture_output=True, - timeout=30, - check=True, + if use_nogit: + backup_root = patch_log_dir / "patch_backups" + ok, err, backups, _feedback = _apply_patch_no_git( + target_path, + patch_path, + backup_root, + seq_offset=len(nogit_backups), ) - method = "applied" - elif required_timeline: - reverse = subprocess.run( - ["git", "apply", "-R", "--check", str(patch_path)], + if not ok: + raise RuntimeError(err or "nogit patch apply failed") + nogit_backups.extend(backups) + method = "applied_nogit" + else: + checked = subprocess.run( + ["git", "apply", "--check", str(patch_path)], cwd=target_repo, capture_output=True, timeout=30, check=False, ) - if reverse.returncode == 0: - method = ( - "already_present" - if _patch_present_in_committed_head( - target_repo, - patch_path, - ) - else "present_in_dirty_worktree" - ) - else: - touched = _patch_touched_paths(patch_content) - before_residue = _three_way_residue_snapshot( - target_repo, - touched, + if checked.returncode == 0: + subprocess.run( + ["git", "apply", str(patch_path)], + cwd=target_repo, + capture_output=True, + timeout=30, + check=True, ) - three_way = subprocess.run( - ["git", "apply", "--3way", str(patch_path)], + method = "applied" + elif required_timeline: + reverse = subprocess.run( + ["git", "apply", "-R", "--check", str(patch_path)], cwd=target_repo, capture_output=True, timeout=30, check=False, ) - if three_way.returncode == 0: - clean, residue = _verify_three_way_clean( + if reverse.returncode == 0: + method = ( + "already_present" + if _patch_present_in_committed_head( + target_repo, + patch_path, + ) + else "present_in_dirty_worktree" + ) + else: + touched = _patch_touched_paths(patch_content) + before_residue = _three_way_residue_snapshot( target_repo, touched, - before_residue, ) - if not clean: - raise RuntimeError(residue) - method = "applied_3way" - else: - detail = ( - three_way.stderr.decode(errors="replace")[:500] - if three_way.stderr - else "git apply --3way failed" + three_way = subprocess.run( + ["git", "apply", "--3way", str(patch_path)], + cwd=target_repo, + capture_output=True, + timeout=30, + check=False, ) - raise RuntimeError(detail) - else: - detail = ( - checked.stderr.decode(errors="replace")[:500] - if checked.stderr - else "git apply --check failed" - ) - raise RuntimeError(detail) + if three_way.returncode == 0: + clean, residue = _verify_three_way_clean( + target_repo, + touched, + before_residue, + ) + if not clean: + raise RuntimeError(residue) + method = "applied_3way" + else: + detail = ( + three_way.stderr.decode(errors="replace")[:500] + if three_way.stderr + else "git apply --3way failed" + ) + raise RuntimeError(detail) + else: + detail = ( + checked.stderr.decode(errors="replace")[:500] + if checked.stderr + else "git apply --check failed" + ) + raise RuntimeError(detail) except (subprocess.CalledProcessError, subprocess.TimeoutExpired, OSError, RuntimeError) as exc: log.warning( - "baseline_executor: git apply failed for patch %s: %s", + "baseline_executor: warm patch apply failed for %s: %s", patch_file, exc, ) - status.update(status="failed", reason="git_apply_failed", detail=str(exc)[:500]) + reason = "nogit_apply_failed" if use_nogit else "git_apply_failed" + status.update(status="failed", reason=reason, detail=str(exc)[:500]) statuses.append(status) if required_timeline: - failed_ref, failure = patch_file, "git_apply_failed" + failed_ref, failure = patch_file, reason break continue @@ -1697,12 +1776,16 @@ def _apply_warm_patches( status["status"] = method statuses.append(status) + if nogit_backups: + params["_warm_patch_nogit_backups"] = nogit_backups + if required_timeline: if failed_ref: - restore = _revert_patches( + restore = _revert_warm_patch_state( target_repo, - pre_sha, - snapshot_manifest, + pre_sha=pre_sha, + snapshot_manifest=snapshot_manifest, + nogit_backups=nogit_backups, ) return { "required": True, @@ -3041,14 +3124,13 @@ async def _run_once( _pre_patch_sha = "" timeout_sec = self._resolve_timeout(params) - # Model path: task.params['model_path'] > $MODEL_PATH > SharedState; - # if none, leave the YAML's hardcoded `model:` for fixture-based tests. - # Live state is read from ctx.extra (the executor is a module-level - # singleton with self.shared_state=None on the Coordinator path). - resolved_model = ( - str(params.get("model_path") or "").strip() - or os.environ.get("MODEL_PATH", "").strip() - or str(getattr(live_shared_state, "model_path", "") or "").strip() + # Model path: unified resolver (params → $MODEL_PATH → SharedState), then + # serving-path normalization (HL_MODEL_BASE / HF cache). If none, leave + # the YAML's hardcoded `model:` for fixture-based tests. + resolved_model = resolve_session_model_path( + params=params, + state_model_path=str(getattr(live_shared_state, "model_path", "") or ""), + for_serving=True, ) # gpu_type: task.params > $GPU_TYPE (cli.py canonicalizes mi325x->mi300x). resolved_gpu = ( @@ -3413,13 +3495,21 @@ def _persist_recipe_snapshot(manifest: dict[str, Any]) -> bool: ) return result finally: - if applied_patches and _pre_patch_sha and not isinstance( - patch_application, dict + # A required timeline's tree is promoted by prelude after this + # returns, so it must stay patched; reverting here handed prelude + # a clean tree and silently lost the replay. + if ( + applied_patches + and not isinstance(patch_application, dict) + and (_pre_patch_sha or params.get("_warm_patch_nogit_backups")) ): - _revert_patches( + _revert_warm_patch_state( patch_target, - _pre_patch_sha, - params.get("_warm_patch_snapshot_manifest"), + pre_sha=_pre_patch_sha, + snapshot_manifest=params.get("_warm_patch_snapshot_manifest"), + nogit_backups=list( + params.get("_warm_patch_nogit_backups") or [] + ), ) if bench_lease is not None: bench_lease.close() @@ -3742,16 +3832,22 @@ def _persist_recipe_snapshot(manifest: dict[str, Any]) -> bool: port=port, ) # Revert warm-replay patches to prevent state leakage into - # subsequent tasks that reuse the same InferenceX checkout. + # subsequent tasks that reuse the same InferenceX checkout. A + # required timeline is exempt: prelude promotes that tree after this + # returns and needs it still patched. if ( applied_patches - and _pre_patch_sha and not isinstance(patch_application, dict) + and ( + _pre_patch_sha + or params.get("_warm_patch_nogit_backups") + ) ): - _revert_patches( + _revert_warm_patch_state( patch_target, - _pre_patch_sha, - params.get("_warm_patch_snapshot_manifest"), + pre_sha=_pre_patch_sha, + snapshot_manifest=params.get("_warm_patch_snapshot_manifest"), + nogit_backups=list(params.get("_warm_patch_nogit_backups") or []), ) if bench_lease is not None: bench_lease.close() diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py index 80e090611c..2696b9f484 100644 --- a/src/hyperloom/orchestrator/actions/executors/explore.py +++ b/src/hyperloom/orchestrator/actions/executors/explore.py @@ -38,6 +38,7 @@ from hyperloom.common.coerce import to_str_list from hyperloom.common.gain_math import gain_pct +from hyperloom.common.model_paths import resolve_session_model_path from hyperloom.common.timeutil import now_iso from hyperloom.inference_optimizer.session.session_paths import runs_dir from ...state.failure_evidence import ( @@ -713,6 +714,7 @@ async def __call__(self, ctx) -> dict[str, Any]: "error": f"config not found: {config_path}", } extra = getattr(ctx, "extra", None) or {} + shared_state = extra.get("shared_state") or extra.get("state") output_root = Path( params.get("output_dir") or extra.get("workspace") @@ -723,7 +725,11 @@ async def __call__(self, ctx) -> dict[str, Any]: # ----- Workload-contract materialization --------------------------- # Re-materialize so variant YAMLs honour the operator's actual # workload (CONC / ISL / OSL / TP / MAX_MODEL_LEN / PRECISION). - resolved_model = str(params.get("model_path") or "").strip() or os.environ.get("MODEL_PATH", "").strip() + resolved_model = resolve_session_model_path( + params=params, + state_model_path=str(getattr(shared_state, "model_path", "") or "") if shared_state else "", + for_serving=True, + ) resolved_gpu = ( str(params.get("gpu_type") or "").strip().lower() or os.environ.get("GPU_TYPE", "").strip().lower() ) diff --git a/src/hyperloom/orchestrator/actions/executors/framework_agent.py b/src/hyperloom/orchestrator/actions/executors/framework_agent.py index 44d2c556cb..63cf71e1b2 100644 --- a/src/hyperloom/orchestrator/actions/executors/framework_agent.py +++ b/src/hyperloom/orchestrator/actions/executors/framework_agent.py @@ -12,6 +12,7 @@ from typing import Any from hyperloom.common.env import is_truthy +from hyperloom.common.model_paths import resolve_session_model_path from hyperloom.inference_optimizer.session.session_paths import runs_dir from ._accuracy_gate import ( accuracy_keep_block, @@ -834,6 +835,9 @@ def _undo_candidate() -> None: slug=slug, session_deadline_sec=session_deadline_sec, variant_expected_sec=variant_expected_sec, + state_model_path=str( + getattr(extra.get("shared_state") or extra.get("state"), "model_path", "") or "" + ), ) except FrameworkScriptMismatchError as exc: reverted = self._revert_patches( @@ -1181,6 +1185,7 @@ async def _bench_candidate( slug: str, session_deadline_sec: float | None = None, variant_expected_sec: float | None = None, + state_model_path: str = "", ) -> tuple[dict[str, Any], dict[str, Any]]: """Run a 1-variant Magpie bench under the patched server + accuracy gate. Mirrors :meth:`IntegratePatchExecutor._bench_patch`. @@ -1194,6 +1199,9 @@ async def _bench_candidate( session context. variant_expected_sec: Expected bench runtime used to decide whether the remaining budget can fit this bench at all. + state_model_path: ``SharedState.model_path``, the last fallback in + the model-path precedence. Passed in because the caller owns the + session context. Returns: A ``(bench, gate_evidence)`` tuple: the bench result dict and a @@ -1202,7 +1210,16 @@ async def _bench_candidate( config_path = Path(params.get("config_path") or self.default_config_path or default_baseline_config()) if not config_path.exists(): raise RuntimeError(f"framework bench: config not found at {config_path}") - resolved_model = str(params.get("model_path") or "").strip() or os.environ.get("MODEL_PATH", "").strip() + # This bench launches a server, so the value has to be a servable path: + # the shared resolver walks HL_MODEL_BASE and the hub cache, where the + # local two-step did not, and handed a bare repo id straight to the + # server. It falls back to the original string, so an unresolvable + # value degrades exactly as before rather than emptying. + resolved_model = resolve_session_model_path( + params=params, + state_model_path=state_model_path, + for_serving=True, + ) resolved_gpu = ( str(params.get("gpu_type") or "").strip().lower() or os.environ.get("GPU_TYPE", "").strip().lower() ) diff --git a/src/hyperloom/orchestrator/actions/executors/integrate_patch.py b/src/hyperloom/orchestrator/actions/executors/integrate_patch.py index d5428f5b55..015ee142b8 100644 --- a/src/hyperloom/orchestrator/actions/executors/integrate_patch.py +++ b/src/hyperloom/orchestrator/actions/executors/integrate_patch.py @@ -19,6 +19,7 @@ from typing import Any from hyperloom.common.coerce import to_str_list +from hyperloom.common.model_paths import resolve_session_model_path from hyperloom.common.timeutil import now_iso from hyperloom.inference_optimizer.gpu_types import amd_gpu_dispatch_identity from hyperloom.inference_optimizer.session.session_paths import runs_dir @@ -4053,7 +4054,7 @@ async def _bench_patch( config_path = Path(params.get("config_path") or self.default_config_path or default_baseline_config()) if not config_path.exists(): raise RuntimeError(f"integrate_patch bench: config not found at {config_path}") - resolved_model = str(params.get("model_path") or "").strip() or os.environ.get("MODEL_PATH", "").strip() + resolved_model = resolve_session_model_path(params=params, for_serving=True) resolved_gpu = ( str(params.get("gpu_type") or "").strip().lower() or os.environ.get("GPU_TYPE", "").strip().lower() ) @@ -4319,7 +4320,7 @@ async def _confirm_stack_rebench( run it is confirming for. """ config_path = Path(params.get("config_path") or self.default_config_path or default_baseline_config()) - resolved_model = str(params.get("model_path") or "").strip() or os.environ.get("MODEL_PATH", "").strip() + resolved_model = resolve_session_model_path(params=params, for_serving=True) resolved_gpu = ( str(params.get("gpu_type") or "").strip().lower() or os.environ.get("GPU_TYPE", "").strip().lower() ) diff --git a/src/hyperloom/orchestrator/actions/executors/sweep.py b/src/hyperloom/orchestrator/actions/executors/sweep.py index 2d2bf15c17..b1633a3ab8 100644 --- a/src/hyperloom/orchestrator/actions/executors/sweep.py +++ b/src/hyperloom/orchestrator/actions/executors/sweep.py @@ -32,6 +32,7 @@ from typing import Any from hyperloom.common.coerce import to_int +from hyperloom.common.model_paths import resolve_session_model_path from hyperloom.inference_optimizer.session.session_paths import runs_dir from ._grid_base import pareto_front from ._grid_runner import ( @@ -241,6 +242,7 @@ async def __call__(self, ctx) -> dict[str, Any]: if not config_path.exists(): return {"status": "failed", "error_class": "missing_config", "error": f"config not found: {config_path}"} extra = getattr(ctx, "extra", None) or {} + shared_state = extra.get("shared_state") or extra.get("state") output_root = Path( params.get("output_dir") or extra.get("workspace") or runs_dir(self.session_dir, "sweep", ctx.task.task_id) ) @@ -249,7 +251,11 @@ async def __call__(self, ctx) -> dict[str, Any]: # Workload-contract materialization: sweep overrides CONC/ISL/OSL/ # NUM_PROMPTS per variant, but TP/MAX_MODEL_LEN/PRECISION/RUN_EVAL/ # ROCR_VISIBLE_DEVICES still flow from env onto the variant base. - resolved_model = str(params.get("model_path") or "").strip() or os.environ.get("MODEL_PATH", "").strip() + resolved_model = resolve_session_model_path( + params=params, + state_model_path=str(getattr(shared_state, "model_path", "") or "") if shared_state else "", + for_serving=True, + ) resolved_gpu = ( str(params.get("gpu_type") or "").strip().lower() or os.environ.get("GPU_TYPE", "").strip().lower() ) diff --git a/src/hyperloom/orchestrator/kernel/conc_sweep.py b/src/hyperloom/orchestrator/kernel/conc_sweep.py index 20cc110de3..5e57302bcb 100644 --- a/src/hyperloom/orchestrator/kernel/conc_sweep.py +++ b/src/hyperloom/orchestrator/kernel/conc_sweep.py @@ -21,6 +21,7 @@ from hyperloom.common import io as _common_io from hyperloom.common.gain_math import conc_pair_comparison +from hyperloom.common.model_paths import resolve_session_model_path from hyperloom.common.timeutil import utc_now_compact from hyperloom.inference_optimizer.session.session_paths import reports_dir, runs_root from ..actions.executors._grid_runner import ( @@ -1210,7 +1211,10 @@ async def run_conc_sweep( workspace.mkdir(parents=True, exist_ok=True) # Re-materialize (idempotent) in case we fell back to the shipped asset. - resolved_model = str(getattr(state, "model_path", "") or "").strip() or os.environ.get("MODEL_PATH", "").strip() + resolved_model = resolve_session_model_path( + state_model_path=str(getattr(state, "model_path", "") or ""), + for_serving=True, + ) # Mirror the main flow (baseline/sweep/...): prefer $GPU_TYPE (cli.py # canonicalizes mi325x/mi308x -> mi300x), fall back to state.gpu_type, then # canonicalize through _gpu_runner_type so the selected Magpie script is a diff --git a/src/hyperloom/orchestrator/kernel/gemm_shape_coverage.py b/src/hyperloom/orchestrator/kernel/gemm_shape_coverage.py index eecb30ca63..078fd94c1e 100644 --- a/src/hyperloom/orchestrator/kernel/gemm_shape_coverage.py +++ b/src/hyperloom/orchestrator/kernel/gemm_shape_coverage.py @@ -79,7 +79,6 @@ r"kernelName1='(?P[^']*)'.*?kernelName2='(?P[^']*)'" ) - def aiter_padded_m_fine(m: int) -> int: """Return aiter's ``gl=0`` padded M (fine-grained lookup key).""" if m <= 256: diff --git a/src/hyperloom/orchestrator/kernel/request_handlers.py b/src/hyperloom/orchestrator/kernel/request_handlers.py index 9610beb3ea..69f41368b2 100644 --- a/src/hyperloom/orchestrator/kernel/request_handlers.py +++ b/src/hyperloom/orchestrator/kernel/request_handlers.py @@ -2805,17 +2805,99 @@ def _is_vllm_block_fp8(precision: str, quant_type: str) -> bool: "fused_moe": ("[aiter] [fused_moe]", "Mxfp4 MoE backend"), } -#: aiter logs the fused-MoE problem it dispatched as -#: ``[fused_moe] using ... (cu, token, dim, inter, experts, topk, act, dtype, -#: q_dtype_a, q_dtype_w, q_type, ...)``. +#: aiter logs every fused-MoE problem it dispatches as a 14-field tuple. The +#: wording before it varies -- measured across 2948 real lines there are three +#: forms, and one of them interposes its own parenthesised kernel names: +#: +#: [fused_moe] using 2stage default for ('gfx950', 256, 256, 4096, ...) +#: [fused_moe] no tuned FlyDSL config for ('gfx950', 256, 256, 4096, ...) +#: [fused_moe] using 2stage (kernelName1='...', kernelName2='...') for ('gfx950', ...) +#: +#: so the tuple is anchored on `` for (`` rather than on the wording. The field +#: order matches aiter's untuned CSV columns after dropping gfx and cu_num, which +#: the runtime supplies itself. _AITER_FUSED_MOE_TUPLE_RE = re.compile( - r"\[fused_moe\] using \S+ \S+ for \(\d+, \d+, \d+, \d+, \d+, \d+, " - r"'[^']*', '[^']*', '([^']*)', '([^']*)'" + r"\[fused_moe\].*? for \(" + r"'(?P[^']*)', " + r"(?P\d+), (?P\d+), (?P\d+), (?P\d+), " + r"(?P\d+), (?P\d+), " + r"'(?P[^']*)', '(?P[^']*)', " + r"'(?P[^']*)', '(?P[^']*)', '(?P[^']*)', " + r"(?PTrue|False), (?PTrue|False)\)" ) +#: Which dtypes fall in each of aiter's width buckets. Mirrors ``bit16_list`` / +#: ``bit8_list`` / ``bit4_list`` in +#: ``csrc/ck_gemm_moe_2stages_codegen/gemm_moe_ck2stages_common.py``. +_AITER_BIT16_DTYPES = frozenset({"bfloat16", "float16"}) +_AITER_BIT8_DTYPES = frozenset({"float8_e4m3fn", "float8_e4m3fnuz", "int8"}) +_AITER_BIT4_DTYPES = frozenset({"float4_e2m1fn_x2", "uint32", "int4"}) + +#: The fields that identify one MoE problem, ignoring the token count (which the +#: tuner sweeps) and cu_num/gfx (which the runtime supplies). +_FMOE_SHAPE_FIELDS = ( + "model_dim", + "inter_dim", + "expert", + "topk", + "act_type", + "dtype", + "q_dtype_a", + "q_dtype_w", + "q_type", + "use_g1u1", + "doweight_stage1", +) + + +def _aiter_moe_dtype_pair_supported(q_dtype_a: str, q_dtype_w: str) -> bool: + """Return whether aiter's CK MoE codegen has a kernel family for this pair. + + ``get_gemm1_kernels_list`` / ``get_gemm2_kernels_list`` pick a family from the + activation/weight widths and raise ``Unsupported data type combination`` for + anything else. Notably a BF16 activation against FP4 weights -- which the + serving path runs happily -- matches no family, so handing it to the tuner + trades a silent no-op for a hard error. + """ + act = q_dtype_a.replace("torch.", "") + weight = q_dtype_w.replace("torch.", "") + if act in _AITER_BIT16_DTYPES and weight in _AITER_BIT16_DTYPES: + return True + if act in _AITER_BIT8_DTYPES and weight in _AITER_BIT8_DTYPES: + return True + # The a8w4 family is FP8-only on the activation side; INT8 does not qualify. + if act.startswith("float8") and weight in _AITER_BIT4_DTYPES: + return True + return act in _AITER_BIT4_DTYPES and weight in _AITER_BIT4_DTYPES + + +def _aiter_fused_moe_dispatch_keys(server_log: str) -> list[dict[str, str]]: + """Return the distinct MoE problems a server log shows aiter dispatching. + + Deduplicated on everything but the token count, preserving first-seen order. + One model routinely yields several problems -- the same checkpoint dispatches + both a BF16-activation and an FP8-activation variant, and the EP path appends + a masked fake-expert slot so ``expert``/``topk`` arrive one higher than the + model config states. Neither is derivable from the config, which is why the + log is the authoritative source for what to tune. + """ + if not server_log: + return [] + try: + text = Path(server_log).read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + seen: dict[tuple[str, ...], dict[str, str]] = {} + for match in _AITER_FUSED_MOE_TUPLE_RE.finditer(text): + fields = match.groupdict() + identity = tuple(fields[name] for name in _FMOE_SHAPE_FIELDS) + if identity not in seen: + seen[identity] = fields + return list(seen.values()) + def _aiter_ck_moe_tuner_supports(server_log: str) -> bool: - """Return whether aiter's CK MoE tuner can tune what the server dispatched. + """Return whether aiter's CK MoE tuner can tune anything the server dispatched. The tuner builds its kernel candidates from the activation/weight dtype pair and rejects some combinations the serving path happily runs. Measured on @@ -2823,23 +2905,150 @@ def _aiter_ck_moe_tuner_supports(server_log: str) -> bool: ``AITER_MXFP4_BF16`` backend) benchmarks fine but fails candidate generation with ``Unsupported data type combination: b16, fp4x2``, so routing it to ``fmoe_ck`` would only trade silent no-op for a hard tuner error. + + A single checkpoint can dispatch several dtype pairs at once, so this asks + whether *any* of them is tunable; per-problem filtering happens where the + tuning input is written. """ if not server_log: return False - try: - text = Path(server_log).read_text(encoding="utf-8", errors="replace") - except OSError: - return False - combos = { - (q_a.replace("torch.", ""), q_w.replace("torch.", "")) - for q_a, q_w in _AITER_FUSED_MOE_TUPLE_RE.findall(text) - } - if not combos: + keys = _aiter_fused_moe_dispatch_keys(server_log) + if not keys: # MoE evidence without a parseable problem tuple: let Forge decide. return True - return not any( - act.startswith("bfloat") and weight.startswith("float4") for act, weight in combos + return any( + _aiter_moe_dtype_pair_supported(key["q_dtype_a"], key["q_dtype_w"]) + for key in keys + ) + + +#: Header aiter's MoE tuner expects for its untuned input CSV. +_FMOE_UNTUNED_CSV_HEADER = ( + "token,model_dim,inter_dim,expert,topk,act_type,dtype," + "q_dtype_a,q_dtype_w,q_type,use_g1u1,doweight_stage1" +) + +#: forge wordings that carry nothing deployable, each distinct from an honest +#: ``no_improvement``. ``build_report`` checks ``has_candidate`` first, so a run +#: holding a usable env reports ``candidate`` even when a sibling crashed -- +#: these arrive only with nothing to deploy. +_FORGE_BARREN_MICRO_DECISIONS = ( + "failed", + "empty_output", + "partial_failure", + "partial_output", +) + + +def _fmoe_token_list(tokens: Any) -> list[int]: + """Positive token counts to sweep, keyed off whatever the caller sends. + + Accepts forge's comma-separated string (what :func:`_normalize_tokens` + produces) or a sequence. Unparseable and non-positive entries are dropped + rather than raising -- one bad entry is not worth the run -- and ``[1]`` is + the floor so there is always a token to key on. + """ + if isinstance(tokens, str): + raw: list[str] = [part.strip() for part in tokens.split(",")] + elif isinstance(tokens, (list, tuple, set, frozenset)): + raw = [str(item).strip() for item in tokens] + elif tokens is None: + raw = [] + else: + raw = [str(tokens).strip()] + + out: set[int] = set() + for item in raw: + if not item: + continue + try: + value = int(item) + except (TypeError, ValueError): + continue + if value > 0: + out.add(value) + return sorted(out) or [1] + + +def _write_fmoe_untuned_csv_from_log( + server_log: str, + tokens: Any, + workspace: Path, +) -> tuple[str, dict[str, Any]]: + """Turn the MoE problems observed in ``server_log`` into a tuning input CSV. + + Returns ``(csv_path, report)``; ``csv_path`` is "" when nothing tunable was + observed. Writing the observed tuple verbatim is the whole point: the + quantisation pair, the per-partition ``inter_dim`` and the EP-inflated + expert/topk counts are all properties of what the serving framework chose, + and every attempt to re-derive them from the model config is a guess that has + already produced tables no runtime lookup could reach. + + Problems whose dtype pair aiter's codegen rejects are dropped rather than + passed through, because one unsupported row aborts the whole tuner run. + """ + report: dict[str, Any] = { + "observed": 0, + "tunable": 0, + "dropped_unsupported": [], + "keys": [], + } + keys = _aiter_fused_moe_dispatch_keys(server_log) + report["observed"] = len(keys) + if not keys: + return "", report + + tunable: list[dict[str, str]] = [] + for key in keys: + pair = (key["q_dtype_a"], key["q_dtype_w"]) + if _aiter_moe_dtype_pair_supported(*pair): + tunable.append(key) + report["keys"].append( + {name: key[name] for name in _FMOE_SHAPE_FIELDS} + ) + else: + combo = f"{pair[0]}/{pair[1]}" + if combo not in report["dropped_unsupported"]: + report["dropped_unsupported"].append(combo) + report["tunable"] = len(tunable) + if not tunable: + return "", report + + token_list = _fmoe_token_list(tokens) + lines = [_FMOE_UNTUNED_CSV_HEADER] + for key in tunable: + for token in token_list: + lines.append( + f"{token},{key['model_dim']},{key['inter_dim']}," + f"{key['expert']},{key['topk']},{key['act_type']},{key['dtype']}," + f"{key['q_dtype_a']},{key['q_dtype_w']},{key['q_type']}," + f"{1 if key['use_g1u1'] == 'True' else 0}," + f"{1 if key['doweight_stage1'] == 'True' else 0}" + ) + + csv_path = workspace / "untuned_fmoe_from_runtime.csv" + try: + workspace.mkdir(parents=True, exist_ok=True) + csv_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + except OSError as exc: + # A full disk or a read-only workspace must cost the MoE tuner its input, + # not the whole tuning run: the dense tuners take their shapes from + # elsewhere and can still produce something useful. + report["write_error"] = f"{type(exc).__name__}: {exc}" + log.warning("Forge GEMM shapes: cannot write %s: %s", csv_path, exc) + return "", report + log.info( + "Forge GEMM shapes: derived %d MoE problem(s) x %d token(s) from %s%s", + len(tunable), + len(token_list), + server_log, + ( + f"; dropped {report['dropped_unsupported']} as untunable by aiter" + if report["dropped_unsupported"] + else "" + ), ) + return str(csv_path), report def _aiter_serving_evidence(server_log: str) -> set[str]: @@ -3536,7 +3745,8 @@ async def _run_forge_gemm_tuning( Forge receives a validated local directory, while result provenance and durable artifact names retain the original logical model identifier. Missing inputs return ``model_path_missing``; inputs that cannot resolve to - a local directory return ``model_path_unavailable``. + a local directory return ``model_path_unavailable`` as a ``skipped`` result, + because forge never ran and so has no verdict to report. """ from ..state.shared_state import SharedState @@ -3570,16 +3780,26 @@ async def _run_forge_gemm_tuning( ).strip() if not raw_model_path: return {"status": "failed", "error_class": "model_path_missing", "error": "model_path is required"} + from hyperloom.common.model_paths import resolve_serving_model_path from hyperloom.inference_optimizer.model_config_utils import ( resolve_local_model_dir, ) - resolved_model_dir = resolve_local_model_dir(raw_model_path) + # Bootstrap already walked HL_MODEL_BASE and the hub cache to decide what to + # serve; probing only the hub cache here would reject a repo id that the + # running server resolved fine. + resolved_model_dir = resolve_local_model_dir( + resolve_serving_model_path(raw_model_path) or raw_model_path + ) if resolved_model_dir is None: + # Forge needs the config on disk to derive shapes, so it cannot run -- + # but not running one tuning backend is a skip, not a session failure. + # Reporting it as failed spends a REVERT verdict on an experiment that + # never started, which is the misattribution this change set removes. return { - "status": "failed", + "status": "skipped", "error_class": "model_path_unavailable", - "error": ( + "skip_reason": ( f"Model path {raw_model_path!r} is neither an existing local " "directory nor an available Hugging Face cache snapshot" ), @@ -3627,6 +3847,20 @@ async def _run_forge_gemm_tuning( resolved_model_path, ) + # MoE shapes come from the runtime, never from inference. The dispatch tuple + # in the server log states the quantisation pair, the per-partition inter_dim + # and the EP-inflated expert/topk counts; none of the three is recoverable + # from the model config, and guessing them is what produced tuned tables no + # runtime lookup could reach. + moe_untuned_csv = str(payload.get("moe_untuned_csv") or "").strip() + if moe_untuned_csv and not _path_is_existing_file(moe_untuned_csv): + moe_untuned_csv = "" + moe_key_report: dict[str, Any] = {} + if not moe_untuned_csv: + moe_untuned_csv, moe_key_report = _write_fmoe_untuned_csv_from_log( + kernel_sig_log, tokens, workspace + ) + tunableop_input = str(payload.get("tunableop_input") or "").strip() forge_framework = _forge_framework_for_vllm( framework=framework, @@ -3747,6 +3981,7 @@ async def _run_forge_gemm_tuning( "skip_gpu_check": True, "tokens": tokens, "untuned_csv": untuned_csv, + "moe_untuned_csv": moe_untuned_csv, "shapes_json": shapes_json, "tunableop_input": tunableop_input, "kernel_signature_log": kernel_sig_log, @@ -3788,6 +4023,11 @@ async def _run_forge_gemm_tuning( result.setdefault("framework", framework) result.setdefault("tuning_framework", forge_framework) result.setdefault("model_path", raw_model_path) + if moe_key_report: + # Kept even when nothing was tunable: "no MoE problem was observed" and + # "the observed pair is one aiter cannot tune" lead to different actions, + # and neither is visible from the tuner's own status. + result.setdefault("moe_key_source", moe_key_report) if shape_alignment is not None: result.setdefault("shape_alignment", shape_alignment) if shape_capture is not None: @@ -3815,6 +4055,24 @@ async def _run_forge_gemm_tuning( if reason: result["skip_reason"] = reason + # The breakdown and the stack read the envelope, not the jsonl audit row, so + # a tuner's own error class has to surface here too. Lifted before the bridge + # so a specific class outranks the generic wording. ``tuners_run`` is forge's + # JSON and may be any shape; this is bookkeeping and must not raise. + _tuner_rows = result.get("tuners_run") + if not isinstance(_tuner_rows, list): + _tuner_rows = [] + if not result.get("error_class"): + for _t in _tuner_rows: + if isinstance(_t, dict) and _t.get("error_class"): + result["error_class"] = str(_t["error_class"]) + break + if not result.get("error"): + for _t in _tuner_rows: + if isinstance(_t, dict) and _t.get("error"): + result["error"] = str(_t["error"]) + break + # Bridge forge schema → coordinator schema: a "candidate" micro_decision with # recommended_env becomes decision="KEEP" + extra_envs. micro = str(result.get("micro_decision") or "").strip().lower() @@ -3847,23 +4105,35 @@ async def _run_forge_gemm_tuning( # Micro-only result: E2E validation still needed. result.setdefault("requires_e2e_validation", True) elif micro in ("no_improvement", "skipped"): + # Left unadorned on purpose: the wordings below are only legible against it. result.setdefault("decision", "REVERT") - elif micro == "failed": + elif micro in _FORGE_BARREN_MICRO_DECISIONS: result.setdefault("decision", "REVERT") - result.setdefault("status", "failed") + if micro == "failed": + result.setdefault("status", "failed") + result.setdefault("error_class", f"forge_{micro}") return result def _persist_forge_gemm_csv_durably(extra_envs: dict, *, model_path: str, session_dir: Path) -> tuple[dict, str]: - """Make a forge GEMM tuned CSV durable + recipe-portable. + """Make forge GEMM tuned CSVs durable + recipe-portable. - The forge KEEP references the tuned CSV by its ephemeral tuner-workspace path, + The forge KEEP references tuned CSVs by their ephemeral tuner-workspace paths, so a recipe replayed after the workspace is gone (or on another box) loses the tuning and aiter falls back to its default config. Mirror integrate_patch's - durability: copy the CSV into the serving aiter's ``configs/model_configs/`` - (where aiter loads it), repoint the env there, and snapshot the realized file - via :func:`snapshot_source_layer` so it travels with the recipe. + durability: copy each CSV into the serving aiter config tree, repoint the env + there, and snapshot the realized files via :func:`snapshot_source_layer` so + they travel with the recipe. + + The copy lands one level below ``configs/model_configs/`` on purpose. aiter + merges every ``model_configs/*{table}*.csv`` it can glob whenever the env var + is unset, and that glob is not recursive. Writing directly into that + directory would hand the table to every later server start -- including after + E2E rejected the candidate, and including servers for other models, since the + scan does not discriminate by model. Replay does not need the scan: it + restores the env var explicitly (see ``prelude._warm_kernel_extra_envs``) and + defers a GEMM column that has no env at all. The snapshot lands under ``/optimization_stack/src/`` (the same durable, run-cleanup-surviving location integrate_patch uses) -- NOT under the @@ -3873,14 +4143,35 @@ def _persist_forge_gemm_csv_durably(extra_envs: dict, *, model_path: str, sessio Best-effort: on any error the env is returned unchanged (never breaks the KEEP). Returns ``(extra_envs, source_snapshot_dir)``. """ - env_key = "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE" - src_csv = str(extra_envs.get(env_key) or "").strip() - if not src_csv or not Path(src_csv).is_file(): + # Below model_configs/, out of reach of aiter's non-recursive auto-merge glob. + _FORGE_DURABLE_SUBDIR = "hyperloom" + _forge_durable_env_stems = { + "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE": "a8w8_blockscale_bpreshuffle_tuned_gemm", + "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE": "a8w8_blockscale_tuned_gemm", + "AITER_CONFIG_GEMM_A8W8_BPRESHUFFLE": "a8w8_bpreshuffle_tuned_gemm", + "AITER_CONFIG_GEMM_A8W8": "a8w8_tuned_gemm", + "AITER_CONFIG_GEMM_A4W4": "a4w4_blockscale_tuned_gemm", + "AITER_CONFIG_GEMM_BF16": "bf16_tuned_gemm", + "AITER_CONFIG_FMOE": "tuned_fmoe", + } + slug = ( + "".join(c if (c.isalnum() or c in "._-") else "_" for c in Path(model_path).name).strip("_").lower() + or "model" + ) + + pending: list[tuple[str, str, Path]] = [] + for env_key, stem in _forge_durable_env_stems.items(): + src_csv = str(extra_envs.get(env_key) or "").strip() + if not src_csv or not Path(src_csv).is_file(): + continue + rel = f"configs/model_configs/{_FORGE_DURABLE_SUBDIR}/{stem}_{slug}.csv" + pending.append((env_key, rel, Path(src_csv))) + if not pending: return extra_envs, "" - # Step 1 -- commit the durable copy + env repoint. This is what makes the - # KEEP survive: the CSV lands in aiter's default config dir and the env - # points there instead of the ephemeral tuner workspace. + # Step 1 -- commit durable copies + env repoints. This is what makes the + # KEEP survive: each CSV lands in aiter's config dir and the env points + # there instead of the ephemeral tuner workspace. try: import importlib.util @@ -3888,24 +4179,20 @@ def _persist_forge_gemm_csv_durably(extra_envs: dict, *, model_path: str, sessio if spec is None or not spec.origin: return extra_envs, "" aiter_pkg = Path(spec.origin).resolve().parent - slug = ( - "".join(c if (c.isalnum() or c in "._-") else "_" for c in Path(model_path).name).strip("_").lower() - or "model" - ) - rel = f"configs/model_configs/a8w8_blockscale_tuned_gemm_{slug}.csv" - dst = aiter_pkg / rel - dst.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(src_csv, dst) updated = dict(extra_envs) - updated[env_key] = str(dst) + rel_paths: list[str] = [] + for env_key, rel, src_path in pending: + dst = aiter_pkg / rel + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src_path, dst) + updated[env_key] = str(dst) + rel_paths.append(rel) except Exception: # noqa: BLE001 — durability is best-effort; never break the KEEP log.exception("forge gemm CSV durable-copy failed; keeping workspace path") return extra_envs, "" # Step 2 -- recipe-portability snapshot. Separate best-effort concern: a - # snapshot failure must NOT discard the copy + repoint committed above (the - # tuned CSV already lives in aiter's config dir and the env already points - # at it), so this runs in its own guard and only affects the returned dir. + # snapshot failure must NOT discard the copy + repoint committed above. snap_dir = "" try: from ..source_snapshot import snapshot_source_layer @@ -3913,12 +4200,13 @@ def _persist_forge_gemm_csv_durably(extra_envs: dict, *, model_path: str, sessio snap = snapshot_source_layer( framework_root=aiter_pkg, base_sha=None, - rel_paths=[rel], - # Durable, run-cleanup-surviving location (mirrors integrate_patch), - # NOT the ephemeral runs/gemm_tuning workspace. + rel_paths=rel_paths, dest_dir=Path(session_dir) / "optimization_stack" / "src" / f"forge_gemm_{slug}", provenance="forge_gemm_tune", - extra={"env_key": env_key, "model": slug}, + extra={ + "env_keys": [env_key for env_key, _, _ in pending], + "model": slug, + }, ) snap_dir = str((snap or {}).get("snapshot_dir") or "") except Exception: # noqa: BLE001 — snapshot is best-effort; the repoint above stands diff --git a/src/hyperloom/orchestrator/phases/kernel.py b/src/hyperloom/orchestrator/phases/kernel.py index a7710fb91d..7130446554 100644 --- a/src/hyperloom/orchestrator/phases/kernel.py +++ b/src/hyperloom/orchestrator/phases/kernel.py @@ -81,6 +81,46 @@ } +def _safe_mtime(path: Path) -> float: + """Return ``path``'s mtime, or ``0`` when it cannot be read. + + Sorting server logs by mtime races the round that is still writing them, and + an ``exists()`` guard does not close the window. Ordering is a heuristic for + picking the newest log, so a vanished file is worth sorting last rather than + aborting the check that owns it. + """ + try: + return path.stat().st_mtime + except OSError: + return 0.0 + + +def _candidate_tuned_file(env: Any, env_var: str) -> str: + """Return the tuned artifact a candidate's env points at. + + One KEEP is described by three different path strings -- the durable copy in + aiter's config tree, the tuner-workspace original, and the E2E merge product + -- so an attempt row cannot re-derive the one the stack ends up holding, and + reconstructing it matched none of them: every forge KEEP read as unadopted. + + Reading the newest stack entry back is not the way out either. The stack + append is skipped when ``(action, variant_name)`` already matches, and a GEMM + variant is named ``_`` -- so a second macro cycle re-tuning + the same tuner finds its entry present, appends nothing, and the newest entry + is the previous round's. The attempt would then claim that round's artifact + along with its gain: the same misreport as before, inverted. + + Both the stack entry and the attempt row take the value from here, which + makes them the same string by construction rather than by lookup. + """ + if not isinstance(env, dict): + return "" + value = env.get(env_var) + if value in (None, ""): + value = next((v for v in env.values() if v not in (None, "")), "") + return str(value or "") + + def _paired_measurement_basis(verdict: Any) -> str: """How the promoted gain was measured, so the ledger cannot overstate it. @@ -1834,6 +1874,27 @@ def _gemm_tuned_config_coverage( the real cause -- an artifact the runtime never applied -- stays invisible. Replaying the lookup against the round's ``server.log`` separates the two. + Its result can block a KEEP, so an unexpected failure must not: it would + turn a diagnostic into the very false REVERT this replaces. Any + exception degrades to "undetermined", matching ``_gemm_apply_verdict``. + """ + try: + return self._gemm_tuned_config_coverage_impl(tuner_name, envs) + except Exception: # noqa: BLE001 + log.warning( + "tuned-config coverage failed for %s; treating it as undetermined", + tuner_name, + exc_info=True, + ) + return None + + def _gemm_tuned_config_coverage_impl( + self, + tuner_name: str, + envs: dict[str, str], + ) -> dict[str, Any] | None: + """Replay aiter's lookup against the round's log (see the caller). + For ``fmoe_ck``, delegates to ``_fmoe_tuned_config_coverage``, which matches fused-MoE dispatch lines against ``candidate_fmoe.csv`` rather than dense ``(M, N, K)`` GEMM lookups. @@ -1851,13 +1912,30 @@ def _gemm_tuned_config_coverage( if not csv_paths: return None run_dir = self.session_dir / "runs" / "integrate" / f"integrate-gemm_tune_{tuner_name}" - logs = sorted(run_dir.rglob("server.log"), key=lambda p: p.stat().st_mtime if p.exists() else 0) + logs = sorted(run_dir.rglob("server.log"), key=_safe_mtime) if not logs: return None try: log_text = logs[-1].read_text(encoding="utf-8", errors="replace") except OSError: return None + + def _unreadable(kind: str) -> None: + """Log that the artifact could not be read, so the caller stays out of it. + + A CSV we cannot parse is an absence of evidence, not evidence the + runtime ignored the table. Returning a 0% report would let that + absence block a KEEP whose throughput genuinely improved -- the + same conflation this change set exists to remove. + """ + log.warning( + "gemm E2E: tuner=%s %s tuned CSV yielded no keys from %s; " + "coverage is undetermined and will not block the KEEP", + tuner_name, + kind, + csv_paths, + ) + missed, hit = parse_aiter_shape_lookups(log_text) requested = missed | hit if not requested: @@ -1865,6 +1943,9 @@ def _gemm_tuned_config_coverage( tuned: set[tuple[int, int, int]] = set() for path in csv_paths: tuned |= tuned_csv_shapes(path) + if not tuned: + _unreadable("dense") + return None report = tuned_config_coverage(tuned, requested) report["server_log"] = str(logs[-1]) report["runtime_lookup_miss"] = len(missed) @@ -2059,10 +2140,7 @@ def _gemm_apply_verdict( if not csv_paths: return None run_dir = self.session_dir / "runs" / "integrate" / f"integrate-gemm_tune_{tuner_name}" - logs = sorted( - run_dir.rglob("server.log"), - key=lambda p: p.stat().st_mtime if p.exists() else 0, - ) + logs = sorted(run_dir.rglob("server.log"), key=_safe_mtime) if not logs: # Say so. This whole change exists to stop checks from failing # quietly, and a missing log is the one way this one can. @@ -2640,7 +2718,38 @@ async def _handle_gemm_tuning_result(self, result: dict[str, Any]) -> None: """ self._sync_profile_state_after_gemm_roofline(result) self.shared_state.record_gemm_tuning(result) - await self._validate_gemm_tuning_e2e(result) + try: + await self._validate_gemm_tuning_e2e(result) + except Exception as exc: # noqa: BLE001 + # Validation spans server restarts, log parsing and CSV merges, and + # is reached from two entrypoints that only guard the tuning call + # itself. An unexpected failure here has to read as "this candidate + # was never measured", not take the KERNEL phase down with it -- + # tuning that produced nothing measurable is the outcome this whole + # change exists to record honestly. + log.exception("gemm E2E validation raised; recording it as a fault") + e2e = result.setdefault("e2e_results", {}) + if isinstance(e2e, dict): + faults = e2e.setdefault("faults", []) + if isinstance(faults, list): + faults.append( + { + "tuner": "*", + "error_class": "e2e_validation_exception", + "error": f"{type(exc).__name__}: {exc}", + } + ) + # The bridge stamped KEEP + the raw combined env on the micro result; + # the normal exit rewrites both so Orchestration never bundles an + # integrate against an unmeasured candidate. This arm was not + # measured, so it reads as REVERT. + result["decision"] = "REVERT" + result["requires_e2e_validation"] = False + result["e2e_validated"] = False + result["micro_decision"] = "e2e_validation_exception" + for stale in ("recommended_env", "extra_envs"): + if result.get(stale): + result[stale] = {} try: from hyperloom.inference_optimizer.breakdown.recorder import instrument @@ -2832,6 +2941,7 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: A round the run stopped ends the sweep with its tuners unrecorded. """ from ..kernel.request_handlers import integrate_handler + from hyperloom.common.model_paths import resolve_session_model_path backend = str(result.get("backend") or "geak").strip().lower() candidates = self._gemm_e2e_candidates(result) @@ -2844,6 +2954,9 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: stacked_envs: dict[str, str] = {} kept: list[dict[str, Any]] = [] reverted: list[dict[str, Any]] = [] + faults: list[dict[str, Any]] = [] + # Set by the last KEEP; the attempt row claims this exact string. + adopted_tuned_file = "" try: from ..actions.executors.explore import _compute_explore_variant_timeout @@ -2947,42 +3060,111 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: running_tput, ) - try: - integrate_result = await integrate_handler( - { - "task_id": f"gemm_tune_e2e_{tuner_name}", - "kernel_id": f"gemm_tune_{tuner_name}", - "source": "forge_gemm_tuning", - "base_tput": running_tput, - "extra_server_args": extra_server_args, - "extra_envs": test_envs, - "keep_threshold_pct": 3.0, - "budget_minutes": per_tuner_budget_minutes, - }, - session_dir=self.session_dir, - ) - except Exception as exc: # noqa: BLE001 - log.warning( - "gemm E2E: integrate failed for %s: %s", - tuner_name, - exc, - ) - reverted.append({**cand, "reason": repr(exc)}) - continue + from ..state.kernel_decision_settings import _MAX_INTEGRATE_FAULT_ATTEMPTS + + integrate_verdict: dict[str, Any] | None = None + run_stopped = False + integrate_payload = { + "task_id": f"gemm_tune_e2e_{tuner_name}", + "kernel_id": f"gemm_tune_{tuner_name}", + "source": "forge_gemm_tuning", + "base_tput": running_tput, + "model_path": resolve_session_model_path( + state_model_path=str(getattr(self.shared_state, "model_path", "") or ""), + for_serving=True, + ), + "extra_server_args": extra_server_args, + "extra_envs": test_envs, + "keep_threshold_pct": 3.0, + "budget_minutes": per_tuner_budget_minutes, + } + for fault_attempt in range(1, _MAX_INTEGRATE_FAULT_ATTEMPTS + 1): + try: + integrate_result = await integrate_handler( + integrate_payload, + session_dir=self.session_dir, + ) + except Exception as exc: # noqa: BLE001 + if fault_attempt < _MAX_INTEGRATE_FAULT_ATTEMPTS: + log.warning( + "gemm E2E: integrate raised for %s (fault attempt %d/%d): %s", + tuner_name, + fault_attempt, + _MAX_INTEGRATE_FAULT_ATTEMPTS, + exc, + ) + continue + log.warning( + "gemm E2E: integrate raised for %s: %s", + tuner_name, + exc, + ) + faults.append( + { + **cand, + "reason": "integrate_fault:handler_exception", + "fault": True, + "error_class": "handler_exception", + "error": repr(exc), + "fault_attempts": fault_attempt, + } + ) + break - stopped = stopped_by_the_run_class(integrate_result.get("error_class")) - if stopped is not None: - # Recording the rest would report a clock as a verdict on them. - log.info( - "gemm E2E: %s left unmeasured — %s", - tuner_name, - stopped.interrupted, - ) + stopped = stopped_by_the_run_class(integrate_result.get("error_class")) + if stopped is not None: + log.info( + "gemm E2E: %s left unmeasured — %s", + tuner_name, + stopped.interrupted, + ) + run_stopped = True + break + + if self.shared_state._is_integrate_fault(integrate_result): + error_class = str( + integrate_result.get("error_class") or "integrate_fault" + ).strip() + if fault_attempt < _MAX_INTEGRATE_FAULT_ATTEMPTS: + log.warning( + "gemm E2E: retrying tuner=%s after integrate fault %s " + "(attempt %d/%d)", + tuner_name, + error_class, + fault_attempt, + _MAX_INTEGRATE_FAULT_ATTEMPTS, + ) + continue + log.warning( + "gemm E2E: tuner=%s integrate fault (%s) — unmeasured, " + "not a REVERT verdict", + tuner_name, + error_class, + ) + faults.append( + { + **cand, + "reason": f"integrate_fault:{error_class}", + "fault": True, + "error_class": error_class, + "integrate_status": integrate_result.get("status"), + "error": integrate_result.get("error"), + "fault_attempts": fault_attempt, + } + ) + break + + integrate_verdict = integrate_result break - decision = str(integrate_result.get("decision") or "").upper() - new_tput = float(integrate_result.get("new_tput") or 0.0) - gain_pct = float(integrate_result.get("gain_pct") or 0.0) + if run_stopped: + break + if integrate_verdict is None: + continue + + decision = str(integrate_verdict.get("decision") or "").upper() + new_tput = float(integrate_verdict.get("new_tput") or 0.0) + gain_pct = float(integrate_verdict.get("gain_pct") or 0.0) log.info( "gemm E2E: tuner=%s decision=%s new_tput=%.1f gain=%.2f%%", @@ -3064,6 +3246,11 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: "gain_pct": gain_pct, } ) + # The one place this path names its artifact. The stack entry + # below and the attempt row further down both read it, so the + # breakdown's string match cannot be defeated by a stack append + # that was skipped as already-applied. + adopted_tuned_file = _candidate_tuned_file(env, cand.get("env_var", "")) lifted = self._lift_to_current_best( "gemm_tuning", @@ -3076,10 +3263,7 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: "workspace": result.get("workspace"), }, entry_extra={ - "tuned_file": ( - env.get(cand["env_var"]) - or next(iter(env.values()), "") - ), + "tuned_file": adopted_tuned_file, "gain_pct": gain_pct, "backend": backend, "source": "kernel_entry_auto", @@ -3127,12 +3311,26 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: source="forge_gemm_tuning_e2e", measurement_basis=_paired_measurement_basis(paired), ) + # Name the artifact this run adopted, so the breakdown can tell it + # was. Forge never set ``tuned_file`` (it reports per-tuner envs + # instead), which left the history row's path empty and the adoption + # lookup matching on "". The value is the one the stack entry above + # carries, taken from the same call rather than looked up. + if adopted_tuned_file: + result["tuned_file"] = adopted_tuned_file log.info( "gemm E2E: %d tuners KEEP (total gain=+%.2f%%), %d REVERT", len(kept), total_gain, len(reverted), ) + elif faults: + stacked_envs = {} + total_gain = 0.0 + log.info( + "gemm E2E: %d tuner(s) hit integrate fault(s), no E2E verdict", + len(faults), + ) else: stacked_envs = {} total_gain = 0.0 @@ -3143,17 +3341,28 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: # Rewrite the stored result to the E2E-validated outcome so Orchestration # never sees the raw combined recommended_env and issues a bundled integrate. - result["e2e_results"] = {"kept": kept, "reverted": reverted} + result["e2e_results"] = {"kept": kept, "reverted": reverted, "faults": faults} result["recommended_env_raw"] = dict(result.get("recommended_env") or {}) result["extra_envs_raw"] = dict(result.get("extra_envs") or {}) result["recommended_env"] = dict(stacked_envs) result["extra_envs"] = dict(stacked_envs) - result["e2e_gain_pct"] = round(float(total_gain), 4) + if faults and not kept and not reverted: + result["e2e_gain_pct"] = None + else: + result["e2e_gain_pct"] = round(float(total_gain), 4) result["e2e_validated"] = True result["requires_e2e_validation"] = False if kept: result["status"] = "complete" result["decision"] = "KEEP" + elif reverted: + result["status"] = "complete" + result["decision"] = "REVERT" + result["micro_decision"] = "candidate_no_e2e_gain" + elif faults: + result["status"] = "failed" + result["decision"] = "REVERT" + result["micro_decision"] = "integrate_fault" else: result["status"] = "complete" result["decision"] = "REVERT"