Stop harness when search high is saturated

tests/test_core_flow.py

@@ -538,6 +538,64 @@ class CoreFlowTests(unittest.TestCase):
             self.assertFalse(context["harness_stop"]["should_stop"])
             self.assertIsNone(build_harness_stop_proposal(context))
 
+    def test_harness_stop_when_incumbent_saturates_search_high(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            study_path = _write_study_assets(tmp_path)
+            study = load_study_spec(study_path)
+            result_path = tmp_path / "trial-0001.json"
+            result_path.write_text(
+                json.dumps(
+                    {
+                        "status": "completed",
+                        "best_sampling_u": 0.99609375,
+                        "best_request_rate": 9.0,
+                        "best_pass_rate": 1.0,
+                        "probes": [
+                            {
+                                "threshold": 0.99609375,
+                                "feasible": True,
+                                "payload": {
+                                    "request_count": 10,
+                                    "pass_rate": 1.0,
+                                    "request_rate": 9.0,
+                                    "early_stopped": False,
+                                    "early_stop_reason": "",
+                                    "latency_summary": {"failed_reason_counts": {}},
+                                },
+                            }
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+            state = StudyState(
+                study_id=study.study_id,
+                best_trial_id="trial-0001",
+                best_request_rate=9.0,
+                best_request_rate_per_gpu=9.0,
+                trials=[
+                    TrialSummary(
+                        trial_id="trial-0001",
+                        status="completed",
+                        best_request_rate=9.0,
+                        best_request_rate_per_gpu=9.0,
+                        result_path=str(result_path),
+                        config_patch={"env_patch": {}, "flag_patch": {}},
+                    )
+                ],
+            )
+            context = build_harness_context(
+                study=study,
+                window_summary={"prompt_tokens_p95": 2048},
+                state=state,
+            )
+            self.assertTrue(context["harness_stop"]["should_stop"])
+            self.assertEqual(context["harness_stop"]["reason"], "search_high_saturated_by_incumbent")
+            proposal = build_harness_stop_proposal(context)
+            self.assertIsNotNone(proposal)
+            self.assertTrue(proposal.should_stop)
+
     def test_trace_input_length_filter_keeps_only_matching_rows(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             tmp_path = Path(tmp)