diff --git a/docs/reference/environment-variables.md b/docs/reference/environment-variables.md index fd01405fba..3946665cb9 100644 --- a/docs/reference/environment-variables.md +++ b/docs/reference/environment-variables.md @@ -189,6 +189,8 @@ The following variables control the kernel optimization backend ladder. | `HYPERLOOM_GEMM_SHAPE_CAPTURE_TIMEOUT_SEC` | `1800` | Timeout in seconds for the dense vLLM TunableOp recording benchmark. Block-FP8 fallback uses the standard Roofline/ProfileExecutor timeout. Values below `60` are clamped to `60`. | | `INFERENCE_OPTIMIZER`
`_KERNEL_OPT_MAX_PARTIAL` | Unset | Cap on how many `PARTIAL` kernel-opt verdicts an action can yield before it short-circuits to `NEEDS_REVIEW`. Useful for keeping budget contained when GEAK is consistently timing out. | | `KERNEL_OPT_BACKEND_BUDGET_MIN` | `60` | Wall-clock budget in minutes for one optimization, mirrored by the `kernel_optimization.py` wrapper. The env deliberately wins over the payload `budget_minutes`, which is LLM-authored from a prompt template, so an operator raising the budget is not silently overridden. forge-loop reserves half the window for finalize, so `60` leaves roughly 30 minutes of real iteration. | +| `AITER_LOG_TUNED_CONFIG` | `1` (set for every serving run) | Makes aiter log each tuned-config lookup it *hits*, not only the ones it misses. Two checks have no input without it: the GEMM demand list, which learns the shapes the runtime actually asks for (config-derived shapes covered 0.4% of them), and the apply verdict, which cannot tell "the tuned table was never read" from "it was read and did not help". A scan of 60 production logs found it set in none of them, so it is now injected by default. An operator value wins — set `0` to turn hit logging off, at the cost of both checks going inconclusive. Every miss already prints a line regardless of this setting; hit logging adds roughly one line per lookup that succeeds. | +| `HYPERLOOM_GEMM_PAIRED_PAIRS` | `0` (off) | How many interleaved baseline/tuned pairs to re-measure before a GEMM tuning KEEP is reported as confirmed. One end-to-end measurement cannot separate a gain from drift on this fleet: three rounds of a single unchanged configuration spanned 58%, and one controlled repeat moved 16%. Each pair costs two extra benchmark rounds. When `0`, the gain is still promoted — it is the best number available — but recorded as an unpaired block comparison rather than presented as a paired one. | --- 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 afb8f56599..a2f26bade6 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 @@ -3169,3 +3169,304 @@ async def _fake(payload, *, session_dir): # Fallback budget is 15 minutes. assert captured["budget"] == 15 + + +class TestForgeGemmE2EApplyGate: + """A measured gain is only creditable if the artifact was actually used. + + Both checks answer a question throughput cannot: the shape keys never + resolved (coverage), or the table never reached the server (apply verdict). + Each is a positive finding, so each blocks the KEEP -- while "cannot tell" + deliberately does not, because hit lines require AITER_LOG_TUNED_CONFIG=1 + and a scan of 60 production logs found it set in none of them. + """ + + @staticmethod + def _result(): + return { + "recommended_env": {"X": "1"}, + "extra_envs": {"X": "1"}, + "requires_e2e_validation": True, + "tuners_run": [ + { + "status": "ok", + "improved_shapes": 2, + "tuner": "dense", + "env_var": "X", + "env_value": "1", + "best_micro_speedup": 1.1, + }, + ], + } + + @staticmethod + def _wire(monkeypatch, *, coverage, verdict): + fake = _make_integrate([{"decision": "KEEP", "new_tput": 130.0, "gain_pct": 30.0}]) + monkeypatch.setattr(krh_mod, "integrate_handler", fake) + monkeypatch.setattr( + KernelPhase, "_gemm_tuned_config_coverage", lambda self, *a, **k: coverage + ) + monkeypatch.setattr( + KernelPhase, "_gemm_apply_verdict", lambda self, *a, **k: verdict + ) + return fake + + @pytest.mark.asyncio + async def test_unmerged_artifact_blocks_a_measured_keep(self, tmp_path, monkeypatch): + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + self._wire( + monkeypatch, + coverage=None, + verdict={ + "verdict": "not_merged", + "blocks_keep": True, + "conclusive": True, + "detail": "1 tuned table(s) absent from the server's merge list", + }, + ) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + # +30% was measured, and is still refused: the server was running its + # bundled default table, so the delta is drift, not tuning. + assert coord.shared_state.optimization_stack == [] + assert coord.shared_state.cumulative_gain_validated == 0.0 + assert result["decision"] == "REVERT" + reverted = result["e2e_results"]["reverted"] + assert len(reverted) == 1 + assert "tuned_config_never_applied[not_merged]" in reverted[0]["reason"] + assert reverted[0]["apply_verdict"]["verdict"] == "not_merged" + + @pytest.mark.asyncio + async def test_unreachable_shape_keys_block_a_measured_keep(self, tmp_path, monkeypatch): + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + self._wire( + monkeypatch, + coverage={ + "artifact_applied": False, + "not_applied_reason": "no_shape_key_matched", + "requested": 42, + "covered": 0, + }, + verdict=None, + ) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + assert coord.shared_state.optimization_stack == [] + assert result["decision"] == "REVERT" + reason = result["e2e_results"]["reverted"][0]["reason"] + assert "tuned_config_never_applied[no_shape_key_matched]" in reason + + @pytest.mark.asyncio + async def test_both_blockers_are_named(self, tmp_path, monkeypatch): + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + self._wire( + monkeypatch, + coverage={ + "artifact_applied": False, + "not_applied_reason": "artifact_table_not_consulted", + "requested": 7, + "covered": 0, + }, + verdict={"verdict": "not_merged", "blocks_keep": True, "conclusive": True}, + ) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + reason = result["e2e_results"]["reverted"][0]["reason"] + assert "artifact_table_not_consulted+not_merged" in reason + + @pytest.mark.asyncio + async def test_inconclusive_verdict_does_not_block(self, tmp_path, monkeypatch): + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + self._wire( + monkeypatch, + coverage={"artifact_applied": True, "coverage_pct": 88.0, "covered": 7, "requested": 8}, + verdict={ + "verdict": "inconclusive_no_hit_logging", + "blocks_keep": False, + "conclusive": False, + "detail": "misses logged but hit logging was off", + }, + ) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + assert result["decision"] == "KEEP" + assert len(coord.shared_state.optimization_stack) == 1 + assert coord.shared_state.current_best["tput"] == 130.0 + + @pytest.mark.asyncio + async def test_served_verdict_keeps(self, tmp_path, monkeypatch): + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + self._wire( + monkeypatch, + coverage={"artifact_applied": True, "coverage_pct": 100.0, "covered": 8, "requested": 8}, + verdict={ + "verdict": "served", + "blocks_keep": False, + "conclusive": True, + "hits": 512, + }, + ) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + assert result["decision"] == "KEEP" + kept = result["e2e_results"]["kept"] + assert kept[0]["apply_verdict"]["hits"] == 512 + assert coord.shared_state.cumulative_gain_validated == pytest.approx(30.0) + + @pytest.mark.asyncio + async def test_missing_evidence_leaves_the_decision_alone(self, tmp_path, monkeypatch): + """No server log at all must not turn into an accusation.""" + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + self._wire(monkeypatch, coverage=None, verdict=None) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + assert result["decision"] == "KEEP" + assert len(coord.shared_state.optimization_stack) == 1 + + +class TestForgeGemmPairedConfirmation: + """The promoted gain must say whether it was confirmed against drift. + + ``base_tput`` and ``new_tput`` are measured at different times, so the two + are a block comparison and any drift between them lands in the result. + Interleaving separates the two, and when it is not run the number is still + promoted -- but labelled for what it is. + """ + + @staticmethod + def _run_e2e(coord, monkeypatch, tputs): + fake = _make_integrate([ + {"decision": "KEEP", "new_tput": t, "gain_pct": (t - 100.0)} for t in tputs + ]) + monkeypatch.setattr(krh_mod, "integrate_handler", fake) + monkeypatch.setattr( + KernelPhase, "_gemm_tuned_config_coverage", lambda self, *a, **k: None + ) + monkeypatch.setattr(KernelPhase, "_gemm_apply_verdict", lambda self, *a, **k: None) + return fake + + @staticmethod + def _result(): + return { + "recommended_env": {"X": "1"}, + "extra_envs": {"X": "1"}, + "requires_e2e_validation": True, + "tuners_run": [ + { + "status": "ok", + "improved_shapes": 2, + "tuner": "dense", + "env_var": "X", + "env_value": "1", + "best_micro_speedup": 1.1, + }, + ], + } + + @pytest.mark.asyncio + async def test_unpaired_by_default_and_labelled_as_such(self, tmp_path, monkeypatch): + monkeypatch.delenv("HYPERLOOM_GEMM_PAIRED_PAIRS", raising=False) + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + basis: dict[str, str] = {} + monkeypatch.setattr( + coord, + "_update_cumulative_gain_validated", + lambda tput, **kw: basis.update( + {"basis": kw.get("measurement_basis", ""), "tput": tput} + ), + ) + fake = self._run_e2e(coord, monkeypatch, [130.0]) + + await coord._validate_gemm_tuning_e2e(self._result()) + + # One integrate call: the confirmation pass did not run. + assert len(fake.calls) == 1 + assert basis["basis"] == "e2e_rebench_unpaired" + assert basis["tput"] == 130.0 + + @pytest.mark.asyncio + async def test_paired_confirmation_runs_interleaved_and_labels_the_gain( + self, tmp_path, monkeypatch + ): + monkeypatch.setenv("HYPERLOOM_GEMM_PAIRED_PAIRS", "2") + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + basis: dict[str, str] = {} + monkeypatch.setattr( + coord, + "_update_cumulative_gain_validated", + lambda tput, **kw: basis.update({"basis": kw.get("measurement_basis", "")}), + ) + # 1 validation call, then A,B,A,B: baseline ~100, candidate ~130. + fake = self._run_e2e(coord, monkeypatch, [130.0, 100.0, 130.0, 101.0, 131.0]) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + assert len(fake.calls) == 5 + # The confirmation pass alternates env-free and env-carrying runs. + assert [bool(c["extra_envs"]) for c in fake.calls[1:]] == [False, True, False, True] + paired = result["paired_confirmation"] + assert paired["decisive"] is True + assert paired["reason"] == "candidate_faster" + assert len(paired["pairs"]) == 2 + assert basis["basis"] == "e2e_paired" + + @pytest.mark.asyncio + async def test_drifting_pairs_are_not_labelled_confirmed(self, tmp_path, monkeypatch): + monkeypatch.setenv("HYPERLOOM_GEMM_PAIRED_PAIRS", "2") + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + basis: dict[str, str] = {} + monkeypatch.setattr( + coord, + "_update_cumulative_gain_validated", + lambda tput, **kw: basis.update({"basis": kw.get("measurement_basis", "")}), + ) + # The pairs disagree about which side is faster: the machine moved. + self._run_e2e(coord, monkeypatch, [130.0, 100.0, 130.0, 140.0, 120.0]) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + assert result["paired_confirmation"]["reason"] == "sign_disagreement" + assert result["paired_confirmation"]["decisive"] is False + assert basis["basis"] == "e2e_paired_sign_disagreement" + + @pytest.mark.asyncio + async def test_confirmation_failure_falls_back_to_insufficient_pairs( + self, tmp_path, monkeypatch + ): + monkeypatch.setenv("HYPERLOOM_GEMM_PAIRED_PAIRS", "2") + coord = _coord(tmp_path, baseline_tput=100.0, framework="sglang") + calls: list[dict] = [] + + async def _fake(payload, *, session_dir): + calls.append(payload) + if len(calls) == 1: + return {"decision": "KEEP", "new_tput": 130.0, "gain_pct": 30.0} + raise RuntimeError("benchmark host went away") + + monkeypatch.setattr(krh_mod, "integrate_handler", _fake) + monkeypatch.setattr( + KernelPhase, "_gemm_tuned_config_coverage", lambda self, *a, **k: None + ) + monkeypatch.setattr(KernelPhase, "_gemm_apply_verdict", lambda self, *a, **k: None) + result = self._result() + + await coord._validate_gemm_tuning_e2e(result) + + # A confirmation that could not run must not revert the artifact, and + # must not claim to have confirmed anything either. + assert result["decision"] == "KEEP" + assert result["paired_confirmation"]["reason"] == "insufficient_pairs" diff --git a/src/hyperloom/inference_optimizer/tests/test_gemm_tuning_trace_row.py b/src/hyperloom/inference_optimizer/tests/test_gemm_tuning_trace_row.py new file mode 100644 index 0000000000..adf9fafd30 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_gemm_tuning_trace_row.py @@ -0,0 +1,206 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""The GEMM-tuning audit row has to survive a tuner that failed. + +``reports/trace/gemm_tuning.jsonl`` exists to answer one question: did this +tuner run? It used to record only ``tuner``/``best_micro_speedup``/``kept``, +which cannot separate a crash from a clean search that found nothing. Across one +campaign 38 of 337 tuner runs ended ``failed`` or ``empty_output`` and the trace +showed none of them. + +Both cases below are transcribed from the runs in the issue. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import hyperloom.orchestrator.kernel.request_handlers as krh +from hyperloom.inference_optimizer.session.session_paths import gemm_tuning_steps_path + + +def _rows(session_dir: Path) -> list[dict]: + text = gemm_tuning_steps_path(session_dir).read_text(encoding="utf-8") + return [json.loads(line) for line in text.splitlines() if line.strip()] + + +class TestFailureReachesTheAuditRow: + def test_a_crashed_tuner_is_not_indistinguishable_from_a_barren_one(self, tmp_path): + # DeepSeek-V4-Pro/20260815T002915Z: 288s of GPU time, an aiter failure + # and a subprocess_error class, none of which reached the trace. + krh._trace_gemm_tuning_run( + { + "status": "failed", + "backend": "forge", + "engine": "forge", + "decision": "REVERT", + "micro_decision": "failed", + "framework": "vllm-aiter", + "precision": "mxfp4", + "error_class": None, + "tuners_run": [ + { + "tuner": "fmoe_ck", + "status": "failed", + "elapsed_s": 288.54, + "error_class": "subprocess_error", + "error": ( + "Tuner exited with code 1: [Tuning not Finished] some " + "shapes are not tuned or all failed" + ), + } + ], + }, + session_dir=tmp_path, + ) + + (row,) = _rows(tmp_path) + tuner = row["tuners_run"][0] + assert tuner["status"] == "failed" + assert tuner["elapsed_s"] == 288.54 + assert tuner["error_class"] == "subprocess_error" + assert "Tuning not Finished" in tuner["error"] + # The envelope named no class although the tuner did; take the tuner's. + assert row["error_class"] == "subprocess_error" + + def test_an_argparse_rejection_no_longer_reads_as_a_clean_result(self, tmp_path): + # Llama-3.1-8B-Instruct/20260812T091612Z: rejected in 10.8s and recorded + # status "ok" / "no_improvement", identical in the trace to the six runs + # that really tuned. This is the blind spot that hid #1211 for 3 weeks. + krh._trace_gemm_tuning_run( + { + "status": "ok", + "engine": "forge", + "micro_decision": "no_improvement", + "tuners_run": [ + { + "tuner": "sglang_dense_bf16", + "status": "failed", + "elapsed_s": 10.79, + "error_class": "unsupported_argument", + "error": "gemm_tuner.py: error: unrecognized arguments: --libtype", + } + ], + }, + session_dir=tmp_path, + ) + + (row,) = _rows(tmp_path) + tuner = row["tuners_run"][0] + assert tuner["status"] == "failed" + assert tuner["error_class"] == "unsupported_argument" + assert "unrecognized arguments" in tuner["error"] + # Envelope status stays as the handler reported it -- the row records + # what happened, it does not re-decide it. The per-tuner fields are what + # make the two runs distinguishable. + assert row["status"] == "ok" + assert row["error_class"] == "unsupported_argument" + + +class TestRowStaysCompact: + def test_a_successful_run_gains_only_status_and_elapsed(self, tmp_path): + krh._trace_gemm_tuning_run( + { + "status": "ok", + "engine": "forge", + "tuners_run": [ + { + "tuner": "vllm_moe_triton", + "status": "ok", + "elapsed_s": 35.4, + "best_micro_speedup": 1.0973, + } + ], + }, + session_dir=tmp_path, + ) + + (row,) = _rows(tmp_path) + assert set(row["tuners_run"][0]) == { + "tuner", "best_micro_speedup", "kept", "status", "elapsed_s", + } + + def test_the_three_original_keys_survive_being_null(self, tmp_path): + # kept is null on every row observed so far; an absent key would be + # indistinguishable from false. + krh._trace_gemm_tuning_run( + {"status": "ok", "engine": "forge", "tuners_run": [{"tuner": "a8w8"}]}, + session_dir=tmp_path, + ) + + (row,) = _rows(tmp_path) + tuner = row["tuners_run"][0] + assert tuner == {"tuner": "a8w8", "best_micro_speedup": None, "kept": None} + + def test_a_long_error_is_truncated(self, tmp_path): + krh._trace_gemm_tuning_run( + { + "status": "failed", + "engine": "forge", + "tuners_run": [{"tuner": "fmoe_ck", "error": "x" * 5000}], + }, + session_dir=tmp_path, + ) + + (row,) = _rows(tmp_path) + error = row["tuners_run"][0]["error"] + assert len(error) == krh._TRACE_TUNER_ERROR_MAXLEN + 3 + assert error.endswith("...") + + +class TestRobustness: + def test_the_envelope_class_wins_when_it_has_one(self, tmp_path): + krh._trace_gemm_tuning_run( + { + "status": "failed", + "engine": "forge", + "error_class": "handler_error", + "tuners_run": [{"tuner": "a8w8", "error_class": "subprocess_error"}], + }, + session_dir=tmp_path, + ) + + assert _rows(tmp_path)[0]["error_class"] == "handler_error" + + def test_the_first_named_class_is_taken_not_the_last(self, tmp_path): + krh._trace_gemm_tuning_run( + { + "status": "failed", + "engine": "forge", + "tuners_run": [ + {"tuner": "ok_one", "status": "ok"}, + {"tuner": "first_bad", "error_class": "unsupported_argument"}, + {"tuner": "second_bad", "error_class": "subprocess_error"}, + ], + }, + session_dir=tmp_path, + ) + + assert _rows(tmp_path)[0]["error_class"] == "unsupported_argument" + + def test_non_dict_entries_are_skipped_without_losing_the_rest(self, tmp_path): + krh._trace_gemm_tuning_run( + { + "status": "ok", + "engine": "forge", + "tuners_run": ["garbage", None, {"tuner": "a8w8", "status": "ok"}], + }, + session_dir=tmp_path, + ) + + rows = _rows(tmp_path)[0]["tuners_run"] + assert [t["tuner"] for t in rows] == ["a8w8"] + + def test_a_non_string_error_is_passed_through_untruncated(self, tmp_path): + krh._trace_gemm_tuning_run( + { + "status": "failed", + "engine": "forge", + "tuners_run": [{"tuner": "a8w8", "error": {"code": 2}}], + }, + session_dir=tmp_path, + ) + + assert _rows(tmp_path)[0]["tuners_run"][0]["error"] == {"code": 2} 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 ad855c6d33..a4b1585972 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 @@ -4194,6 +4194,9 @@ async def fake_run(cmd: list[str], *, timeout_sec: int): assert row["engine"] == "forge" assert row["decision"] == "KEEP" assert row["tuners_run"][0]["tuner"] == "fmoe_ck" + # A clean run gains nothing it did not carry before, bar status. + assert "error" not in row["tuners_run"][0] + assert "error_class" not in row["tuners_run"][0] def test_forge_uses_per_token_only_for_explicit_env(self, tmp_path, monkeypatch): monkeypatch.setenv("KERNEL_OPT_BACKEND_ORDER", "forge") diff --git a/src/hyperloom/inference_optimizer/tests/test_workload_envs_golden_lock.py b/src/hyperloom/inference_optimizer/tests/test_workload_envs_golden_lock.py index edaa1184b0..8ef8cc23e9 100644 --- a/src/hyperloom/inference_optimizer/tests/test_workload_envs_golden_lock.py +++ b/src/hyperloom/inference_optimizer/tests/test_workload_envs_golden_lock.py @@ -2,7 +2,14 @@ # SPDX-License-Identifier: MIT """Behavior-lock tests for ``materialize_config_with_envs``: golden snapshots of -the real materialized YAML, plus RUN_EVAL warn-once and unset/extra_envs restore.""" +the real materialized YAML, plus RUN_EVAL warn-once and unset/extra_envs restore. + +The snapshots carry ``AITER_LOG_TUNED_CONFIG`` because every serving run now gets +it: aiter logs a tuned-config MISS unconditionally but the matching HIT only +under that flag, and both GEMM shape discovery and apply verification read hits. +Its position differs between the snapshots because ``extra_server_args`` seeds +``EXTRA_*_ARGS`` earlier in the dict when a caller passes one. +""" from __future__ import annotations @@ -62,6 +69,7 @@ def _write(path, bench): NUM_PROMPTS: 320 NUM_WARMUPS: 8 EXTRA_SGLANG_ARGS: --variant 4 --watchdog-timeout 1800 + AITER_LOG_TUNED_CONFIG: '1' MAGPIE_TRUST_REMOTE_CODE: '1' BENCH_TRUST_REMOTE_CODE: '1' HF_HUB_TRUST_REMOTE_CODE: '1' @@ -78,6 +86,7 @@ def _write(path, bench): NUM_PROMPTS: 320 NUM_WARMUPS: 8 EXTRA_VLLM_ARGS: --max-num-seqs 256 + AITER_LOG_TUNED_CONFIG: '1' MAGPIE_TRUST_REMOTE_CODE: '1' BENCH_TRUST_REMOTE_CODE: '1' HF_HUB_TRUST_REMOTE_CODE: '1' @@ -99,6 +108,7 @@ def _write(path, bench): true, "detailed_annotations": true}' NUM_PROMPTS: 776 NUM_WARMUPS: 8 + AITER_LOG_TUNED_CONFIG: '1' EXTRA_SGLANG_ARGS: --watchdog-timeout 1800 MAGPIE_TRUST_REMOTE_CODE: '1' BENCH_TRUST_REMOTE_CODE: '1' diff --git a/src/hyperloom/orchestrator/actions/executors/_workload_envs.py b/src/hyperloom/orchestrator/actions/executors/_workload_envs.py index bb2b1f015d..dbb9fb88d4 100644 --- a/src/hyperloom/orchestrator/actions/executors/_workload_envs.py +++ b/src/hyperloom/orchestrator/actions/executors/_workload_envs.py @@ -1224,6 +1224,19 @@ def materialize_config_with_envs( log.warning("Dropping invalid extra_envs key %s before benchmark materialization", _dk) for key, value in safe_extra_envs.items(): envs[str(key)] = str(value) + # ── aiter tuned-config lookup logging ──────────────────────────────────── + # aiter logs a line for every tuned-GEMM table lookup it MISSES + # unconditionally, but the corresponding HIT line only when this is set. Two + # things depend on having it on: + # * GEMM tuning takes its shape list from the misses -- shapes derived from + # config.json instead cover 0.4% of what the runtime actually asks for; + # * apply verification counts hits to decide whether a tuned artifact was + # ever read. Without hit lines, "0 hits" and "hit logging was off" are + # indistinguishable, and treating the latter as the former would REVERT + # every arm. + # A scan of 60 production server logs found the flag set in none of them, so + # this is not a hypothetical gap. setdefault keeps an operator override. + envs.setdefault("AITER_LOG_TUNED_CONFIG", "1") _sync_repo_aliases( bench, envs, diff --git a/src/hyperloom/orchestrator/actions/executors/baseline.py b/src/hyperloom/orchestrator/actions/executors/baseline.py index ba469f752d..18d9e9918f 100644 --- a/src/hyperloom/orchestrator/actions/executors/baseline.py +++ b/src/hyperloom/orchestrator/actions/executors/baseline.py @@ -660,6 +660,13 @@ def _should_establish_quality_ref(task_kind: str | None, params: dict[str, Any] return not (params or {}).get("quality_ref_exempt") +# Above this cold-start delta the measured round is unlikely to have settled +# either: the observed pathological case climbed 14,202 -> 19,374 -> 22,425 +# tok/s across three rounds of one unchanged config, i.e. +36% into round 2 and +# another +16% into round 3. +_COLD_START_DELTA_WARN_PCT = 25.0 + + def _is_double_run_accuracy_handoff( result: dict[str, Any], salvaged: dict[str, Any] | None, @@ -2244,6 +2251,48 @@ def _hit(text: str) -> bool: return False return False + @staticmethod + def _record_baseline_convergence( + result: dict[str, Any], + warmup_tput: Any, + ) -> None: + """Record how steady the baseline anchor actually is. + + The double-run discards round 1 by design, which leaves exactly one + usable measurement -- and one measurement cannot be shown to be steady. + That is worth stating rather than assuming: the whole gain ledger is + graded against this number with a 3% KEEP threshold, and a real session + once produced 14,202 -> 19,374 -> 22,425 tok/s from one unchanged + configuration. Establishing convergence needs a third round. + + So the verdict is recorded (it will read ``insufficient_rounds``) along + with the cold-start delta, and a warning is raised only when that delta + is large enough to suggest round 2 had not settled either. Never raises, + and never fails the baseline -- halting here would stall the session. + """ + try: + from hyperloom.orchestrator.measurement.convergence import assess_convergence + + warm = float(warmup_tput or 0.0) + measured = float(result.get("output_throughput") or 0.0) + verdict = assess_convergence([warm, measured]) + record: dict[str, Any] = verdict.to_dict() + if warm > 0 and measured > 0: + delta_pct = (measured - warm) / warm * 100.0 + record["cold_start_delta_pct"] = round(delta_pct, 2) + if delta_pct > _COLD_START_DELTA_WARN_PCT: + result.setdefault("nonfatal_warnings", []) + result["nonfatal_warnings"].append("baseline_cold_start_delta_high") + log.warning( + "baseline_executor: measured round is %.1f%% above the warm-up round " + "(%.1f -> %.1f tok/s); the server may still have been ramping, so the " + "anchor every later gain is graded against may be low", + delta_pct, warm, measured, + ) + result["baseline_convergence"] = record + except Exception: # noqa: BLE001 - observability must never break a baseline + log.debug("baseline convergence record failed", exc_info=True) + @staticmethod def _eval_failure_evidence(result: dict[str, Any]) -> tuple[bool, str]: """Detect an eval-rooted baseline failure and capture bounded evidence. @@ -3496,6 +3545,7 @@ def _persist_recipe_snapshot(manifest: dict[str, Any]) -> bool: "baseline_double_run_discarded_first", ) result["warmup_round_tput"] = warmup_tput + self._record_baseline_convergence(result, warmup_tput) # The Coordinator promotes ``subprocess_runtime_sec`` into the # explore soft-kill anchor. Explore variants restart the server, # so report round 1's full boot+client wall-clock; round 2's diff --git a/src/hyperloom/orchestrator/actions/executors/bypass_runner.py b/src/hyperloom/orchestrator/actions/executors/bypass_runner.py index 883f07c294..8de49f8244 100644 --- a/src/hyperloom/orchestrator/actions/executors/bypass_runner.py +++ b/src/hyperloom/orchestrator/actions/executors/bypass_runner.py @@ -840,7 +840,10 @@ def _server_env( """Build the server subprocess env from the materialized benchmark envs. The whole mapping is exported, so an env-only candidate is a real experiment - rather than a rerun of the baseline. + rather than a rerun of the baseline. That also carries + ``AITER_LOG_TUNED_CONFIG`` through to the server, which bypass runs need: + without it their logs have no tuned-config hit lines, and both the GEMM + demand list and the apply verdict silently lose their input. """ profiler_dirs = ( dict.fromkeys(("VLLM_TORCH_PROFILER_DIR", "SGLANG_TORCH_PROFILER_DIR", "ATOM_TORCH_PROFILER_DIR"), profile_dir) diff --git a/src/hyperloom/orchestrator/kernel/request_handlers.py b/src/hyperloom/orchestrator/kernel/request_handlers.py index 13227856b6..00b961283f 100644 --- a/src/hyperloom/orchestrator/kernel/request_handlers.py +++ b/src/hyperloom/orchestrator/kernel/request_handlers.py @@ -2882,9 +2882,48 @@ def _resolve_vllm_aiter_routing( ) flags["aiter_fused_moe"] = model_supports_aiter_ck_fused_moe(model_path, tp) + + _warn_if_moe_routing_is_coarser_than_the_log(server_log, flags) return flags +def _warn_if_moe_routing_is_coarser_than_the_log( + server_log: str, flags: dict[str, bool] +) -> None: + """Say so when one log shows both MoE backends and routing picks one. + + The decision above is a substring scan: seeing an aiter fused-MoE marker + anywhere routes the whole run to the aiter tuner family, and + ``vllm_moe_triton`` then never runs. A run can dispatch both -- aiter CK over + part of the token range and vLLM's Triton path over the rest -- and forge's + own parser records exactly that as ``impl="mixed"``. Whichever way the single + flag falls, the range served by the other backend is left untuned. + + Reported rather than acted on here: changing this routing changes which + tuners run for every aiter-served vLLM model, which is a bigger step than + the tuner-side addition that already covers the CK half. Forge adds + ``fmoe_ck`` from the same evidence, so the gap this warns about is the + Triton half. + """ + if not flags.get("aiter_fused_moe"): + return + try: + from forge_gemm_tune.evidence import parse_log_file + except ImportError: + return + try: + moe = (parse_log_file(server_log).get("dispatch") or {}).get("moe") or {} + except Exception: # noqa: BLE001 - a reporting aid must not break routing + return + if moe.get("impl") == "mixed" or moe.get("vllm_config_hit"): + log.warning( + "gemm routing: %s shows both aiter CK and vLLM Triton MoE dispatch " + "(impl=%s, stages=%s); routing sends the whole run to the aiter " + "tuner family, so the token range Triton serves goes untuned", + server_log, moe.get("impl"), moe.get("stages_seen"), + ) + + def _vllm_block_fp8_profile_capture_required( *, framework: str, @@ -4874,6 +4913,47 @@ async def run_collective_handler(payload: dict, *, session_dir: Path) -> Handler return result +# A tuner error is a diagnostic pointer, not the diagnosis: the full text lives +# in the run's own result.json and tune.log. 400 characters is enough to carry +# the argparse line or the aiter marker that says which of the two it was. +_TRACE_TUNER_ERROR_MAXLEN = 400 + +# Emitted even when null. ``kept`` is null on every row observed so far, and an +# absent key would be indistinguishable from ``false``. +_TRACE_TUNER_ALWAYS_KEYS = ("tuner", "best_micro_speedup", "kept") + + +def _trace_tuner_row(tuner: dict[str, Any]) -> dict[str, Any]: + """One per-tuner entry for the audit row, keeping why it ended as it did. + + The row used to carry only ``tuner``/``best_micro_speedup``/``kept``, which + cannot separate a tuner that crashed from one that ran and found nothing -- + the single question the audit trail exists to answer. Across one campaign 38 + of 337 tuner runs ended ``failed`` or ``empty_output`` and the trace showed + none of them; one of those was 82 runs rejected by argparse in 11 seconds + and recorded as a clean ``no_improvement`` (#1211), which stayed invisible + for three weeks because this row had nowhere to put it. + """ + error = tuner.get("error") + if isinstance(error, str) and len(error) > _TRACE_TUNER_ERROR_MAXLEN: + error = error[:_TRACE_TUNER_ERROR_MAXLEN] + "..." + row = { + "tuner": tuner.get("tuner") or tuner.get("name"), + "best_micro_speedup": tuner.get("best_micro_speedup"), + "kept": tuner.get("kept"), + "status": tuner.get("status"), + "elapsed_s": tuner.get("elapsed_s"), + "error_class": tuner.get("error_class"), + "error": error, + } + # A clean run stays as compact as before: everything added here is dropped + # when it is null, so a successful row gains only status and elapsed_s. + return { + k: v for k, v in row.items() + if k in _TRACE_TUNER_ALWAYS_KEYS or v is not None + } + + def _trace_gemm_tuning_run(result: Any, *, session_dir: Path) -> None: """Append one ``gemm_tuning.jsonl`` audit row for a GEMM-tuning run. @@ -4892,17 +4972,15 @@ def _trace_gemm_tuning_run(result: Any, *, session_dir: Path) -> None: from hyperloom.inference_optimizer.session.session_paths import gemm_tuning_steps_path engine = str(result.get("engine") or result.get("backend") or "").strip().lower() or "unknown" - tuners: list[dict[str, Any]] = [] - for t in result.get("tuners_run") or []: - if not isinstance(t, dict): - continue - tuners.append( - { - "tuner": t.get("tuner") or t.get("name"), - "best_micro_speedup": t.get("best_micro_speedup"), - "kept": t.get("kept"), - } - ) + tuners: list[dict[str, Any]] = [ + _trace_tuner_row(t) for t in (result.get("tuners_run") or []) if isinstance(t, dict) + ] + # The envelope reported no error class even when a tuner had named one, so a + # crashed run and a barren one looked alike at the top level too. Take the + # first one a tuner supplied rather than leaving the field null. + error_class = result.get("error_class") or next( + (t["error_class"] for t in tuners if t.get("error_class")), None + ) row = { "kind": "gemm_tuning", "ts": datetime.now(timezone.utc).isoformat(timespec="microseconds"), @@ -4919,7 +4997,7 @@ def _trace_gemm_tuning_run(result: Any, *, session_dir: Path) -> None: "workspace": result.get("workspace"), "requires_e2e_validation": result.get("requires_e2e_validation"), "tuners_run": tuners, - "error_class": result.get("error_class"), + "error_class": error_class, } row = {k: v for k, v in row.items() if v is not None} try: diff --git a/src/hyperloom/orchestrator/measurement/__init__.py b/src/hyperloom/orchestrator/measurement/__init__.py new file mode 100644 index 0000000000..2482e9baa2 --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Measurement-trust helpers: convergence, paired A/B, apply verification.""" diff --git a/src/hyperloom/orchestrator/measurement/apply_verification.py b/src/hyperloom/orchestrator/measurement/apply_verification.py new file mode 100644 index 0000000000..06ac91438a --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/apply_verification.py @@ -0,0 +1,180 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Check whether a tuned artifact was actually read by the server. + +Three things being true at once -- the artifact exists, the env var is set, and +e2e throughput went up -- still does not mean the tuning did anything. Two +independent ways for it to mean nothing have both been observed: + +* **the keys are unreachable**: the table has rows, but none of them match the + (M, N, K, ...) the runtime asks for; +* **the table never arrived**: the artifact was written, but the merge step did + not pick it up and the server loaded its bundled default. + +Neither is a tuner-selection problem, so no amount of choosing the right tuner +detects them. This is a separate, deterministic check on the serving log. + +The one trap it has to avoid: aiter logs a *miss* unconditionally but a *hit* +only when ``AITER_LOG_TUNED_CONFIG=1``. Reading "no hit lines" as "zero hits" +would mark every arm that ran without the flag as a failed apply -- and a scan +of 60 production logs found the flag set in none of them. So "we cannot tell" +is a distinct verdict from "it was not used", and only the latter reverts. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path + +log = logging.getLogger(__name__) + +# Verdicts that should block a KEEP. +BLOCKING_VERDICTS = frozenset({"not_merged", "zero_hit"}) + + +@dataclass(frozen=True) +class ApplyVerdict: + """Whether the tuned table reached the server and was read.""" + + verdict: str + hits: int = 0 + misses: int = 0 + merged_tables: list[str] = field(default_factory=list) + unmerged_artifacts: list[str] = field(default_factory=list) + detail: str = "" + + @property + def blocks_keep(self) -> bool: + return self.verdict in BLOCKING_VERDICTS + + @property + def conclusive(self) -> bool: + return self.verdict in {"served", "not_merged", "zero_hit"} + + def to_dict(self) -> dict[str, object]: + return { + "verdict": self.verdict, + "hits": self.hits, + "misses": self.misses, + "blocks_keep": self.blocks_keep, + "conclusive": self.conclusive, + "merged_tables": list(self.merged_tables), + "unmerged_artifacts": list(self.unmerged_artifacts), + "detail": self.detail, + } + + +def _parse(server_log: Path) -> dict | None: + """Parse the serving log with forge's evidence module, if it is installed.""" + try: + from forge_gemm_tune.evidence import parse_log_file + except ImportError: + log.info("forge_gemm_tune not importable; apply verification unavailable") + return None + try: + return parse_log_file(server_log) + except Exception: # noqa: BLE001 - a parse failure must not fail the run + log.debug("apply verification parse failed for %s", server_log, exc_info=True) + return None + + +def verify_applied( + server_log: Path | str, + artifact_paths: list[str] | None = None, + *, + hit_logging: bool | None = None, + runtime_table_names: list[str] | None = None, +) -> ApplyVerdict: + """Decide whether the tuned artifacts were merged and read. + + Args: + server_log: The serving log written by the run under test. + artifact_paths: Tuned CSVs that were supposed to be deployed. + hit_logging: Whether ``AITER_LOG_TUNED_CONFIG`` was on for this run. + ``None`` means unknown, which keeps a zero-hit result inconclusive. + runtime_table_names: Canonical table names the runtime resolves these + artifacts under (e.g. ``bf16_tuned_gemm.csv``). Needed because the + file we deploy is named after the candidate, not after the table. + + Returns: + A verdict. ``blocks_keep`` is true only for the cases that are + positively wrong; everything else, including "cannot tell", leaves the + decision to the caller. + """ + path = Path(server_log) + if not path.is_file(): + return ApplyVerdict("unknown", detail=f"no serving log at {path}") + + report = _parse(path) + if report is None: + return ApplyVerdict("unknown", detail="log parser unavailable") + + av = report.get("apply_verdict") or {} + hits = int(av.get("hit") or 0) + misses = int(av.get("miss") or 0) + merged = [str(m) for m in (report.get("merged_tables") or [])] + consulted = [str(c) for c in (report.get("consulted_tables") or [])] + + # 1. Did the artifact reach the server at all? + # + # Judge by the tables the lookups actually named, not by the merge line. + # Setting AITER_CONFIG_* -- which is exactly what a candidate run does -- + # makes aiter skip the merge step entirely: it prints no merge line and + # resolves against the override, so the lookup is the only place the path + # appears. Reading an absent merge line as "not merged" would have + # reverted every candidate. + # + # Both names are accepted because both are legitimate: the deployed file + # is named after the candidate when it is an override, and after the + # table when the server merged it into its own config directory. + wanted = [str(a) for a in (artifact_paths or []) if str(a).strip()] + if wanted and (consulted or merged): + seen = {Path(p).name for p in consulted} | {Path(m).name for m in merged} + canonical = {str(n).strip() for n in (runtime_table_names or []) if str(n).strip()} + if not (seen & (canonical or set())) : + missing = [a for a in wanted if Path(a).name not in seen] + if len(missing) == len(wanted): + return ApplyVerdict( + "not_merged", hits, misses, merged, missing, + detail=( + f"none of the {len(wanted)} tuned table(s) appear in what the " + f"runtime consulted ({sorted(seen)}); the server loaded its " + "bundled defaults" + ), + ) + + # 2. Was anything read? Hit lines are gated behind AITER_LOG_TUNED_CONFIG, + # so their absence is only informative when the flag was on. + if hits > 0: + return ApplyVerdict( + "served", hits, misses, merged, + detail=f"{hits} lookup(s) hit the tuned table", + ) + if misses > 0: + # Misses logged and no hits. With hit logging on, that is a real zero -- + # the case this gate exists for. Without it, "never read" and "not + # recorded" are the same picture, and a scan of 60 production logs found + # the flag set in none of them, so the default has to stay inconclusive. + if hit_logging: + return ApplyVerdict( + "zero_hit", hits, misses, merged, + detail=( + f"{misses} lookup(s) with hit logging on, none matched the " + "tuned table" + ), + ) + return ApplyVerdict( + "inconclusive_no_hit_logging", hits, misses, merged, + detail=( + "misses logged but hit logging was off or unknown; cannot " + "distinguish 'never read' from 'not recorded' -- set " + "AITER_LOG_TUNED_CONFIG=1" + ), + ) + + return ApplyVerdict( + "no_lookups", hits, misses, merged, + detail="the server made no tuned-config lookups at all", + ) diff --git a/src/hyperloom/orchestrator/measurement/convergence.py b/src/hyperloom/orchestrator/measurement/convergence.py new file mode 100644 index 0000000000..266880b14a --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/convergence.py @@ -0,0 +1,170 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Decide whether a throughput measurement has converged. + +A real run measured the same configuration three times: 14,202.70 -> 19,373.98 +-> 22,424.80 tok/s. A 58% spread, monotonically rising -- the measurement window +sat on the warm-up climb, not on steady state. At that noise level a 3% KEEP +threshold cannot separate anything, so every number downstream of it, including +the ones used to argue about it, is unusable. + +A controlled repeat pinned the cause. One resident vLLM server, five *identical* +benchmark passes: + + round 1: 63.90 req/s (TTFT 1906.52 ms) + round 2: 133.85 (415.12) + round 3: 139.04 (363.99) + round 4: 137.29 (368.15) + round 5: 117.13 (621.12) + +Full spread 117.6%. Drop round 1 -- whose TTFT is 5x the rest, i.e. plainly cold +start -- and rounds 2-4 span 3.9%. So the fix is not a looser threshold, which +would let the genuine climb through as well; it is to discard the warm-up round +and then require consecutive rounds to agree. + +Round 5 falling back to 117.13 is the other lesson: that box was shared, and +another workload arrived. Convergence on one side is necessary but not +sufficient -- paired alternating measurement is what removes drift between the +A and B legs, and it lives in its own module. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field + +log = logging.getLogger(__name__) + +# Same order of magnitude as the KEEP threshold: a measurement that cannot +# resolve the decision it feeds is not converged. +DEFAULT_TOLERANCE_PCT = 3.0 + +# Rounds discarded before judging. The observed cold-start round on its own took +# the spread from 3.9% to 117.6%. +DEFAULT_WARMUP_ROUNDS = 1 + +# Below this many usable rounds there is nothing to compare against. +MIN_ROUNDS_FOR_VERDICT = 2 + +# A rising pair is not a trend: with two noisy samples, half of all steady +# measurements rise. Claiming "still warming up" needs three points; with two, +# spread alone decides. +MIN_ROUNDS_FOR_TREND = 3 + + +@dataclass(frozen=True) +class ConvergenceVerdict: + """Why a series was accepted or rejected, with the numbers behind it.""" + + converged: bool + reason: str + value: float | None + used: list[float] = field(default_factory=list) + discarded: list[float] = field(default_factory=list) + spread_pct: float | None = None + monotonic: bool = False + + def to_dict(self) -> dict[str, object]: + return { + "converged": self.converged, + "reason": self.reason, + "value": self.value, + "rounds_used": list(self.used), + "rounds_discarded": list(self.discarded), + "spread_pct": self.spread_pct, + "monotonic_increasing": self.monotonic, + } + + +def _spread_pct(values: list[float]) -> float | None: + """Max-to-min spread as a percentage of the minimum.""" + usable = [v for v in values if v > 0] + if len(usable) < 2: + return None + lo, hi = min(usable), max(usable) + return (hi - lo) / lo * 100.0 + + +def _is_monotonic_increasing(values: list[float]) -> bool: + """True only for a series long enough for a rise to mean something.""" + return len(values) >= MIN_ROUNDS_FOR_TREND and all( + b > a for a, b in zip(values, values[1:], strict=False) + ) + + +def assess_convergence( + rounds: list[float], + *, + tolerance_pct: float = DEFAULT_TOLERANCE_PCT, + warmup_rounds: int = DEFAULT_WARMUP_ROUNDS, +) -> ConvergenceVerdict: + """Judge a throughput series measured under one unchanged configuration. + + Args: + rounds: Throughputs in chronological order. + tolerance_pct: Allowed spread across the retained rounds. + warmup_rounds: Leading rounds discarded before judging. + + Returns: + A verdict carrying every round, so a reader can audit the call rather + than trusting a single surviving number. + """ + series = [float(r) for r in rounds if isinstance(r, (int, float))] + positive = [r for r in series if r > 0] + if not positive: + return ConvergenceVerdict(False, "no_measurements", None, [], series) + + discarded = positive[:warmup_rounds] + used = positive[warmup_rounds:] + if len(used) < MIN_ROUNDS_FOR_VERDICT: + # One usable round cannot be shown to be steady. Saying so beats + # reporting it as if it were. + return ConvergenceVerdict( + False, "insufficient_rounds", None, used, discarded, + spread_pct=_spread_pct(used), + ) + + spread = _spread_pct(used) + monotonic = _is_monotonic_increasing(used) + + if monotonic: + # Still climbing: the last round is the least settled, so taking it + # would systematically overstate the result. + return ConvergenceVerdict( + False, "monotonic_increasing", None, used, discarded, + spread_pct=spread, monotonic=True, + ) + if spread is not None and spread > tolerance_pct: + return ConvergenceVerdict( + False, "spread_exceeds_tolerance", None, used, discarded, + spread_pct=spread, + ) + + value = sum(used) / len(used) + return ConvergenceVerdict( + True, "converged", value, used, discarded, spread_pct=spread, + ) + + +def converged_throughput( + rounds: list[float], + *, + tolerance_pct: float = DEFAULT_TOLERANCE_PCT, + warmup_rounds: int = DEFAULT_WARMUP_ROUNDS, +) -> float | None: + """The steady-state throughput, or None when the series has not settled. + + Callers must treat None as "cannot decide" and refuse the KEEP, rather than + falling back to the last round -- that fallback is what turned a warm-up + climb into a reported gain. + """ + verdict = assess_convergence( + rounds, tolerance_pct=tolerance_pct, warmup_rounds=warmup_rounds + ) + if not verdict.converged: + log.info( + "throughput not converged (%s): used=%s discarded=%s spread=%.1f%%", + verdict.reason, verdict.used, verdict.discarded, verdict.spread_pct or 0.0, + ) + return verdict.value diff --git a/src/hyperloom/orchestrator/measurement/paired.py b/src/hyperloom/orchestrator/measurement/paired.py new file mode 100644 index 0000000000..31cee1dac7 --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/paired.py @@ -0,0 +1,120 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Decide an A/B comparison from interleaved pairs rather than two blocks. + +Convergence (see :mod:`.convergence`) establishes that each side is steady. It +does not establish that the two sides were measured under the same machine +state. Measuring all of A and then all of B leaves any drift between the two +blocks -- temperature, clocks, a neighbour's workload -- indistinguishable from +the effect being measured. One controlled repeat on this fleet showed the point +directly: five identical passes against one resident server held ~137 req/s for +three rounds and then fell to 117 when another workload landed, a 16% swing that +owes nothing to the configuration. + +So pairs are measured interleaved (A, B, A, B, ...) and the verdict comes from +the *paired differences*: + +* the **median** difference, not the mean, so one disturbed pair cannot carry + the result; +* at least two pairs, because a single pair is not a comparison; +* **agreement in sign** -- if one pair says A is faster and another says B is, + the machine moved more than the change did, and the honest answer is + ``inconclusive`` rather than whichever side the average happened to land on. + +The threshold is applied to the median difference, so "B is better" means better +by more than the KEEP margin, not merely different. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from statistics import median + +log = logging.getLogger(__name__) + +# Same order as the KEEP margin: a difference the measurement cannot resolve is +# not a difference. +DEFAULT_THRESHOLD_PCT = 3.0 + +# One pair is a coincidence. +MIN_PAIRS = 2 + + +@dataclass(frozen=True) +class PairedVerdict: + """The outcome of a paired A/B comparison, with the evidence behind it.""" + + decisive: bool + reason: str + # Positive => B (the candidate) is faster than A (the baseline). + median_delta_pct: float | None + pairs: list[tuple[float, float]] = field(default_factory=list) + deltas_pct: list[float] = field(default_factory=list) + + @property + def candidate_wins(self) -> bool: + return self.decisive and self.reason == "candidate_faster" + + def to_dict(self) -> dict[str, object]: + return { + "decisive": self.decisive, + "reason": self.reason, + "median_delta_pct": self.median_delta_pct, + "candidate_wins": self.candidate_wins, + "pairs": [list(p) for p in self.pairs], + "deltas_pct": self.deltas_pct, + } + + +def interleaved_plan(n_pairs: int) -> list[str]: + """The measurement order for ``n_pairs`` pairs: A, B, A, B, ... + + Returned rather than assumed so callers cannot accidentally run blocks and + then feed the results in here, which would defeat the whole mechanism. + """ + return [side for _ in range(max(int(n_pairs), 0)) for side in ("A", "B")] + + +def assess_paired( + pairs: list[tuple[float, float]], + *, + threshold_pct: float = DEFAULT_THRESHOLD_PCT, + min_pairs: int = MIN_PAIRS, +) -> PairedVerdict: + """Judge interleaved ``(baseline, candidate)`` throughput pairs. + + Args: + pairs: One tuple per interleaved round, in measurement order. + threshold_pct: Margin the median difference must clear. + min_pairs: Pairs required before any verdict other than + ``insufficient_pairs``. + + Returns: + A verdict carrying every pair, so the call can be audited rather than + trusted. + """ + usable = [ + (float(a), float(b)) + for a, b in pairs + if isinstance(a, (int, float)) and isinstance(b, (int, float)) and a > 0 and b > 0 + ] + if len(usable) < max(int(min_pairs), 1): + return PairedVerdict(False, "insufficient_pairs", None, usable, []) + + deltas = [round((b - a) / a * 100.0, 4) for a, b in usable] + signs = {1 if d > 0 else (-1 if d < 0 else 0) for d in deltas if d != 0} + med = round(median(deltas), 4) + + if len(signs) > 1: + # The pairs disagree about which side is faster. Nothing about the + # configuration explains that; the machine moved. + log.info("paired A/B inconclusive: deltas disagree in sign %s", deltas) + return PairedVerdict(False, "sign_disagreement", med, usable, deltas) + + if med > threshold_pct: + return PairedVerdict(True, "candidate_faster", med, usable, deltas) + if med < -threshold_pct: + return PairedVerdict(True, "candidate_slower", med, usable, deltas) + return PairedVerdict(True, "within_noise", med, usable, deltas) diff --git a/src/hyperloom/orchestrator/measurement/tests/__init__.py b/src/hyperloom/orchestrator/measurement/tests/__init__.py new file mode 100644 index 0000000000..eaf7480ba3 --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/tests/__init__.py @@ -0,0 +1,2 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT diff --git a/src/hyperloom/orchestrator/measurement/tests/test_apply_verification.py b/src/hyperloom/orchestrator/measurement/tests/test_apply_verification.py new file mode 100644 index 0000000000..f17214ede4 --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/tests/test_apply_verification.py @@ -0,0 +1,274 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for tuned-artifact apply verification. + +The load-bearing distinction: aiter logs a miss unconditionally but a hit only +when AITER_LOG_TUNED_CONFIG=1. "No hit lines" therefore does not mean "zero +hits", and a check that conflates them would revert every arm that ran without +the flag -- which, in a scan of 60 production logs, was all of them. + +``forge_gemm_tune`` is not a Hyperloom dependency, so importorskip on it left +this whole module -- and therefore the KEEP gate's entire decision surface -- +without automated coverage in CI. The verdict logic is exercised against a +stand-in parser instead, and the real parser is used as well wherever forge +happens to be installed. +""" + +from __future__ import annotations + +import sys +import types + +import pytest + +from hyperloom.orchestrator.measurement import apply_verification as av +from hyperloom.orchestrator.measurement.apply_verification import verify_applied + +_MISS = ( + "[aiter] shape is M:{m}, N:4096, K:4096 dtype='torch.bfloat16' " + "otype='torch.bfloat16' bias=False, scaleAB=False, bpreshuffle=False, " + "not found tuned config in /tmp/aiter_configs/bf16_tuned_gemm.csv" +) +# Transcribed from a real MI355X run: the hit line names the table it resolved +# in, which is the only place the path appears once AITER_CONFIG_* is set. +_HIT = ( + "[aiter] shape is M:{m}, N:4096, K:4096 dtype='torch.bfloat16' " + "otype='torch.bfloat16' bias=False, scaleAB=False, bpreshuffle=False " + "found padded_M: {m}, N:4096, K:4096 is tuned on cu_num = 256 in " + "/tmp/aiter_configs/bf16_tuned_gemm.csv, libtype is asm, kernel name is knl" +) +_MERGE = ( + "[aiter] merge tuned file under model_configs/ and configs/ " + "/srv/cfg/bf16_tuned_gemm.csv:/srv/cfg/other.csv" +) +_BF16 = ["bf16_tuned_gemm.csv"] + + +def _log(tmp_path, lines): + p = tmp_path / "server.log" + p.write_text("\n".join(lines) + "\n", encoding="utf-8") + return p + + +@pytest.fixture(params=["stub_parser", "real_forge"]) +def parser(request, monkeypatch): + """Run every case against a stand-in parser, and against forge when present. + + The stand-in is deliberately minimal -- it reproduces only the three facts + the verdict depends on (hit/miss counts, merged tables, consulted tables) -- + so the decision logic stays under test on a machine that has no forge. + """ + if request.param == "real_forge": + # Skip on the submodule production actually imports, not the top-level + # package. A box can have forge_gemm_tune installed without + # ``evidence`` in it, and then the top-level check passes, the parser + # comes back None, every verdict is "unknown", and eleven cases fail on + # a developer machine for a reason that has nothing to do with them. + pytest.importorskip( + "forge_gemm_tune.evidence", reason="real parser unavailable" + ) + return None + + import re + + def _fake_parse_log_file(path): + text = __import__("pathlib").Path(path).read_text(encoding="utf-8", errors="replace") + hits = misses = 0 + merged: list[str] = [] + consulted: set[str] = set() + for line in text.splitlines(): + if "merge tuned file" in line: + merged.extend(p for p in line.split()[-1].split(":") if p) + elif "not found tuned config in" in line: + misses += 1 + consulted.add(line.split("not found tuned config in")[1].split(",")[0].strip()) + elif "found padded_M" in line: + hits += 1 + m = re.search(r"is tuned on cu_num\s*=\s*\d+\s+in\s+([^,]+)", line) + if m: + consulted.add(m.group(1).strip()) + return { + "apply_verdict": {"hit": hits, "miss": misses}, + "merged_tables": sorted(set(merged)), + "consulted_tables": sorted(consulted), + } + + fake = types.ModuleType("forge_gemm_tune") + fake_ev = types.ModuleType("forge_gemm_tune.evidence") + fake_ev.parse_log_file = _fake_parse_log_file # type: ignore[attr-defined] + fake.evidence = fake_ev # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "forge_gemm_tune", fake) + monkeypatch.setitem(sys.modules, "forge_gemm_tune.evidence", fake_ev) + return None + + +class TestServed: + def test_hits_mean_served(self, tmp_path, parser): + p = _log(tmp_path, [_MERGE, _HIT.format(m=16), _MISS.format(m=15)]) + v = verify_applied(p, ["/work/bf16_tuned_gemm.csv"], runtime_table_names=_BF16) + assert v.verdict == "served" + assert v.hits == 1 and v.misses == 1 + assert not v.blocks_keep and v.conclusive + + +class TestArtifactArrival: + def test_an_override_run_prints_no_merge_line_and_still_counts_as_arrived( + self, tmp_path, parser + ): + # Setting AITER_CONFIG_* makes aiter skip the merge step: no merge line + # at all, and the lookups name our own file. Reading that as "not + # merged" would revert every candidate, which is what the merge-list + # comparison used to do. + ours = "/work/run/merged_tuned_dense_bf16.csv" + p = _log(tmp_path, [ + _MISS.format(m=15).replace( + "/tmp/aiter_configs/bf16_tuned_gemm.csv", ours + ), + ]) + v = verify_applied(p, [ours], runtime_table_names=_BF16) + assert v.verdict != "not_merged" + + def test_the_runtime_table_name_is_accepted_too(self, tmp_path, parser): + # The deployed file is named after the candidate; the server resolves it + # under the canonical table name. Both are the same artifact. + p = _log(tmp_path, [_MERGE, _HIT.format(m=16)]) + v = verify_applied( + p, ["/work/run/merged_tuned_dense_bf16.csv"], runtime_table_names=_BF16 + ) + assert v.verdict == "served" + + def test_a_genuinely_absent_artifact_still_blocks(self, tmp_path, parser): + # Nothing the runtime touched resembles what we deployed. + p = _log(tmp_path, [_MERGE, _MISS.format(m=15)]) + v = verify_applied( + p, + ["/work/a8w8_blockscale_tuned_gemm.csv"], + runtime_table_names=["a8w8_blockscale_tuned_gemm.csv"], + ) + assert v.verdict == "not_merged" and v.blocks_keep + assert v.unmerged_artifacts == ["/work/a8w8_blockscale_tuned_gemm.csv"] + + def test_match_is_by_basename(self, tmp_path, parser): + p = _log(tmp_path, [_MERGE, _HIT.format(m=16)]) + v = verify_applied( + p, ["/some/other/dir/bf16_tuned_gemm.csv"], runtime_table_names=_BF16 + ) + assert v.verdict == "served" + + +class TestHitLoggingTrap: + def test_misses_without_hit_logging_is_not_a_failure(self, tmp_path, parser): + # No hit lines because the flag was off -- must NOT block. + p = _log(tmp_path, [_MERGE, _MISS.format(m=15), _MISS.format(m=17)]) + v = verify_applied( + p, ["/work/bf16_tuned_gemm.csv"], runtime_table_names=_BF16, + hit_logging=False, + ) + assert v.verdict == "inconclusive_no_hit_logging" + assert not v.blocks_keep + assert not v.conclusive + assert "AITER_LOG_TUNED_CONFIG" in v.detail + + def test_unknown_hit_logging_stays_inconclusive(self, tmp_path, parser): + p = _log(tmp_path, [_MERGE, _MISS.format(m=15)]) + v = verify_applied(p, ["/work/bf16_tuned_gemm.csv"], runtime_table_names=_BF16) + assert v.verdict == "inconclusive_no_hit_logging" and not v.blocks_keep + + def test_zero_hits_with_logging_on_is_a_real_failure(self, tmp_path, parser): + # This is the verdict the gate exists for, and it was unreachable: the + # parser answers "inconclusive" for hits==0 whatever the flag, so the + # branch was dead. Now that every serving run sets the flag, "0 hits and + # N misses" is a genuine zero and has to block. + p = _log(tmp_path, [_MERGE, _MISS.format(m=15), _MISS.format(m=17)]) + v = verify_applied( + p, ["/work/bf16_tuned_gemm.csv"], runtime_table_names=_BF16, + hit_logging=True, + ) + assert v.verdict == "zero_hit" + assert v.blocks_keep and v.conclusive + assert v.misses == 2 + + def test_every_blocking_verdict_is_reachable(self, tmp_path, parser): + """A verdict listed as blocking that nothing can return is not a gate.""" + reached = set() + p = _log(tmp_path, [_MERGE, _MISS.format(m=15)]) + reached.add( + verify_applied( + p, ["/work/a8w8_tuned_gemm.csv"], + runtime_table_names=["a8w8_tuned_gemm.csv"], + ).verdict + ) + reached.add( + verify_applied( + p, ["/work/bf16_tuned_gemm.csv"], runtime_table_names=_BF16, + hit_logging=True, + ).verdict + ) + assert av.BLOCKING_VERDICTS <= reached + + +class TestDegraded: + def test_missing_log(self, tmp_path, parser): + v = verify_applied(tmp_path / "nope.log", ["/work/x.csv"]) + assert v.verdict == "unknown" and not v.blocks_keep + + def test_no_lookups_at_all(self, tmp_path, parser): + p = _log(tmp_path, ["server started", "ready"]) + v = verify_applied(p, []) + assert v.verdict == "no_lookups" and not v.blocks_keep + + def test_no_artifacts_supplied_skips_the_arrival_check(self, tmp_path, parser): + p = _log(tmp_path, [_MERGE, _HIT.format(m=16)]) + assert verify_applied(p, None).verdict == "served" + + def test_to_dict_is_serialisable(self, tmp_path, parser): + p = _log(tmp_path, [_MERGE, _HIT.format(m=16)]) + d = verify_applied( + p, ["/work/bf16_tuned_gemm.csv"], runtime_table_names=_BF16 + ).to_dict() + assert d["verdict"] == "served" and d["blocks_keep"] is False + + +class TestTheEnvToTableMapDoesNotDrift: + """The same mapping exists here and in KernelForge, and cannot be shared. + + A name that drifts makes the apply check compare our deployed file against + the wrong runtime table, conclude the artifact never arrived, and revert a + candidate that was fine. The two same-repo copies are now one constant; + this covers the copy that lives in the other repository. + """ + + def test_every_env_var_maps_to_the_same_table_as_kernelforge(self): + forge_utils = pytest.importorskip( + "forge_gemm_tune.utils", reason="KernelForge not installed here" + ) + from hyperloom.orchestrator.phases.kernel import _AITER_ENV_TO_TABLE + + forge_env_vars = set(getattr(forge_utils, "TUNER_ENV_VARS", {}).values()) + aiter_only = {v for v in forge_env_vars if v.startswith("AITER_CONFIG")} + + missing = aiter_only - set(_AITER_ENV_TO_TABLE) + assert not missing, ( + f"KernelForge writes {sorted(missing)} but the apply check has no " + "table for them, so artifacts under those names read as never " + "having arrived" + ) + + def test_the_fp4_key_is_the_one_aiter_actually_reads(self): + # AITER_CONFIG_GEMM_A4W4, not the "_BLOCKSCALE" variant. The suffixed + # name was a dead key that silently dropped every tuned fp4 GEMM. + from hyperloom.orchestrator.phases.kernel import _AITER_ENV_TO_TABLE + + assert "AITER_CONFIG_GEMM_A4W4" in _AITER_ENV_TO_TABLE + assert "AITER_CONFIG_GEMM_A4W4_BLOCKSCALE" not in _AITER_ENV_TO_TABLE + + def test_the_merge_step_and_the_apply_check_read_one_map(self): + # They were separate copies until one was almost edited alone. + import inspect + + from hyperloom.orchestrator.phases import kernel + + src = inspect.getsource(kernel.KernelPhase._merge_gemm_candidate_with_runtime) + assert "_AITER_ENV_TO_TABLE" in src + assert "a8w8_blockscale_tuned_gemm.csv" not in src diff --git a/src/hyperloom/orchestrator/measurement/tests/test_convergence.py b/src/hyperloom/orchestrator/measurement/tests/test_convergence.py new file mode 100644 index 0000000000..bce9b0913c --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/tests/test_convergence.py @@ -0,0 +1,124 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for the throughput convergence judge. + +The numbers below are the real ones. ``_RCA_ROUNDS`` is the three-round series +that started this (58% spread, monotonically rising). ``_T6_ROUNDS`` is the +five-pass controlled repeat that showed the spread is cold start, not noise: +117.6% across all five, 3.9% once the first is dropped. +""" + +from __future__ import annotations + +from hyperloom.orchestrator.measurement.convergence import ( + assess_convergence, + converged_throughput, +) + +# warmup / measure / accuracy from a real session. +_RCA_ROUNDS = [14202.70, 19373.98, 22424.80] + +# Five identical benchmark passes against one resident server (req/s). +_T6_ROUNDS = [63.90, 133.85, 139.04, 137.29, 117.13] + + +class TestTheSeriesThatStartedThis: + def test_rca_series_is_not_converged(self): + verdict = assess_convergence(_RCA_ROUNDS) + assert verdict.converged is False + assert verdict.value is None + + def test_rca_series_is_rejected_on_spread(self): + # Dropping the warm-up round leaves 19373.98 -> 22424.80: a 15.8% gap, + # five times the tolerance. Taking the last round here is what turned a + # warm-up climb into a reported "gain". + verdict = assess_convergence(_RCA_ROUNDS) + assert verdict.reason == "spread_exceeds_tolerance" + assert verdict.spread_pct > 15.0 + + def test_two_rounds_are_not_called_a_trend(self): + # A rising pair is not evidence of warm-up: half of all steady pairs + # rise. The trend verdict needs three points. + assert assess_convergence(_RCA_ROUNDS).monotonic is False + + def test_warmup_round_is_discarded_not_averaged(self): + verdict = assess_convergence(_RCA_ROUNDS) + assert verdict.discarded == [14202.70] + assert 14202.70 not in verdict.used + + +class TestT6ControlledRepeat: + def test_all_five_rounds_would_look_catastrophic(self): + # Keeping the cold round: >100% spread. + verdict = assess_convergence(_T6_ROUNDS, warmup_rounds=0) + assert verdict.converged is False + assert verdict.spread_pct > 100.0 + + def test_dropping_the_cold_round_still_fails_on_the_shared_box(self): + # Rounds 2-5 include the round-5 dip caused by another workload landing + # on the shared machine, so this is correctly NOT converged -- the fix + # for that is paired measurement, not a looser threshold. + verdict = assess_convergence(_T6_ROUNDS) + assert verdict.converged is False + assert verdict.reason == "spread_exceeds_tolerance" + + def test_rounds_two_to_four_are_converged(self): + # The steady window: 3.9% spread, inside the 3%-order tolerance once the + # dip is excluded. + verdict = assess_convergence(_T6_ROUNDS[1:4], warmup_rounds=0, tolerance_pct=5.0) + assert verdict.converged is True + assert verdict.spread_pct < 4.0 + assert verdict.value == sum(_T6_ROUNDS[1:4]) / 3 + + +class TestVerdicts: + def test_steady_series_converges_to_the_mean_of_used_rounds(self): + verdict = assess_convergence([50.0, 100.0, 101.0, 100.5]) + assert verdict.converged is True + assert verdict.discarded == [50.0] + assert verdict.value == (100.0 + 101.0 + 100.5) / 3 + + def test_single_usable_round_cannot_be_judged(self): + verdict = assess_convergence([50.0, 100.0]) + assert verdict.converged is False + assert verdict.reason == "insufficient_rounds" + + def test_empty_series(self): + verdict = assess_convergence([]) + assert verdict.converged is False and verdict.reason == "no_measurements" + + def test_non_positive_rounds_are_ignored(self): + verdict = assess_convergence([0.0, -1.0, 100.0, 100.5, 100.2]) + assert verdict.converged is True + assert 0.0 not in verdict.used and -1.0 not in verdict.used + + def test_tight_but_still_climbing_series_is_rejected(self): + # Where the trend rule earns its keep: the spread is inside tolerance, + # so only the monotonic check catches that this is still warming up. + verdict = assess_convergence([50.0, 100.0, 101.0, 102.0]) + assert verdict.converged is False + assert verdict.reason == "monotonic_increasing" + assert verdict.spread_pct < 3.0 + + def test_decreasing_series_is_allowed_when_tight(self): + # Only *increasing* series indicate an unfinished warm-up; a tight + # decreasing one is just noise around steady state. + verdict = assess_convergence([50.0, 100.5, 100.2, 100.0]) + assert verdict.converged is True + + def test_verdict_reports_every_round_for_audit(self): + d = assess_convergence(_RCA_ROUNDS).to_dict() + assert d["rounds_discarded"] == [14202.70] + assert d["rounds_used"] == [19373.98, 22424.80] + assert d["converged"] is False + assert d["spread_pct"] > 15.0 + + +class TestConvergedThroughput: + def test_returns_none_rather_than_the_last_round(self): + # The whole point: no number is better than the climbing one. + assert converged_throughput(_RCA_ROUNDS) is None + + def test_returns_value_when_settled(self): + assert converged_throughput([50.0, 100.0, 100.5]) == 100.25 diff --git a/src/hyperloom/orchestrator/measurement/tests/test_paired.py b/src/hyperloom/orchestrator/measurement/tests/test_paired.py new file mode 100644 index 0000000000..76fe8ea00f --- /dev/null +++ b/src/hyperloom/orchestrator/measurement/tests/test_paired.py @@ -0,0 +1,82 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for the paired A/B judge.""" + +from __future__ import annotations + +from hyperloom.orchestrator.measurement.paired import ( + assess_paired, + interleaved_plan, +) + + +class TestPlan: + def test_order_alternates(self): + assert interleaved_plan(3) == ["A", "B", "A", "B", "A", "B"] + + def test_zero_and_negative_are_empty(self): + assert interleaved_plan(0) == [] and interleaved_plan(-1) == [] + + +class TestVerdicts: + def test_consistent_win_is_decisive(self): + v = assess_paired([(100.0, 110.0), (102.0, 112.0), (99.0, 109.0)]) + assert v.decisive and v.candidate_wins + assert v.reason == "candidate_faster" + assert v.median_delta_pct > 3.0 + + def test_consistent_loss_is_decisive(self): + v = assess_paired([(100.0, 90.0), (101.0, 91.0)]) + assert v.decisive and v.reason == "candidate_slower" + assert not v.candidate_wins + + def test_small_consistent_difference_is_within_noise(self): + v = assess_paired([(100.0, 101.0), (100.0, 101.5)]) + assert v.decisive and v.reason == "within_noise" + assert not v.candidate_wins + + def test_sign_disagreement_is_inconclusive(self): + # One pair says the candidate is faster, another says slower. The + # machine moved more than the change did; averaging would invent a + # winner out of that. + v = assess_paired([(100.0, 115.0), (100.0, 88.0)]) + assert not v.decisive and v.reason == "sign_disagreement" + assert not v.candidate_wins + + def test_single_pair_is_not_a_comparison(self): + v = assess_paired([(100.0, 130.0)]) + assert not v.decisive and v.reason == "insufficient_pairs" + + def test_empty(self): + assert assess_paired([]).reason == "insufficient_pairs" + + +class TestMedianNotMean: + def test_one_disturbed_pair_does_not_carry_the_result(self): + # Three pairs agree on ~+4%; a fourth is wrecked by a neighbour landing + # on the box. The mean would be dragged far off; the median holds. + pairs = [(100.0, 104.0), (100.0, 104.5), (100.0, 103.5), (100.0, 160.0)] + v = assess_paired(pairs) + assert v.decisive and v.candidate_wins + assert 3.5 < v.median_delta_pct < 6.0 + + def test_deltas_are_reported_for_audit(self): + v = assess_paired([(100.0, 110.0), (100.0, 120.0)]) + assert v.deltas_pct == [10.0, 20.0] + assert v.to_dict()["pairs"] == [[100.0, 110.0], [100.0, 120.0]] + + +class TestBadInput: + def test_non_positive_values_are_dropped(self): + v = assess_paired([(0.0, 110.0), (100.0, 110.0), (100.0, 111.0)]) + assert len(v.pairs) == 2 and v.decisive + + def test_dropping_leaves_too_few(self): + v = assess_paired([(0.0, 110.0), (100.0, -1.0)]) + assert v.reason == "insufficient_pairs" + + def test_threshold_is_configurable(self): + pairs = [(100.0, 104.0), (100.0, 104.5)] + assert assess_paired(pairs).reason == "candidate_faster" + assert assess_paired(pairs, threshold_pct=10.0).reason == "within_noise" diff --git a/src/hyperloom/orchestrator/phases/kernel.py b/src/hyperloom/orchestrator/phases/kernel.py index 1f734f3113..707406a598 100644 --- a/src/hyperloom/orchestrator/phases/kernel.py +++ b/src/hyperloom/orchestrator/phases/kernel.py @@ -47,6 +47,53 @@ # on a developer box that happens to have the real checkout mounted. _CONTAINER_AITER_CONFIG_DIR = Path("/sgl-workspace/aiter/aiter/configs") +# Idempotency key of the same-harness GEAK rebench enqueued by +# ``_enqueue_internal_stack_rebench``. Doubles as the placeholder that reserves +# ``geak_pending`` before the task row exists, so the phase guard already sees a +# pending revalidation while the enqueue is in flight. +_GEAK_REVALIDATE_IDEMPOTENCY_KEY = "geak-revalidate" + +# Which table each aiter config env var is resolved under at serving time. Two +# callers need it: the merge step, which has to find the runtime table to merge +# our candidate into, and the apply check, which has to recognise our artifact +# in the runtime's own lookup lines (the deployed file carries the candidate's +# name, not the table's). They were separate copies until one of them was +# almost edited alone -- and a name that drifts reads as "the artifact never +# arrived", which reverts a candidate that was fine. +# +# A third copy lives in KernelForge's TUNER_ENV_VARS and cannot be shared +# across repositories; ``test_aiter_env_table_matches_kernelforge`` asserts the +# two agree wherever forge is importable. +# +# Note AITER_CONFIG_GEMM_A4W4, not the "_BLOCKSCALE" variant: aiter reads +# fp4/mxfp4 (gfx950-only) configs under that name (jit/core.py), and the +# suffixed key was a dead one that silently dropped every tuned fp4 GEMM. +_AITER_ENV_TO_TABLE: dict[str, str] = { + "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE": "a8w8_blockscale_bpreshuffle_tuned_gemm.csv", + "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE": "a8w8_blockscale_tuned_gemm.csv", + "AITER_CONFIG_GEMM_A8W8_BPRESHUFFLE": "a8w8_bpreshuffle_tuned_gemm.csv", + "AITER_CONFIG_GEMM_A8W8": "a8w8_tuned_gemm.csv", + "AITER_CONFIG_GEMM_A4W4": "a4w4_blockscale_tuned_gemm.csv", + "AITER_CONFIG_GEMM_BF16": "bf16_tuned_gemm.csv", + "AITER_CONFIG_FMOE": "tuned_fmoe.csv", +} + + +def _paired_measurement_basis(verdict: Any) -> str: + """How the promoted gain was measured, so the ledger cannot overstate it. + + A gain from ``base_tput`` (measured earlier) against ``new_tput`` (measured + now) is a comparison of two *blocks*, and drift between them is folded into + the result. Recording that distinction is what lets a reader tell a + confirmed number from a plausible one; without it both arrive as + ``e2e_rebench`` and look equally solid. + """ + if verdict is None: + return "e2e_rebench_unpaired" + if getattr(verdict, "candidate_wins", False): + return "e2e_paired" + return f"e2e_paired_{getattr(verdict, 'reason', 'unknown')}" + def _collective_comm_share(state: Any) -> tuple[float | None, str]: """Return the communication share gating the lane, and its provenance. @@ -1584,6 +1631,140 @@ def _gemm_tuned_config_coverage( report["not_applied_reason"] = "no_shape_key_matched" return report + async def _confirm_gemm_gain_paired( + self, + stacked_envs: dict[str, str], + *, + baseline_tput: float, + budget_minutes: int, + extra_server_args: str = "", + ): + """Re-measure baseline and tuned stack interleaved, and judge the pairs. + + ``running_tput`` is compared against a ``baseline_tput`` measured earlier + in the session, so any drift between the two -- clocks, temperature, a + neighbour's workload -- is indistinguishable from the tuning. One + controlled repeat on this fleet moved 16% with nothing changed, and three + rounds of one unchanged configuration spanned 58%. + + Interleaving is the only thing that separates them, and it costs two + extra benchmark rounds per pair, so it is opt-in via + ``HYPERLOOM_GEMM_PAIRED_PAIRS``. When it does not run the gain is still + promoted -- it is the best number available -- but it is *labelled* as an + unpaired block comparison rather than passed off as a paired one. + """ + from ..kernel.request_handlers import integrate_handler + from ..measurement.paired import assess_paired, interleaved_plan + + try: + n_pairs = int(os.environ.get("HYPERLOOM_GEMM_PAIRED_PAIRS", "0") or 0) + except ValueError: + n_pairs = 0 + if n_pairs <= 0 or not stacked_envs or baseline_tput <= 0: + return None + + pairs: list[tuple[float, float]] = [] + pending: float | None = None + for idx, side in enumerate(interleaved_plan(n_pairs)): + envs = {} if side == "A" else dict(stacked_envs) + # The B leg has to be served the same way the KEEP was: fmoe_ck only + # takes effect under --moe-runner-backend aiter, and without it the + # tuned table is never read, so B measures the same thing as A and + # the confirmation reports within_noise for a gain that is real. + side_args = extra_server_args if side == "B" else "" + try: + res = await integrate_handler( + { + "task_id": f"gemm_paired_{side}{idx}", + "kernel_id": f"gemm_paired_{side}{idx}", + "source": "forge_gemm_paired", + "base_tput": baseline_tput, + "extra_server_args": side_args, + "extra_envs": envs, + # Measure, do not decide: the verdict comes from the + # pairs, so a per-round KEEP/REVERT here would be noise + # promoted to a decision. + "keep_threshold_pct": 100.0, + "budget_minutes": budget_minutes, + }, + session_dir=self.session_dir, + ) + except Exception as exc: # noqa: BLE001 + log.warning("forge gemm paired confirmation aborted at %s%d: %s", side, idx, exc) + break + tput = float(res.get("new_tput") or 0.0) + if tput <= 0: + log.warning("forge gemm paired confirmation: %s%d produced no throughput", side, idx) + break + if side == "A": + pending = tput + elif pending is not None: + pairs.append((pending, tput)) + pending = None + + verdict = assess_paired(pairs) + log.info( + "forge gemm paired confirmation: %d pair(s) -> %s (median delta %s%%)", + len(pairs), verdict.reason, verdict.median_delta_pct, + ) + return verdict + + def _gemm_apply_verdict( + self, + tuner_name: str, + envs: dict[str, str], + ) -> dict[str, Any] | None: + """Did the tuned table reach the server's merge list and get read? + + Complements ``_gemm_tuned_config_coverage``, which replays the shape + lookup against the CSV we wrote. That answers "could this table have + served the requests"; it cannot see the case where the table never + arrived and the server loaded its bundled default instead, because the + CSV on our disk still contains the right rows either way. + """ + from ..measurement.apply_verification import verify_applied + + csv_paths = [value for key, value in envs.items() if key.startswith("AITER_CONFIG")] + 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, + ) + 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. + log.warning( + "forge gemm E2E: no server.log under %s; apply verification " + "cannot run for %s", run_dir, tuner_name, + ) + return None + + # The deployed file is named after the candidate, so the runtime's own + # table name has to travel with it or the arrival check compares + # merged_tuned_dense_bf16.csv against bf16_tuned_gemm.csv and concludes + # the artifact never landed. + table_names = [ + name for key in envs + if (name := _AITER_ENV_TO_TABLE.get(key)) + ] + # aiter prints a hit line only under this flag; every serving run now + # sets it by default, but an operator value in the candidate env wins, + # and then a zero-hit result means nothing. + raw_flag = str(envs.get("AITER_LOG_TUNED_CONFIG", "1")).strip().lower() + hit_logging = raw_flag not in ("", "0", "false", "no", "off") + + try: + return verify_applied( + logs[-1], csv_paths, + hit_logging=hit_logging, + runtime_table_names=table_names, + ).to_dict() + except Exception: # noqa: BLE001 - verification must never fail the run + log.warning("apply verification failed for %s", tuner_name, exc_info=True) + return None + def _merge_gemm_candidate_with_runtime( self, env_var: str, candidate_csv_path: str ) -> str | None: @@ -1629,19 +1810,7 @@ def _read_header(path: Path) -> list[str]: if not candidate_path.is_file(): return None - env_var_to_tuned_name = { - "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE": "a8w8_blockscale_bpreshuffle_tuned_gemm.csv", - "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE": "a8w8_blockscale_tuned_gemm.csv", - "AITER_CONFIG_GEMM_A8W8_BPRESHUFFLE": "a8w8_bpreshuffle_tuned_gemm.csv", - "AITER_CONFIG_GEMM_A8W8": "a8w8_tuned_gemm.csv", - # aiter reads fp4/mxfp4 (gfx950-only) configs via AITER_CONFIG_GEMM_A4W4, - # not the "_BLOCKSCALE" variant (aiter jit/core.py). Must match KernelForge's - # TUNER_ENV_VARS or tuned fp4 GEMM CSVs are silently ignored at serving. - "AITER_CONFIG_GEMM_A4W4": "a4w4_blockscale_tuned_gemm.csv", - "AITER_CONFIG_GEMM_BF16": "bf16_tuned_gemm.csv", - "AITER_CONFIG_FMOE": "tuned_fmoe.csv", - } - runtime_filename = env_var_to_tuned_name.get(env_var) + runtime_filename = _AITER_ENV_TO_TABLE.get(env_var) if not runtime_filename: return None tuned_stem = Path(runtime_filename).stem @@ -2092,9 +2261,23 @@ def _gemm_e2e_candidates(self, result: dict[str, Any]) -> list[dict[str, Any]]: for t in result.get("tuners_run") or []: if not isinstance(t, dict): continue - if t.get("status") != "ok": + # partial_output is a real artifact: the tuner wrote fewer rows than + # shapes it was given (the grouped batch budget ran out), but the + # rows it did write are deployable. + if t.get("status") not in ("ok", "partial_output"): continue - if not bool(t.get("candidate")) and int(t.get("improved_shapes") or 0) <= 0: + # improved_shapes can never exceed 0 for tuners with no comparable + # baseline -- TunableOp never times the untuned dispatch, the + # candidate-CSV fallback has no per-shape Pre/Post table, and a + # hipblaslt-only bf16 run has no torch candidate to measure against. + # They report unverified_shapes instead, so gating on improved_shapes + # alone would drop exactly the artifacts that need e2e to say + # anything at all about them. + if ( + not bool(t.get("candidate")) + and int(t.get("improved_shapes") or 0) <= 0 + and int(t.get("unverified_shapes") or 0) <= 0 + ): continue env_var = str(t.get("env_var") or "").strip() env_value = str(t.get("env_value") or "").strip() @@ -2326,10 +2509,22 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: gain_pct, ) + # Two independent ways the artifact can fail to take effect, neither + # of which the throughput delta can see: the keys are unreachable + # (coverage), and the table never reached the server (apply verdict). + # Both are positive findings, not absences of evidence -- so they + # block the KEEP rather than merely annotating it. Crediting a gain + # here would attribute run-to-run drift to tuning that provably did + # not run. + apply_blockers: list[str] = [] + coverage = self._gemm_tuned_config_coverage(tuner_name, env) if coverage is not None: cand = {**cand, "tuned_config_coverage": coverage} if not coverage.get("artifact_applied"): + apply_blockers.append( + str(coverage.get("not_applied_reason") or "no_shape_key_matched") + ) log.error( "gemm E2E: tuner=%s produced an artifact the runtime never " "applied — 0 of %d requested shape(s) resolve to a tuned row; " @@ -2347,7 +2542,29 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: coverage.get("requested") or 0, ) - if decision == "KEEP" and new_tput > running_tput: + applied = self._gemm_apply_verdict(tuner_name, env) + if applied is not None: + cand = {**cand, "apply_verdict": applied} + if applied.get("blocks_keep"): + apply_blockers.append(str(applied.get("verdict") or "not_applied")) + log.error( + "forge gemm E2E: tuner=%s apply verdict=%s — %s", + tuner_name, + applied.get("verdict"), + applied.get("detail"), + ) + elif not applied.get("conclusive"): + # "Cannot tell" is not "did not apply": hit lines need + # AITER_LOG_TUNED_CONFIG=1, and treating their absence as a + # failure would revert every arm that ran without it. + log.info( + "forge gemm E2E: tuner=%s apply verdict=%s (not conclusive) — %s", + tuner_name, + applied.get("verdict"), + applied.get("detail"), + ) + + if decision == "KEEP" and new_tput > running_tput and not apply_blockers: stacked_envs.update(env) running_tput = new_tput kept.append( @@ -2386,19 +2603,40 @@ async def _validate_gemm_tuning_e2e(self, result: dict[str, Any]) -> None: ) else: reason = f"decision={decision}, gain={gain_pct:.2f}%" - if coverage is not None and not coverage.get("artifact_applied"): + if apply_blockers: # Distinguish "the tuning did not pay off" from "the tuned # artifact was never reachable", which is a wiring defect. - reason = f"tuned_config_never_applied ({reason})" + # The second is worth reporting even when the run also + # happened to measure a gain -- especially then. + reason = f"tuned_config_never_applied[{'+'.join(apply_blockers)}] ({reason})" reverted.append({**cand, "reason": reason}) # The watermark covers the whole run, so it waits for the last KEEP. if kept: total_gain = (running_tput - baseline_tput) / baseline_tput * 100.0 if baseline_tput > 0 else 0.0 + # One end-to-end measurement is not enough on this fleet: three + # rounds of a single unchanged configuration spanned 58%. Re-run + # the baseline interleaved with the tuned stack so drift shows up + # as drift. Opt-in, and when it does not run the gain is still + # promoted -- it is the best number available -- but labelled as an + # unpaired block comparison rather than passed off as a paired one. + paired = await self._confirm_gemm_gain_paired( + stacked_envs, + baseline_tput=baseline_tput, + budget_minutes=per_tuner_budget_minutes, + extra_server_args=( + "--moe-runner-backend aiter" + if "AITER_CONFIG_FMOE" in stacked_envs + else "" + ), + ) + if paired is not None: + result["paired_confirmation"] = paired.to_dict() if baseline_tput > 0: self._update_cumulative_gain_validated( running_tput, source="forge_gemm_tuning_e2e", + measurement_basis=_paired_measurement_basis(paired), ) log.info( "gemm E2E: %d tuners KEEP (total gain=+%.2f%%), %d REVERT",