fix(grid): a profiled AgentX round that Magpie completed is a succeeded round of run_grid

src/hyperloom/inference_optimizer/tests/test_grid_runner_behavior_lock.py

@@ -204,6 +204,71 @@ async def _restart_fail(**_kwargs):
         assert landed == [("c0", 1), ("c1", 2)]
 
 
+def _agentx_workspace(slot: Path, **report: object) -> Path:
+    """A ``benchmark_*`` workspace whose report is Magpie's native AgentX report, which is ``diagnostic_only`` when the
+    round profiles."""
+    ws = slot / "benchmark_sglang_20260101_000000"
+    ws.mkdir(parents=True, exist_ok=True)
+    metrics = {
+        "execution_owner": "magpie",
+        "server_launch": {"base_argv": ["python3"], "effective_argv": ["python3"], "effective_env": {}},
+        "recipe": {"tp": 1, "ep": 1, "conc": 8},
+        "throughput": {
+            "output_tokens_per_second": 280.0,
+            "total_tokens_per_second": 25000.0,
+            "duration_seconds": 3600.0,
+        },
+        "requests": {"successful": 1029, "total": 1072},
+        **report.pop("metrics", {}),
+    }
+    document = {"scenario": "agentx", "success": True, "framework": "sglang", "agentx_metrics": metrics, **report}
+    (ws / "benchmark_report.json").write_text(json.dumps(document))
+    return ws
+
+
+class TestADiagnosticAgentxRoundIsAVerdict:
+    """Magpie marks a profiled AgentX replay ``diagnostic_only`` and not publishable: it is no valid measurement, and
+    the round it ends is still a success, which ``round_succeeded`` states (the main flow's ``ProfileExecutor`` reads
+    the same field)."""
+
+    def _run(self, tmp_path, monkeypatch, **report):
+        monkeypatch.setattr(gr, "REPORT_SETTLE_SECONDS", 0.0)
+        monkeypatch.setenv("INFERENCE_OPTIMIZER_RUN_GRID_WARMUP", "0")
+        base = tmp_path / "base.yaml"
+        _write_base_yaml(base)
+
+        def _magpie(cmd, *a, **k):
+            _agentx_workspace(Path(cmd[cmd.index("--output-dir") + 1]), **report)
+            return subprocess.CompletedProcess(cmd, 0, "ok", "")
+
+        results, _ = _run_capturing_variant_notes(
+            multi_node=False, run_side_effect=_magpie, base=base, out=tmp_path / "out"
+        )
+        return results[0]
+
+    def test_a_profiled_replay_succeeds_without_being_a_valid_measurement(self, tmp_path, monkeypatch):
+        result = self._run(
+            tmp_path,
+            monkeypatch,
+            benchmark_valid=False,
+            publishable=False,
+            metrics={"diagnostic_only": True, "profile_capture": {"status": "complete"}},
+        )
+
+        assert (result.status, result.error_class) == ("succeeded", "")
+        assert result.diagnostic_only is True and result.benchmark_valid is False
+
+    def test_a_replay_that_is_neither_valid_nor_diagnostic_still_fails(self, tmp_path, monkeypatch):
+        result = self._run(tmp_path, monkeypatch, benchmark_valid=False, publishable=False)
+
+        assert (result.status, result.error_class) == ("failed", "benchmark_report_invalid_metric")
+
+    def test_a_valid_replay_succeeds(self, tmp_path, monkeypatch):
+        result = self._run(tmp_path, monkeypatch, benchmark_valid=True, publishable=True)
+
+        assert result.status == "succeeded" and result.benchmark_valid is True and result.diagnostic_only is False
+
+
 # keep_going_on_failure asymmetry
 
 

src/hyperloom/orchestrator/actions/executors/_grid_runner.py

@@ -38,9 +38,7 @@
     StoppedByTheRun,
 )
 from ._benchmark_runtime import apply_runtime_benchmark_overrides as apply_runtime_benchmark_overrides
-from ._benchmark_interpreter import (
-    _resolve_probe_python as _resolve_probe_python,
-)
+from ._benchmark_interpreter import _resolve_probe_python as _resolve_probe_python
 from ._subprocess_kill import (
     AGENTX_PREFLIGHT_ERROR_CLASS,
     AGENTX_PREFLIGHT_RETURNCODE,
@@ -58,6 +56,7 @@
 from .benchmark_result import (
     extract_benchmark_measurement,
     harvest_leaked_artifacts,
+    round_succeeded,
     select_run_workspace,
     served_complete_protocol,
     snapshot_workspaces,
@@ -340,8 +339,9 @@ async def _settled_measurement(
             subprocess_started_unix=subprocess_started_unix,
         )
         attempts += 1
-        if measurement.get("valid_measurement") or time.monotonic() >= deadline:
-            if attempts > 1 and measurement.get("valid_measurement"):
+        succeeded = round_succeeded(measurement)
+        if succeeded or time.monotonic() >= deadline:
+            if attempts > 1 and succeeded:
                 log.info(
                     "grid_runner: benchmark_report became valid after %d read(s); "
                     "the report was still being written when the subprocess was reaped",
@@ -1435,7 +1435,7 @@ async def _variant_row(run: _GridRun, at: _VariantSlot, measured: _MeasuredVaria
         settle_seconds=REPORT_SETTLE_SECONDS,
     )
     outcome = _VariantOutcome(measured, report, measurement, _round_warnings(measurement, measured))
-    if measurement.get("valid_measurement"):
+    if round_succeeded(measurement):
         return _measured_row(at, outcome)
     if run.lifecycle_boot_only and measured.round.proc.returncode == 0:
         return _booted_server_row(at, outcome)

src/hyperloom/orchestrator/actions/executors/benchmark_result.py

@@ -842,6 +842,15 @@ def is_valid_measurement(result: dict[str, Any] | None) -> bool:
     return completed is not None and completed > 0
 
 
+def round_succeeded(measurement: dict[str, Any]) -> bool:
+    """Whether a round ran to the verdict its kind accepts.
+
+    A native AgentX round that Magpie marks diagnostic (a profile) is never a valid measurement: the trace it leaves is
+    what it is run for, so Magpie's own ``round_succeeded`` decides it. Any other round succeeds when it is valid.
+    """
+    return measurement.get("round_succeeded", measurement.get("valid_measurement")) is True
+
+
 def served_complete_protocol(result: dict[str, Any]) -> bool:
     """Return whether the run served every request its protocol asked for.
 
@@ -903,6 +912,7 @@ def double_run_requested(params: dict | None) -> bool:
     "harvest_leaked_artifacts",
     "double_run_requested",
     "is_valid_measurement",
+    "round_succeeded",
     "served_complete_protocol",
     "_materialize_rescue_into_workspace",
 ]

src/hyperloom/orchestrator/actions/executors/profile.py

@@ -36,6 +36,7 @@
     ensure_benchmark_serving_patched,
 )
 from ._xdit_patcher import verify_xdit_profiler_baked
+from .benchmark_result import round_succeeded
 from .baseline import BenchmarkRunExecutor
 
 
@@ -1322,8 +1323,7 @@ async def __call__(self, ctx) -> dict[str, Any]:
             result["trace_input_ready"] = True
         return result
 
-    def _measurement_succeeded(self, measurement: dict[str, Any]) -> bool:
-        return measurement.get("round_succeeded", measurement.get("valid_measurement")) is True
+    _measurement_succeeded = staticmethod(round_succeeded)
 
     async def _run_native_profile(self, ctx) -> dict[str, Any]:
         """Collect Magpie-owned captures and certify each time window separately."""