fix(geak): gate 2b revalidation promote on material candidate to reject same-config rebench noise

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,
 )
@@ -585,3 +586,316 @@ 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
+
+
+@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,
+        ),
+        # 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:
+    assert (
+        _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