Stop after gmu ceiling validation is exhausted

src/aituner/harness.py

@@ -2104,12 +2104,6 @@ def _validation_exhausted_guard(
     if baseline_rate <= 0 or incumbent_rate <= 0:
         return default
     gain = incumbent_rate / baseline_rate
-    if gain < _STRONG_INCUMBENT_MIN_GAIN:
-        return {
-            **default,
-            "reason": "incumbent_gain_not_large_enough_for_validation_stop",
-            "incumbent_gain_vs_baseline": gain,
-        }
 
     best_index = next(
         (
@@ -2130,6 +2124,21 @@ def _validation_exhausted_guard(
         for item in recent_diagnostics[best_index + 1 :]
         if item.get("status") in {"completed", "failed"}
     ]
+    incumbent = next(
+        (
+            item
+            for item in recent_diagnostics
+            if item.get("trial_id") == state.best_trial_id
+        ),
+        {},
+    )
+    gmu_ceiling_incumbent = _is_gpu_memory_utilization_ceiling_incumbent(incumbent)
+    if gain < _STRONG_INCUMBENT_MIN_GAIN and not gmu_ceiling_incumbent:
+        return {
+            **default,
+            "reason": "incumbent_gain_not_large_enough_for_validation_stop",
+            "incumbent_gain_vs_baseline": gain,
+        }
     if len(after_best) < _MIN_POST_INCUMBENT_VALIDATION_TRIALS:
         return {
             **default,
@@ -2154,6 +2163,7 @@ def _validation_exhausted_guard(
     families: set[str] = set()
     for item in after_best:
         families.update(_validation_families(item))
+    families.update(_validation_families(incumbent))
     has_topology = "topology" in families
     has_runtime = bool(families & {"runtime", "max-num-seqs", "max-num-batched-tokens"})
     enough_evidence = (
@@ -2202,6 +2212,17 @@ def _validation_families(item: dict[str, Any]) -> set[str]:
     return families
 
 
+def _is_gpu_memory_utilization_ceiling_incumbent(item: dict[str, Any]) -> bool:
+    config_patch = item.get("config_patch")
+    if not isinstance(config_patch, dict):
+        return False
+    flag_patch = config_patch.get("flag_patch")
+    if not isinstance(flag_patch, dict):
+        return False
+    gmu = _parse_float_like(flag_patch.get("gpu-memory-utilization"), default=0.0)
+    return gmu >= _GMU_SAFE_CEILING - EPSILON
+
+
 def _strong_incumbent_guard(
     state: StudyState,
     recent_diagnostics: list[dict[str, Any]],