From f7b4dad24598de5da764ca675eef5ccc8471d063 Mon Sep 17 00:00:00 2001 From: lishuoshuo-amd Date: Mon, 3 Aug 2026 18:06:54 +0800 Subject: [PATCH 1/3] fix(geak): gate 2b revalidation promote on material candidate to reject same-config rebench noise --- .../tests/test_geak_gain_alignment.py | 98 +++++++++++++++++++ .../orchestrator/loop/coordinator_helpers.py | 60 ++++++++++++ src/hyperloom/orchestrator/loop/writeback.py | 49 +++++++++- 3 files changed, 202 insertions(+), 5 deletions(-) diff --git a/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py b/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py index 0021ea977d..4858c9ab36 100644 --- a/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py +++ b/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py @@ -585,3 +585,101 @@ def test_report_shows_pending_candidate_excluded_from_headline() -> None: assert "AWAITING" in facts and "13.79" in facts or "13.8" in facts assert "Validated cumulative gain" not in facts assert "audit-only" in warns and "not been" in warns.lower() or "NOT" in warns + + +# ── 2b material guard: same-config rebench noise must not stamp kernel gain ─── + + +@pytest.mark.asyncio +async def test_2b_no_material_candidate_does_not_promote(tmp_path: Path) -> None: + """GEAK returned no kernel/head/overlay/patch AND its accepted_config equals + the pre-KERNEL current_best (pure passthrough). A rebench that beats + current_best by measurement noise must NOT be recorded as a kernel gain.""" + base, current_best, measured = 8668.5946, 8900.0, 9025.191 + coord = _coord(tmp_path, baseline=base, best_tput=current_best) + coord.shared_state.current_best["extra_server_args"] = "--max-num-batched-tokens 24576" + coord.shared_state.current_best["extra_envs"] = {"VLLM_ROCM_USE_AITER": "1"} + coord.shared_state.optimization_stack = [ + {"action": "explore", "variant_name": "kv-cache-fp8", "tput": current_best} + ] + coord.shared_state.resume_pending_revalidation = True + coord.shared_state.geak_pending = {"status": "awaiting_rebench"} + # geak_result is non-empty but ships NO material product; accepted_config is + # the pre-KERNEL current_best config verbatim (passthrough, zero delta). + coord.shared_state.geak_result = { + "status": "ok", + "accepted_config": {"flags": "--max-num-batched-tokens 24576", "env": "VLLM_ROCM_USE_AITER=1"}, + "accepted_kernels": [], + "accepted_heads": [], + "final_overlay": "", + "final_patch": "", + } + + async def _must_not_fallback(**_kwargs): + raise AssertionError("2a fallback must not run for a no-material drop") + + coord._validate_geak_via_geak_harness = _must_not_fallback # type: ignore[assignment] + + result = { + "output_throughput": measured, + "best_variant": {"fingerprint": "abc"}, + "winners": [], + } + await coord._promote_to_shared_state("explore", result, task=_revalidate_task(expected_hash="abc")) + + ss = coord.shared_state + assert ss.current_best["tput"] == pytest.approx(current_best) + assert ss.cumulative_gain_validated == pytest.approx(0.0) + assert ss.cumulative_gain_provenance != "geak_orch_harness_validated" + assert not any(e.get("action") == "geak_e2e" for e in ss.optimization_stack) + assert ss.resume_pending_revalidation is False + assert not ss.geak_pending + + +@pytest.mark.asyncio +async def test_2b_config_delta_candidate_still_promotes(tmp_path: Path) -> None: + """GEAK shipped no overlay/patch/kernel list, but its accepted_config adds a + new flag vs the pre-KERNEL current_best (a kernel enabled via a config + switch). That is a real GEAK product and must still promote.""" + base, current_best, measured = 8668.5946, 8900.0, 9600.0 + coord = _coord(tmp_path, baseline=base, best_tput=current_best) + coord.shared_state.current_best["extra_server_args"] = "--max-num-batched-tokens 24576" + coord.shared_state.current_best["extra_envs"] = {"VLLM_ROCM_USE_AITER": "1"} + coord.shared_state.optimization_stack = [ + {"action": "explore", "variant_name": "kv-cache-fp8", "tput": current_best} + ] + coord.shared_state.resume_pending_revalidation = True + # accepted_config adds VLLM_ROCM_USE_AITER_FP4_ASM_GEMM=1 (a new kernel switch). + result_blob = { + "status": "ok", + "accepted_config": { + "flags": "--max-num-batched-tokens 24576", + "env": "VLLM_ROCM_USE_AITER=1 VLLM_ROCM_USE_AITER_FP4_ASM_GEMM=1", + }, + "accepted_kernels": [], + "accepted_heads": [], + "final_overlay": "", + "final_patch": "", + } + coord.shared_state.geak_result = result_blob + coord._record_geak_candidate(result_blob) + + async def _must_not_fallback(**_kwargs): + raise AssertionError("2a fallback must not run when 2b validates a real delta") + + coord._validate_geak_via_geak_harness = _must_not_fallback # type: ignore[assignment] + + result = { + "output_throughput": measured, + "best_variant": {"fingerprint": "abc"}, + "winners": [], + } + await coord._promote_to_shared_state("explore", result, task=_revalidate_task(expected_hash="abc")) + + ss = coord.shared_state + expected_pct = (measured - base) / base * 100.0 + assert ss.current_best["tput"] == pytest.approx(measured) + assert ss.cumulative_gain_validated == pytest.approx(expected_pct) + assert ss.cumulative_gain_provenance == "geak_orch_harness_validated" + assert any(e.get("action") == "geak_e2e" for e in ss.optimization_stack) + assert not ss.geak_pending diff --git a/src/hyperloom/orchestrator/loop/coordinator_helpers.py b/src/hyperloom/orchestrator/loop/coordinator_helpers.py index 53d1067039..db4fab082c 100644 --- a/src/hyperloom/orchestrator/loop/coordinator_helpers.py +++ b/src/hyperloom/orchestrator/loop/coordinator_helpers.py @@ -625,6 +625,66 @@ def _geak_revalidation_decision( return "validated" +def _geak_result_has_material( + result: Any, + *, + prev_best_flags: str = "", + prev_best_envs: Any = None, +) -> bool: + """Decide whether a GEAK result carries a material optimization product. + + Guards the 2b promote path against pure passthrough noise: when GEAK ships + nothing (empty result) OR ships no kernel/head/overlay/patch AND echoes the + pre-KERNEL current_best config back unchanged, a rebench that beats + current_best is measurement variance, not a kernel gain. + + Material means ANY of: + * ``accepted_kernels`` non-empty (kernel rewrites); + * ``accepted_heads`` non-empty (attention-head optimizations); + * ``final_overlay`` non-empty (authored-kernel overlay dir); + * ``final_patch`` non-empty (source patch); + * ``accepted_config`` differs from the pre-KERNEL current_best config + (a kernel enabled via a config switch, e.g. an ASM-GEMM env flag). + + An empty/absent result returns ``True`` (cannot judge, so do not block the + native resume revalidation path). + + Args: + result: The normalized GEAK ``geak_result`` blob. + prev_best_flags: Pre-KERNEL current_best ``extra_server_args``. + prev_best_envs: Pre-KERNEL current_best ``extra_envs`` mapping. + + Returns: + ``True`` when a material product exists (or cannot be judged); else + ``False``. + """ + from hyperloom.orchestrator.actions.executors._canonical_fingerprint import ( + canonical_fingerprint, + ) + + if not isinstance(result, dict) or not result: + return True + if result.get("accepted_kernels"): + return True + if result.get("accepted_heads"): + return True + if str(result.get("final_overlay") or "").strip(): + return True + if str(result.get("final_patch") or "").strip(): + return True + accepted_cfg = result.get("accepted_config") or {} + accepted_flags = str(accepted_cfg.get("flags") or "").strip() + parsed_envs, extra_flags = _split_env_and_flags(str(accepted_cfg.get("env") or "")) + if extra_flags: + accepted_flags = (accepted_flags + " " + extra_flags).strip() + got_fp = canonical_fingerprint(accepted_flags, parsed_envs) + prev_fp = canonical_fingerprint( + str(prev_best_flags or ""), + dict(prev_best_envs or {}), + ) + return got_fp != prev_fp + + def _normalize_geak_overlay_dir(overlay: str) -> str: """Normalize a GEAK ``final_overlay`` path to the loadable overlay dir. diff --git a/src/hyperloom/orchestrator/loop/writeback.py b/src/hyperloom/orchestrator/loop/writeback.py index 72b0aca956..8905d1010e 100644 --- a/src/hyperloom/orchestrator/loop/writeback.py +++ b/src/hyperloom/orchestrator/loop/writeback.py @@ -31,6 +31,7 @@ _dedupe_extra_server_args, _merge_cumulative_extra_server_args, _parse_baseline_workload_extra, + _geak_result_has_material, _geak_revalidation_decision, _geak_sweep_measured_tput, _normalize_geak_overlay_dir, @@ -2811,20 +2812,58 @@ async def _promote_explore( min_engaged_gain_pct=_MIN_KERNEL_ENGAGED_GAIN_PCT, current_best=cb_tput, ) + ps = ( + self.shared_state.geak_result + if isinstance(getattr(self.shared_state, "geak_result", None), dict) + else {} + ) + # A rebench that beats current_best is only a KERNEL gain when + # GEAK actually produced something. Without a material product + # (kernel/head/overlay/patch or a config delta vs the pre-KERNEL + # best) the win is same-config measurement noise; drop it. + if decision == "validated" and not _geak_result_has_material( + ps, + prev_best_flags=str(cb_now.get("extra_server_args") or ""), + prev_best_envs=cb_now.get("extra_envs") or {}, + ): + decision = "no_material" if decision == "validated": # Write the headline from the measured orchestrator-harness # rebench: lift current_best + optimization_stack + the # validated gain and clear geak_pending. - ps = ( - self.shared_state.geak_result - if isinstance(getattr(self.shared_state, "geak_result", None), dict) - else {} - ) self._promote_geak_from_candidate( ps, measured_tput=float(measured), provenance="geak_orch_harness_validated", ) + elif decision == "no_material": + # No material GEAK product; the rebench beating current_best + # is same-config measurement noise. Do not touch the + # headline / stack / gain; record + clear the candidate. + log.info( + "geak 2b rebench beat current_best but GEAK shipped no " + "material product (measured=%r current_best=%r) -> " + "no_material drop", + measured, + cb_tput, + ) + try: + await self._record_observation( + "coordinator", + "observation", + { + "kind": "geak_no_material", + "measured_tput": float(measured), + "current_best_tput": ( + float(cb_tput) if isinstance(cb_tput, (int, float)) else None + ), + "baseline_tput": float(self.shared_state.baseline_tput or 0.0), + }, + ) + except Exception: # noqa: BLE001 - observation is best-effort + log.exception("geak no_material: observation emit failed") + self.shared_state.geak_pending = {} + self.shared_state.resume_pending_revalidation = False elif decision == "no_promote": # Well-measured + engaged over baseline, but does not beat # current_best. This is a real result, NOT inconclusive, so From 1bfb0c0ea54d9e33d1a40ecdbcfbcfb99e0ad186 Mon Sep 17 00:00:00 2001 From: lishuoshuo-amd Date: Mon, 3 Aug 2026 18:34:26 +0800 Subject: [PATCH 2/3] fix(geak): treat empty geak_result as no-material at 2b unless a prior geak_e2e exists, and reject provisional journey KEEP on drop --- .../tests/test_geak_gain_alignment.py | 96 +++++++++++++++++++ .../orchestrator/loop/coordinator_helpers.py | 25 +++-- src/hyperloom/orchestrator/loop/writeback.py | 41 ++++++-- 3 files changed, 147 insertions(+), 15 deletions(-) diff --git a/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py b/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py index 4858c9ab36..0be9041219 100644 --- a/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py +++ b/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py @@ -27,6 +27,7 @@ from hyperloom.inference_optimizer.breakdown.reporters._renderers.final import render as render_final from hyperloom.orchestrator.loop.coordinator import Coordinator from hyperloom.orchestrator.loop.coordinator_helpers import ( + _geak_result_has_material, _geak_revalidation_decision, _normalize_geak_overlay_dir, ) @@ -683,3 +684,98 @@ async def _must_not_fallback(**_kwargs): assert ss.cumulative_gain_provenance == "geak_orch_harness_validated" assert any(e.get("action") == "geak_e2e" for e in ss.optimization_stack) assert not ss.geak_pending + + +@pytest.mark.asyncio +async def test_2b_empty_result_without_prior_geak_e2e_does_not_promote(tmp_path: Path) -> None: + """A validated 2b decision with an EMPTY geak_result and NO pre-existing + geak_e2e stack entry has no material to validate: it is same-config noise + (geak_result lost / never populated), so it must NOT promote.""" + base, current_best, measured = 8668.5946, 8900.0, 9025.191 + coord = _coord(tmp_path, baseline=base, best_tput=current_best) + coord.shared_state.optimization_stack = [ + {"action": "explore", "variant_name": "kv-cache-fp8", "tput": current_best} + ] + coord.shared_state.resume_pending_revalidation = True + coord.shared_state.geak_pending = {"status": "awaiting_rebench"} + coord.shared_state.geak_result = {} # empty: cannot be judged by the helper + + async def _must_not_fallback(**_kwargs): + raise AssertionError("2a fallback must not run for a no-material drop") + + coord._validate_geak_via_geak_harness = _must_not_fallback # type: ignore[assignment] + + result = { + "output_throughput": measured, + "best_variant": {"fingerprint": "abc"}, + "winners": [], + } + await coord._promote_to_shared_state("explore", result, task=_revalidate_task(expected_hash="abc")) + + ss = coord.shared_state + assert ss.current_best["tput"] == pytest.approx(current_best) + assert ss.cumulative_gain_validated == pytest.approx(0.0) + assert ss.cumulative_gain_provenance != "geak_orch_harness_validated" + assert not any(e.get("action") == "geak_e2e" for e in ss.optimization_stack) + assert ss.resume_pending_revalidation is False + assert not ss.geak_pending + + +# ── material-guard helper unit boundaries ──────────────────────────────────── + + +@pytest.mark.parametrize( + ("result", "prev_flags", "prev_envs", "expected"), + [ + # Empty / non-dict -> cannot judge -> True (caller disambiguates). + ({}, "", {}, True), + (None, "", {}, True), + # No product, config identical to prev best -> non-material. + ( + {"accepted_config": {"flags": "--a 1", "env": "X=1"}, "accepted_kernels": []}, + "--a 1", + {"X": "1"}, + False, + ), + # Env order differs but semantics identical -> non-material. + ( + {"accepted_config": {"flags": "", "env": "A=1 B=2"}}, + "", + {"B": "2", "A": "1"}, + False, + ), + # accepted_kernels is a list of blank entries -> non-material. + ( + {"accepted_config": {"flags": "--a 1", "env": ""}, "accepted_kernels": ["", " "]}, + "--a 1", + {}, + False, + ), + # accepted_kernels has a real entry -> material. + ( + {"accepted_config": {"flags": "--a 1", "env": ""}, "accepted_kernels": ["fused_rope"]}, + "--a 1", + {}, + True, + ), + # final_overlay is whitespace only -> non-material (config identical). + ( + {"accepted_config": {"flags": "--a 1", "env": ""}, "final_overlay": " "}, + "--a 1", + {}, + False, + ), + # accepted_config adds a new env vs prev best -> material. + ( + {"accepted_config": {"flags": "--a 1", "env": "X=1 NEW=1"}}, + "--a 1", + {"X": "1"}, + True, + ), + ], +) +def test_geak_result_has_material_boundaries(result, prev_flags, prev_envs, expected) -> None: + assert ( + _geak_result_has_material(result, prev_best_flags=prev_flags, prev_best_envs=prev_envs) + is expected + ) diff --git a/src/hyperloom/orchestrator/loop/coordinator_helpers.py b/src/hyperloom/orchestrator/loop/coordinator_helpers.py index db4fab082c..7aacf62608 100644 --- a/src/hyperloom/orchestrator/loop/coordinator_helpers.py +++ b/src/hyperloom/orchestrator/loop/coordinator_helpers.py @@ -634,20 +634,21 @@ def _geak_result_has_material( """Decide whether a GEAK result carries a material optimization product. Guards the 2b promote path against pure passthrough noise: when GEAK ships - nothing (empty result) OR ships no kernel/head/overlay/patch AND echoes the - pre-KERNEL current_best config back unchanged, a rebench that beats - current_best is measurement variance, not a kernel gain. + no kernel/head/overlay/patch AND echoes the pre-KERNEL current_best config + back unchanged, a rebench that beats current_best is measurement variance, + not a kernel gain. Material means ANY of: - * ``accepted_kernels`` non-empty (kernel rewrites); - * ``accepted_heads`` non-empty (attention-head optimizations); + * ``accepted_kernels`` has a non-empty entry (kernel rewrites); + * ``accepted_heads`` has a non-empty entry (attention-head optimizations); * ``final_overlay`` non-empty (authored-kernel overlay dir); * ``final_patch`` non-empty (source patch); * ``accepted_config`` differs from the pre-KERNEL current_best config (a kernel enabled via a config switch, e.g. an ASM-GEMM env flag). - An empty/absent result returns ``True`` (cannot judge, so do not block the - native resume revalidation path). + An empty/absent result cannot be judged here and returns ``True``; the sole + 2b call site disambiguates it (a pre-existing ``geak_e2e`` stack entry means + a resume revalidation of an already-material win, otherwise no material). Args: result: The normalized GEAK ``geak_result`` blob. @@ -662,11 +663,17 @@ def _geak_result_has_material( canonical_fingerprint, ) + def _has_nonempty(entries: Any) -> bool: + # A list whose items are all empty/blank (e.g. ``[""]``) is not material. + if not isinstance(entries, (list, tuple, set)): + return bool(entries) + return any(str(e).strip() for e in entries) + if not isinstance(result, dict) or not result: return True - if result.get("accepted_kernels"): + if _has_nonempty(result.get("accepted_kernels")): return True - if result.get("accepted_heads"): + if _has_nonempty(result.get("accepted_heads")): return True if str(result.get("final_overlay") or "").strip(): return True diff --git a/src/hyperloom/orchestrator/loop/writeback.py b/src/hyperloom/orchestrator/loop/writeback.py index 8905d1010e..ad64db68dd 100644 --- a/src/hyperloom/orchestrator/loop/writeback.py +++ b/src/hyperloom/orchestrator/loop/writeback.py @@ -2821,12 +2821,23 @@ async def _promote_explore( # GEAK actually produced something. Without a material product # (kernel/head/overlay/patch or a config delta vs the pre-KERNEL # best) the win is same-config measurement noise; drop it. - if decision == "validated" and not _geak_result_has_material( - ps, - prev_best_flags=str(cb_now.get("extra_server_args") or ""), - prev_best_envs=cb_now.get("extra_envs") or {}, - ): - decision = "no_material" + # An empty geak_result cannot be judged by the helper, so + # disambiguate here: a pre-existing ``geak_e2e`` stack entry + # means this is a resume revalidation of an already-material win + # (let it through); otherwise there is no material to validate. + if decision == "validated": + stack_now = self.shared_state.optimization_stack or [] + has_prior_geak_e2e = any( + isinstance(e, dict) and e.get("action") == "geak_e2e" for e in stack_now + ) + if not ps and not has_prior_geak_e2e: + decision = "no_material" + elif ps and not _geak_result_has_material( + ps, + prev_best_flags=str(cb_now.get("extra_server_args") or ""), + prev_best_envs=cb_now.get("extra_envs") or {}, + ): + decision = "no_material" if decision == "validated": # Write the headline from the measured orchestrator-harness # rebench: lift current_best + optimization_stack + the @@ -2862,6 +2873,24 @@ async def _promote_explore( ) except Exception: # noqa: BLE001 - observation is best-effort log.exception("geak no_material: observation emit failed") + # Stamp the drop on geak_result and reject any provisional + # KEEP in kernel_journey so a session audit does not read a + # dropped candidate as an accepted kernel (no-op when the + # journey has no KEEP / the file is absent). + if isinstance(ps, dict) and ps: + ps["revalidation_status"] = "no_material" + self.shared_state.geak_result = ps + try: + self.phase_kernel._reject_geak_kernel_journey( + ps, + measured_tput=float(measured), + current_best_tput=( + float(cb_tput) if isinstance(cb_tput, (int, float)) else 0.0 + ), + provenance="geak_no_material", + ) + except Exception: # noqa: BLE001 - journey reject is best-effort + log.exception("geak no_material: journey rejection failed") self.shared_state.geak_pending = {} self.shared_state.resume_pending_revalidation = False elif decision == "no_promote": From b6357ada9fdc4f8224ef5ca63a24ae5a0a422a6a Mon Sep 17 00:00:00 2001 From: lishuoshuo-amd Date: Mon, 3 Aug 2026 18:46:27 +0800 Subject: [PATCH 3/3] fix(geak): harden no-material 2b guard against empty-config wipe and stale recovery re-enqueue - do not treat a missing/all-empty accepted_config as material (avoids promoting an empty config that wipes current_best) - tag no_material journey reject with geak_no_material_product reason - always stamp geak_result.revalidation_status=no_material (empty or not) - skip KERNEL crash-recovery re-enqueue when result was dropped as no_material - cover empty-config boundaries, resume promote, and journey revert in tests --- .../tests/test_geak_gain_alignment.py | 120 ++++++++++++++++++ .../orchestrator/loop/coordinator_helpers.py | 14 +- src/hyperloom/orchestrator/loop/writeback.py | 40 +++--- src/hyperloom/orchestrator/phases/kernel.py | 14 +- 4 files changed, 166 insertions(+), 22 deletions(-) diff --git a/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py b/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py index 0be9041219..98f90fd699 100644 --- a/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py +++ b/src/hyperloom/inference_optimizer/tests/test_geak_gain_alignment.py @@ -772,6 +772,22 @@ async def _must_not_fallback(**_kwargs): {"X": "1"}, True, ), + # accepted_config MISSING while prev best is non-empty -> non-material + # (a bare mismatch must not promote and wipe the existing config). + ( + {"status": "ok", "accepted_kernels": []}, + "--max-num-batched-tokens 24576", + {"VLLM_ROCM_USE_AITER": "1"}, + False, + ), + # accepted_config present but all-empty while prev best is non-empty -> + # non-material (same wipe hazard). + ( + {"status": "ok", "accepted_config": {"flags": "", "env": ""}}, + "--max-num-batched-tokens 24576", + {"VLLM_ROCM_USE_AITER": "1"}, + False, + ), ], ) def test_geak_result_has_material_boundaries(result, prev_flags, prev_envs, expected) -> None: @@ -779,3 +795,107 @@ def test_geak_result_has_material_boundaries(result, prev_flags, prev_envs, expe _geak_result_has_material(result, prev_best_flags=prev_flags, prev_best_envs=prev_envs) is expected ) + + +@pytest.mark.asyncio +async def test_2b_no_material_reverts_provisional_journey_keep(tmp_path: Path) -> None: + """A passthrough 2b drop must REVERT a provisional kernel_journey KEEP and + tag it with the no-material reason (not the beat-current_best reason).""" + base, current_best, measured = 8668.5946, 8900.0, 9025.191 + coord = _coord(tmp_path, baseline=base, best_tput=current_best) + coord.shared_state.current_best["extra_server_args"] = "--max-num-batched-tokens 24576" + coord.shared_state.current_best["extra_envs"] = {"VLLM_ROCM_USE_AITER": "1"} + coord.shared_state.optimization_stack = [ + {"action": "explore", "variant_name": "kv-cache-fp8", "tput": current_best} + ] + coord.shared_state.resume_pending_revalidation = True + coord.shared_state.geak_pending = {"status": "awaiting_rebench"} + journey_path = tmp_path / "kernel_journey.json" + journey_path.write_text( + json.dumps( + { + "kernels": [ + { + "kernel_id": "provisional-kernel", + "e2e": { + "integrated": True, + "e2e_gain_pct": 2.0, + "validated": True, + "decision": "KEEP", + }, + } + ] + } + ), + encoding="utf-8", + ) + # No material product; accepted_config is the pre-KERNEL best verbatim. + geak_result = { + "status": "ok", + "accepted_config": {"flags": "--max-num-batched-tokens 24576", "env": "VLLM_ROCM_USE_AITER=1"}, + "accepted_kernels": [], + "accepted_heads": [], + "final_overlay": "", + "final_patch": "", + "kernel_journey_path": str(journey_path), + } + coord.shared_state.geak_result = geak_result + coord._record_geak_kernel_journey(geak_result) + provisional_rows = { + row["kernel_id"]: row for row in assemble_parts(tmp_path)["kernel_journey"]["kernels"] + } + assert provisional_rows["provisional-kernel"]["e2e"]["decision"] == "KEEP" + + async def _must_not_fallback(**_kwargs): + raise AssertionError("2a fallback must not run for a no-material drop") + + coord._validate_geak_via_geak_harness = _must_not_fallback # type: ignore[assignment] + + result = { + "output_throughput": measured, + "best_variant": {"fingerprint": "abc"}, + "winners": [], + } + await coord._promote_to_shared_state("explore", result, task=_revalidate_task(expected_hash="abc")) + + ss = coord.shared_state + assert ss.current_best["tput"] == pytest.approx(current_best) + assert not any(e.get("action") == "geak_e2e" for e in ss.optimization_stack) + assert ss.geak_result["revalidation_status"] == "no_material" + rejected_rows = { + row["kernel_id"]: row for row in assemble_parts(tmp_path)["kernel_journey"]["kernels"] + } + e2e = rejected_rows["provisional-kernel"]["e2e"] + assert e2e["decision"] == "REVERT" + assert e2e["validated"] is False + assert e2e["rejection_reason"] == "geak_no_material_product" + + +@pytest.mark.asyncio +async def test_2b_empty_result_with_prior_geak_e2e_still_promotes(tmp_path: Path) -> None: + """Resume revalidation: geak_result was lost (empty) but a geak_e2e stack + entry already recorded the win. The 2b validated decision must still promote + (the material was proven in the original KERNEL cycle).""" + base, current_best, measured = 8668.5946, 8900.0, 9600.0 + coord = _coord(tmp_path, baseline=base, best_tput=current_best) + coord.shared_state.optimization_stack = [{"action": "geak_e2e", "tput": current_best}] + coord.shared_state.resume_pending_revalidation = True + coord.shared_state.geak_result = {} # lost on resume + + async def _must_not_fallback(**_kwargs): + raise AssertionError("2a fallback must not run when 2b validates a resume win") + + coord._validate_geak_via_geak_harness = _must_not_fallback # type: ignore[assignment] + + result = { + "output_throughput": measured, + "best_variant": {"fingerprint": "abc"}, + "winners": [], + } + await coord._promote_to_shared_state("explore", result, task=_revalidate_task(expected_hash="abc")) + + ss = coord.shared_state + expected_pct = (measured - base) / base * 100.0 + assert ss.cumulative_gain_validated == pytest.approx(expected_pct) + assert ss.cumulative_gain_provenance == "geak_orch_harness_validated" + assert ss.resume_pending_revalidation is False diff --git a/src/hyperloom/orchestrator/loop/coordinator_helpers.py b/src/hyperloom/orchestrator/loop/coordinator_helpers.py index 7aacf62608..e4b69c3206 100644 --- a/src/hyperloom/orchestrator/loop/coordinator_helpers.py +++ b/src/hyperloom/orchestrator/loop/coordinator_helpers.py @@ -633,6 +633,7 @@ def _geak_result_has_material( ) -> bool: """Decide whether a GEAK result carries a material optimization product. + FOR THE 2b REVALIDATION CALL SITE ONLY (writeback ``geak_fallback`` path). Guards the 2b promote path against pure passthrough noise: when GEAK ships no kernel/head/overlay/patch AND echoes the pre-KERNEL current_best config back unchanged, a rebench that beats current_best is measurement variance, @@ -643,8 +644,12 @@ def _geak_result_has_material( * ``accepted_heads`` has a non-empty entry (attention-head optimizations); * ``final_overlay`` non-empty (authored-kernel overlay dir); * ``final_patch`` non-empty (source patch); - * ``accepted_config`` differs from the pre-KERNEL current_best config - (a kernel enabled via a config switch, e.g. an ASM-GEMM env flag). + * ``accepted_config`` is present with a non-empty flags/env that differs + from the pre-KERNEL current_best config (a kernel enabled via a config + switch, e.g. an ASM-GEMM env flag). A missing or all-empty + ``accepted_config`` is NEVER material: an empty config that merely + differs from a non-empty current_best would otherwise promote and wipe + the existing config. An empty/absent result cannot be judged here and returns ``True``; the sole 2b call site disambiguates it (a pre-existing ``geak_e2e`` stack entry means @@ -684,6 +689,11 @@ def _has_nonempty(entries: Any) -> bool: parsed_envs, extra_flags = _split_env_and_flags(str(accepted_cfg.get("env") or "")) if extra_flags: accepted_flags = (accepted_flags + " " + extra_flags).strip() + # A missing / all-empty accepted_config carries no config optimization; a + # bare fingerprint mismatch against a non-empty current_best is NOT material + # (promoting it would wipe the existing config to empty). + if not accepted_flags and not parsed_envs: + return False got_fp = canonical_fingerprint(accepted_flags, parsed_envs) prev_fp = canonical_fingerprint( str(prev_best_flags or ""), diff --git a/src/hyperloom/orchestrator/loop/writeback.py b/src/hyperloom/orchestrator/loop/writeback.py index ad64db68dd..7c44f800d5 100644 --- a/src/hyperloom/orchestrator/loop/writeback.py +++ b/src/hyperloom/orchestrator/loop/writeback.py @@ -2873,24 +2873,28 @@ async def _promote_explore( ) except Exception: # noqa: BLE001 - observation is best-effort log.exception("geak no_material: observation emit failed") - # Stamp the drop on geak_result and reject any provisional - # KEEP in kernel_journey so a session audit does not read a - # dropped candidate as an accepted kernel (no-op when the - # journey has no KEEP / the file is absent). - if isinstance(ps, dict) and ps: - ps["revalidation_status"] = "no_material" - self.shared_state.geak_result = ps - try: - self.phase_kernel._reject_geak_kernel_journey( - ps, - measured_tput=float(measured), - current_best_tput=( - float(cb_tput) if isinstance(cb_tput, (int, float)) else 0.0 - ), - provenance="geak_no_material", - ) - except Exception: # noqa: BLE001 - journey reject is best-effort - log.exception("geak no_material: journey rejection failed") + # Stamp the drop on geak_result (always, so an empty {} is + # distinguishable from never-populated on resume/debug and + # acts as a tombstone against KERNEL crash-recovery + # re-enqueue) and reject any provisional KEEP in + # kernel_journey so a session audit does not read a dropped + # candidate as an accepted kernel (no-op when the journey + # has no KEEP / the file is absent). + ps_stamped = dict(ps) if isinstance(ps, dict) else {} + ps_stamped["revalidation_status"] = "no_material" + self.shared_state.geak_result = ps_stamped + try: + self.phase_kernel._reject_geak_kernel_journey( + ps_stamped, + measured_tput=float(measured), + current_best_tput=( + float(cb_tput) if isinstance(cb_tput, (int, float)) else 0.0 + ), + provenance="geak_no_material", + rejection_reason="geak_no_material_product", + ) + except Exception: # noqa: BLE001 - journey reject is best-effort + log.exception("geak no_material: journey rejection failed") self.shared_state.geak_pending = {} self.shared_state.resume_pending_revalidation = False elif decision == "no_promote": diff --git a/src/hyperloom/orchestrator/phases/kernel.py b/src/hyperloom/orchestrator/phases/kernel.py index 3ab5835ad2..0508c85ddf 100644 --- a/src/hyperloom/orchestrator/phases/kernel.py +++ b/src/hyperloom/orchestrator/phases/kernel.py @@ -778,7 +778,16 @@ async def _enqueue_geak_revalidation(*, reason: str) -> bool: # so a prior cycle's result.json does not short-circuit a fresh entry. result_path = out_dir / "result.json" recovered = _read_geak_result(result_path) - if recovered.get("status") == "ok" and not self._geak_win_already_recorded(): + # Tombstone: a result already dropped by 2b as no-material must not be + # re-recovered from the stale (still status=ok) result.json, else each + # KERNEL entry re-enqueues a wasted rebench in a loop. + prev_geak = ( + self.shared_state.geak_result + if isinstance(getattr(self.shared_state, "geak_result", None), dict) + else {} + ) + dropped_no_material = str(prev_geak.get("revalidation_status") or "") == "no_material" + if recovered.get("status") == "ok" and not self._geak_win_already_recorded() and not dropped_no_material: log.info( "GEAK result.json exists but state has no recorded win " "(crash before handback); promoting recovered result." @@ -1364,6 +1373,7 @@ def _reject_geak_kernel_journey( measured_tput: float, current_best_tput: float, provenance: str, + rejection_reason: str = "rebench_did_not_beat_current_best", ) -> None: """Replace provisional GEAK e2e KEEPs after a failed final rebench.""" @@ -1402,7 +1412,7 @@ def _reject_geak_kernel_journey( "revalidation_measured_tput": measured_tput, "revalidation_current_best_tput": current_best_tput, "revalidation_provenance": provenance, - "rejection_reason": "rebench_did_not_beat_current_best", + "rejection_reason": rejection_reason, } ) try: