diff --git a/src/hyperloom/inference_optimizer/tests/test_grid_runner_behavior_lock.py b/src/hyperloom/inference_optimizer/tests/test_grid_runner_behavior_lock.py index ff7631d4c2..f0d34d9a7f 100644 --- a/src/hyperloom/inference_optimizer/tests/test_grid_runner_behavior_lock.py +++ b/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 diff --git a/src/hyperloom/orchestrator/actions/executors/_grid_runner.py b/src/hyperloom/orchestrator/actions/executors/_grid_runner.py index 63784f539a..21606c0001 100644 --- a/src/hyperloom/orchestrator/actions/executors/_grid_runner.py +++ b/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) diff --git a/src/hyperloom/orchestrator/actions/executors/benchmark_result.py b/src/hyperloom/orchestrator/actions/executors/benchmark_result.py index 3a18e79859..af149ac9b3 100644 --- a/src/hyperloom/orchestrator/actions/executors/benchmark_result.py +++ b/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", ] diff --git a/src/hyperloom/orchestrator/actions/executors/profile.py b/src/hyperloom/orchestrator/actions/executors/profile.py index 717f11772e..020a4153b3 100644 --- a/src/hyperloom/orchestrator/actions/executors/profile.py +++ b/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."""