diff --git a/pyproject.toml b/pyproject.toml index ad6d66e843..aeb5e0bb2c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -125,6 +125,16 @@ ast = [ claude = [ "claude-agent-sdk>=0.2.110", ] +# InferaSim (Infera) serving-projection benchmark backend +# (HYPERLOOM_BENCHMARK_BACKEND=inferasim). Lazy-imported by +# ``orchestrator.actions.executors.inferasim_bridge``; not needed unless the +# inferasim backend is selected. Infera is not yet on PyPI, so most deployments +# instead point HYPERLOOM_INFERASIM_ROOT at an Infera checkout (or +# HYPERLOOM_INFERASIM_PYTHON at an interpreter that can import ``infera``); this +# extra is a convenience for when it is pip-installable. +inferasim = [ + "amd-infera", +] # Local dev tooling: pre-commit hooks, formatters, type checker. dev = [ "pre-commit>=4.0", @@ -236,6 +246,8 @@ hyperloom = [ "assets/configs/*.yaml", "assets/agentx/*.sh", "assets/agentx/*.py", + # InferaSim serving-projection benchmark backend workload template. + "assets/inferasim/*.yaml", # Host-side evidence probe, injected into the benchmark process via a # PYTHONPATH prefix (see _framework_rewrite_evidence). Shipped so wheel # installs can arm it. diff --git a/src/hyperloom/inference_optimizer/assets/inferasim/inferasim_workload.yaml b/src/hyperloom/inference_optimizer/assets/inferasim/inferasim_workload.yaml new file mode 100644 index 0000000000..2a5a37f5c4 --- /dev/null +++ b/src/hyperloom/inference_optimizer/assets/inferasim/inferasim_workload.yaml @@ -0,0 +1,42 @@ +# InferaSim workload template used by Hyperloom's inferasim benchmark backend. +# +# This is an env-driven Infera (inferasim) workload spec. The inferasim bridge +# (hyperloom.orchestrator.actions.executors.inferasim_bridge) points +# `inferasim inference --config` at this file and sets: +# INFERASIM_MODEL - model preset name (e.g. gpt_oss_120B), from the resolved +# Hyperloom model or HYPERLOOM_INFERASIM_MODEL +# INFERASIM_TP/PP/EP - parallelism (also forced via CLI overrides) +# +# The model preset (`.yaml`) is resolved from Infera's own +# configs/models/megatron search path, so this template works regardless of +# where it lives. To use a fully custom Infera workload instead, set +# HYPERLOOM_INFERASIM_WORKLOAD=/path/to/your_workload.yaml. +work_group: ${INFERASIM_TEAM:hyperloom} +user_name: ${INFERASIM_USER:hyperloom} +exp_name: ${INFERASIM_EXP_NAME:hyperloom-inferasim} +workspace: ./output + +modules: + pre_trainer: + framework: megatron + config: pre_trainer.yaml + + # Model preset to project; overridden per-run via INFERASIM_MODEL. + model: ${INFERASIM_MODEL:gpt_oss_120B}.yaml + overrides: + # Sequence sizing; the serving request length is set separately by the + # bridge via --input-len / --output-len. + seq_length: ${INFERASIM_SEQ_LENGTH:4096} + max_position_embeddings: ${INFERASIM_MAX_POSITION_EMBEDDINGS:4096} + + # Parallelism (env defaults; the bridge also forces these via CLI + # overrides so an explicit workload YAML is honored too). + tensor_model_parallel_size: ${INFERASIM_TP:1} + pipeline_model_parallel_size: ${INFERASIM_PP:1} + expert_model_parallel_size: ${INFERASIM_EP:1} + + # Keep the projection self-contained (no data / checkpoints). + mock_data: true + train_data_path: null + valid_data_path: null + test_data_path: null diff --git a/src/hyperloom/inference_optimizer/tests/test_backend_gating.py b/src/hyperloom/inference_optimizer/tests/test_backend_gating.py index 57cb5f2960..3abe6eb6f4 100644 --- a/src/hyperloom/inference_optimizer/tests/test_backend_gating.py +++ b/src/hyperloom/inference_optimizer/tests/test_backend_gating.py @@ -147,6 +147,34 @@ def test_lifecycle_magpie_default_unchanged(tmp_path, monkeypatch): assert info["eligible"] is True # magpie built-in script -> eligible +def test_lifecycle_ineligible_against_external_server(tmp_path, monkeypatch): + """An external endpoint makes the reuse protocol ineligible single-node. + + Magpie runs client-only there, so there is no local server to boot on round + one or re-attach to on round two. Same config as the eligible case above: + only the hand-off env differs. + """ + import yaml + from hyperloom.orchestrator.actions.executors import _server_lifecycle as sl + + monkeypatch.delenv(bb.BENCHMARK_BACKEND_ENV, raising=False) + monkeypatch.setenv("HYPERLOOM_MN_EXT_SERVICE_URL", "http://infera:8000") + cfg = { + "benchmark": { + "framework": "vllm", + "benchmark_script": "vllm_mi300x.sh", + "envs": {"PORT": 8888}, + "profiler": {"torch_profiler": {"enabled": False}}, + } + } + cfg_path = tmp_path / "cfg.yaml" + cfg_path.write_text(yaml.safe_dump(cfg), encoding="utf-8") + + info = sl.resolve_lifecycle_params(cfg_path) + assert info["eligible"] is False + assert "external server" in info["reason"] + + def test_lifecycle_magpie_non_builtin_ineligible(tmp_path, monkeypatch): """magpie backend: a non-built-in script stays ineligible (unchanged).""" import yaml diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_plateau_round_window.py b/src/hyperloom/inference_optimizer/tests/test_explore_plateau_round_window.py new file mode 100644 index 0000000000..7ce8bac04f --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_explore_plateau_round_window.py @@ -0,0 +1,216 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""EXPLORE plateau: the gain window, and what counts as an unproductive round. + +The plateau is the only exit EXPLORE has that is not a clock. These cover the +two ways it could never be reached, so a regression that re-disables it shows up +here rather than as a phase that always spends its whole budget. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from hyperloom.orchestrator.phases.machine_state import ( + DEFAULT_PLATEAU_EXPLORE_EMPTY_STREAK, + compute_plateau_explore, +) + +GATE = "HYPERLOOM_EXPLORE_PLATEAU_ROUND_WINDOW" + + +def _state(winners, rounds): + return SimpleNamespace( + explore_search={"winners_history": list(winners)}, + specialist_rounds=list(rounds), + ) + + +def _barren_rounds(n, *, start=0, proposals=12): + """Rounds that proposed a full grid and kept none of it.""" + return [{"round_id": f"r{i}", "proposals_total": proposals, "proposals_kept": 0} for i in range(start, start + n)] + + +def _productive_round(idx, *, kept=1, proposals=12): + return {"round_id": f"r{idx}", "proposals_total": proposals, "proposals_kept": kept} + + +# --- the gain window --------------------------------------------------------- + + +def test_single_keep_no_longer_disables_the_gain_arm(monkeypatch): + """One early win must not hold the plateau off for the rest of the cycle. + + A winner is recorded only because it cleared the KEEP threshold, so a window + over the last N winners always sums above the plateau floor. Scoped to + recent rounds instead, the sum falls away once those rounds stop keeping. + """ + monkeypatch.setenv(GATE, "1") + winners = [{"round_id": "r0", "gain_pct": 1.5}] + rounds = [_productive_round(0)] + _barren_rounds(DEFAULT_PLATEAU_EXPLORE_EMPTY_STREAK, start=1) + triggered, ev = compute_plateau_explore(_state(winners, rounds)) + assert triggered is True + assert ev["gain_window"] == "recent_rounds" + assert ev["recent_keep_gain_pct"] == 0.0 + + +def test_legacy_window_is_unsatisfiable_after_one_keep(monkeypatch): + """The behaviour the gate restores, pinned so the contrast is explicit.""" + monkeypatch.setenv(GATE, "0") + winners = [{"round_id": "r0", "gain_pct": 1.5}] + rounds = [_productive_round(0)] + _barren_rounds(DEFAULT_PLATEAU_EXPLORE_EMPTY_STREAK, start=1) + triggered, ev = compute_plateau_explore(_state(winners, rounds)) + assert triggered is False + assert ev["gain_window"] == "recent_winners" + assert ev["recent_keep_gain_pct"] == pytest.approx(1.5) + + +def test_a_win_inside_the_window_still_holds_the_plateau_off(monkeypatch): + """The arm must still say 'no' while recent rounds are producing gain.""" + monkeypatch.setenv(GATE, "1") + rounds = _barren_rounds(4) + [_productive_round(4)] + winners = [{"round_id": "r4", "gain_pct": 2.0}] + triggered, ev = compute_plateau_explore(_state(winners, rounds)) + assert triggered is False + assert ev["recent_keep_gain_pct"] == pytest.approx(2.0) + + +def test_gain_sums_only_winners_from_the_recent_rounds(monkeypatch): + monkeypatch.setenv(GATE, "1") + rounds = [_productive_round(i) for i in range(8)] + winners = [{"round_id": f"r{i}", "gain_pct": 1.0} for i in range(8)] + _, ev = compute_plateau_explore(_state(winners, rounds), lookback=3) + assert ev["recent_keep_gain_pct"] == pytest.approx(3.0) + assert ev["winners_seen"] == 3 + + +# --- what counts as unproductive -------------------------------------------- + + +def test_rounds_that_keep_nothing_count_toward_the_streak(monkeypatch): + """Proposing a grid and keeping none of it is not a productive round.""" + monkeypatch.setenv(GATE, "1") + rounds = [_productive_round(0)] + _barren_rounds(6, start=1) + _, ev = compute_plateau_explore(_state([], rounds)) + assert ev["empty_streak"] == 6 + assert ev["empty_streak_basis"] == "kept_nothing" + + +def test_legacy_streak_ignores_rounds_that_merely_kept_nothing(monkeypatch): + monkeypatch.setenv(GATE, "0") + _, ev = compute_plateau_explore(_state([], _barren_rounds(6))) + assert ev["empty_streak"] == 0 + assert ev["empty_streak_basis"] == "no_proposals" + + +def test_a_kept_round_breaks_the_streak(monkeypatch): + monkeypatch.setenv(GATE, "1") + rounds = _barren_rounds(4) + [_productive_round(4)] + _barren_rounds(2, start=5) + _, ev = compute_plateau_explore(_state([], rounds)) + assert ev["empty_streak"] == 2 + + +def test_rounds_with_no_proposals_still_count(monkeypatch): + """The original signal is a subset of the new one, not a casualty of it.""" + monkeypatch.setenv(GATE, "1") + rounds = [{"round_id": f"r{i}", "proposals_total": 0, "proposals_kept": 0} for i in range(5)] + _, ev = compute_plateau_explore(_state([], rounds)) + assert ev["empty_streak"] == 5 + + +def test_malformed_round_is_filtered_before_the_streak_walk(monkeypatch): + """Non-dict rows are dropped by the cycle filter, so they neither count nor break. + + Worth pinning: the streak predicates guard against non-dicts, which reads as + though a malformed row would break the streak. It cannot reach them. + """ + monkeypatch.setenv(GATE, "1") + rounds = _barren_rounds(3) + ["not-a-dict"] + _barren_rounds(2, start=4) + _, ev = compute_plateau_explore(_state([], rounds)) + assert ev["specialist_rounds_seen"] == 5 + assert ev["empty_streak"] == 5 + + +def test_legacy_kept_count_fallback_key(monkeypatch): + """Older round summaries used ``kept_count``.""" + monkeypatch.setenv(GATE, "1") + rounds = [{"round_id": f"r{i}", "proposal_count": 4, "kept_count": 0} for i in range(5)] + _, ev = compute_plateau_explore(_state([], rounds)) + assert ev["empty_streak"] == 5 + + +# --- degrading safely -------------------------------------------------------- + + +def test_winners_without_round_attribution_fall_back(monkeypatch): + """Missing attribution must not read as an empty window and exit the phase.""" + monkeypatch.setenv(GATE, "1") + winners = [{"gain_pct": 1.5}, {"gain_pct": 2.0}] + rounds = [_productive_round(0)] + _barren_rounds(6, start=1) + triggered, ev = compute_plateau_explore(_state(winners, rounds)) + assert ev["gain_window"] == "recent_winners" + assert triggered is False + + +def test_rounds_without_ids_fall_back(monkeypatch): + monkeypatch.setenv(GATE, "1") + winners = [{"round_id": "r0", "gain_pct": 1.5}] + rounds = [{"proposals_total": 12, "proposals_kept": 0} for _ in range(6)] + _, ev = compute_plateau_explore(_state(winners, rounds)) + assert ev["gain_window"] == "recent_winners" + + +def test_no_winners_at_all_uses_the_round_window(monkeypatch): + monkeypatch.setenv(GATE, "1") + triggered, ev = compute_plateau_explore(_state([], _barren_rounds(6))) + assert ev["gain_window"] == "recent_rounds" + assert triggered is True + + +def test_empty_state_is_inert(monkeypatch): + monkeypatch.setenv(GATE, "1") + triggered, ev = compute_plateau_explore(SimpleNamespace()) + assert triggered is False + assert ev["empty_streak"] == 0 + assert ev["recent_keep_gain_pct"] == 0.0 + + +def test_lookback_disabled_short_circuits(monkeypatch): + monkeypatch.setenv(GATE, "1") + triggered, ev = compute_plateau_explore(_state([], _barren_rounds(9)), lookback=0) + assert triggered is False + assert ev["reason"] == "lookback_disabled" + + +def test_streak_below_threshold_does_not_trigger(monkeypatch): + """Both arms are required; a short streak is not a plateau.""" + monkeypatch.setenv(GATE, "1") + rounds = [_productive_round(0)] + _barren_rounds(2, start=1) + triggered, ev = compute_plateau_explore(_state([], rounds)) + assert ev["recent_keep_gain_pct"] == 0.0 + assert triggered is False + + +def test_unset_gate_defaults_to_the_round_window(monkeypatch): + monkeypatch.delenv(GATE, raising=False) + _, ev = compute_plateau_explore(_state([], _barren_rounds(6))) + assert ev["gain_window"] == "recent_rounds" + assert ev["empty_streak_basis"] == "kept_nothing" + + +@pytest.mark.parametrize("raw", ["0", "false", "no", "off", "OFF"]) +def test_gate_accepts_falsey_spellings(monkeypatch, raw): + monkeypatch.setenv(GATE, raw) + _, ev = compute_plateau_explore(_state([], _barren_rounds(6))) + assert ev["empty_streak_basis"] == "no_proposals" + + +def test_unparseable_gain_is_skipped_not_fatal(monkeypatch): + monkeypatch.setenv(GATE, "1") + rounds = [_productive_round(0)] + winners = [{"round_id": "r0", "gain_pct": "nonsense"}] + _, ev = compute_plateau_explore(_state(winners, rounds)) + assert ev["recent_keep_gain_pct"] == 0.0 diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_screen.py b/src/hyperloom/inference_optimizer/tests/test_explore_screen.py new file mode 100644 index 0000000000..2ef1874024 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_explore_screen.py @@ -0,0 +1,352 @@ +"""The screen may shorten EXPLORE's grid; it may never lose a winner. + +Pruning on a cheap probe is the one change here that can make Hyperloom worse +without making it look worse: a dropped variant produces no measurement, so a +discarded winner leaves no trace in the ledger. So most of these tests are about +what the screen refuses to drop -- anything when it is off, when the probe cannot +run, when the grid is too small to be worth pruning, and any individual variant +whose own probe came back unreadable. +""" + +from __future__ import annotations + +import json + +import pytest + +from hyperloom.orchestrator.actions.executors import _explore_screen +from hyperloom.orchestrator.actions.executors._explore_screen import ( + BASELINE, + ENV_BACKEND, + ENV_ENABLED, + ENV_GPUS, + ENV_INFERA_ROOT, + ENV_LAYERS, + ENV_MARGIN_PCT, + kernels_from_log, + screen_enabled, + screen_variants, +) +from hyperloom.orchestrator.actions.executors._grid_base import GridVariant + +CONFIG = """ +benchmark: + framework: vllm + model: /models/gpt-oss-120b + envs: + TP: 8 + CONC: 32 + ISL: 1024 +""" + + +@pytest.fixture +def config(tmp_path): + path = tmp_path / "bench.yaml" + path.write_text(CONFIG) + return path + + +KERNEL_LOG = ( + "INFO [rocm.py:556] Using ROCM_AITER_UNIFIED_ATTN backend " + "(selected via --attention-backend).\n" + "INFO [mxfp4.py:514] Using 'TRITON' Mxfp4 MoE backend.\n" +) + + +@pytest.fixture +def session(tmp_path): + """A session that has already benchmarked the stack, so its kernels are known.""" + slot = tmp_path / "session" / "explore-001" / "base" + slot.mkdir(parents=True) + (slot / "server.log").write_text(KERNEL_LOG) + return tmp_path / "session" + + +@pytest.fixture +def enabled(monkeypatch, tmp_path): + monkeypatch.setenv(ENV_ENABLED, "1") + monkeypatch.setenv(ENV_INFERA_ROOT, str(tmp_path)) + return monkeypatch + + +def variants(n=4): + return [GridVariant(name=f"v{i}", extra_server_args=f"--max-num-seqs {i}") for i in range(n)] + + +MATCHING_KERNELS = {"attention": "ROCM_AITER_UNIFIED_ATTN", "moe": "TRITON"} + + +def fake_probe(readings, baseline=10.0, kernels=None): + """Stand in for the GPU probe: a canned reading and kernel set per variant.""" + resolved = MATCHING_KERNELS if kernels is None else kernels + + def probe(variant, bench, timeout_sec, backend): + if variant.name == BASELINE: + return baseline, resolved + return readings.get(variant.name), resolved + + return probe + + +# --- the screen stays out of the way --------------------------------------- + + +def test_disabled_by_default(config, session): + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert [v.name for v in kept] == ["v0", "v1", "v2", "v3"] + assert dropped == [] + + +def test_enabled_flag_is_explicit(monkeypatch): + assert screen_enabled() is False + monkeypatch.setenv(ENV_ENABLED, "1") + assert screen_enabled() is True + monkeypatch.setenv(ENV_ENABLED, "0") + assert screen_enabled() is False + + +def test_small_grid_is_not_worth_pruning(config, enabled, session): + kept, dropped = screen_variants(variants(2), config, session_dir=session) + assert len(kept) == 2 and dropped == [] + + +def test_no_infera_checkout_means_no_screen(config, session, monkeypatch): + monkeypatch.setenv(ENV_ENABLED, "1") + monkeypatch.delenv(ENV_INFERA_ROOT, raising=False) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert len(kept) == 4 and dropped == [] + + +def test_unreadable_config_means_no_screen(tmp_path, enabled, session): + missing = tmp_path / "absent.yaml" + kept, dropped = screen_variants(variants(), missing, session_dir=session) + assert len(kept) == 4 and dropped == [] + + +def test_failed_baseline_probe_benchmarks_the_whole_grid(config, enabled, session, monkeypatch): + """Without a reference reading there is nothing to measure a margin against.""" + monkeypatch.setattr(_explore_screen, "_probe", lambda v, b, t, backend: (None, {})) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert len(kept) == 4 and dropped == [] + + +# --- the probe has to be running the deployment's kernels ------------------- + + +def test_a_probe_on_different_kernels_prunes_nothing(config, enabled, session, monkeypatch): + """The failure this exists for, and the one that is expensive to miss. + + Measured on gpt-oss-120b/MI355X at the same TP=8, model and flags: the server + resolves ROCM_AITER_UNIFIED_ATTN with a Triton MXFP4 MoE, the offline probe + ROCM_AITER_FA with an AITER one. Read across that gap the screen is not + noisy but confidently wrong: it ranks a stack the deployment never runs. + """ + monkeypatch.setattr( + _explore_screen, + "_probe", + fake_probe( + {"v0": 99.0, "v1": 99.0, "v2": 99.0, "v3": 99.0}, + kernels={"attention": "ROCM_AITER_FA", "moe": "AITER_MXFP4_BF16"}, + ), + ) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert len(kept) == 4 and dropped == [] + + +def test_one_matching_kernel_is_not_enough(config, enabled, session, monkeypatch): + """Pinning attention alone left the same lever reading 65% worse.""" + monkeypatch.setattr( + _explore_screen, + "_probe", + fake_probe( + {"v0": 99.0, "v1": 9.0, "v2": 9.0, "v3": 9.0}, + kernels={"attention": "ROCM_AITER_UNIFIED_ATTN", "moe": "AITER_MXFP4_BF16"}, + ), + ) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert len(kept) == 4 and dropped == [] + + +def test_no_pruning_when_nothing_says_what_the_deployment_runs(config, enabled, tmp_path, monkeypatch): + """An unverifiable regime is not a matching one.""" + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 99.0, "v1": 9.0, "v2": 9.0, "v3": 9.0})) + kept, dropped = screen_variants(variants(), config, session_dir=tmp_path / "empty") + assert len(kept) == 4 and dropped == [] + + +def test_target_kernels_come_from_a_benchmark_already_paid_for(session): + """EXPLORE boots the stack before it proposes anything; the answer is on disk.""" + assert _explore_screen._target_kernels(session) == MATCHING_KERNELS + + +def test_kernels_are_read_from_either_thing_vllm_logs(): + assert kernels_from_log("Overriding with ROCM_AITER_FA out of potential backends") == {"attention": "ROCM_AITER_FA"} + assert kernels_from_log( + "Using TRITON_ATTN backend (selected via --attention-backend).\nUsing 'TRITON' Mxfp4 MoE backend." + ) == {"attention": "TRITON_ATTN", "moe": "TRITON"} + assert kernels_from_log("nothing about kernels here") == {} + + +def test_target_backend_is_read_from_the_deployments_own_flags(tmp_path, enabled, monkeypatch): + monkeypatch.delenv(ENV_BACKEND, raising=False) + path = tmp_path / "pinned.yaml" + path.write_text(CONFIG + " EXTRA_VLLM_ARGS: --attention-backend TRITON_ATTN\n") + with open(path) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + assert _explore_screen._target_backend(bench, None) == "TRITON_ATTN" + + +def test_the_pin_precedes_the_variants_own_flags(config, enabled): + """A variant testing a backend must override the pin, not be overridden by it.""" + with open(config) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + variant = GridVariant(name="v", extra_server_args="--attention-backend TRITON_ATTN") + cmd = _explore_screen._probe_command(variant, bench, "/tmp/o.json", "ROCM_AITER_UNIFIED_ATTN") + args = next(c for c in cmd if c.startswith("--server-args=")) + assert args.endswith("--attention-backend TRITON_ATTN") + + +# --- the screen cuts only the decisive losers ------------------------------- + + +def test_cuts_only_what_is_beyond_the_margin(config, enabled, session, monkeypatch): + # Baseline 10ms, margin 10%: 11.0ms survives, 11.1ms does not. + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 14.0, "v1": 9.0, "v2": 11.0, "v3": 11.1})) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert [v.name for v in kept] == ["v1", "v2"] + assert {d["name"] for d in dropped} == {"v0", "v3"} + assert all(d["reason"] == "screen_decisively_slower" for d in dropped) + + +def test_a_grid_of_near_ties_is_forwarded_whole(config, enabled, session, monkeypatch): + """Differences the screen cannot resolve are left to the real benchmark.""" + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 10.4, "v1": 9.7, "v2": 10.9, "v3": 10.1})) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert len(kept) == 4 and dropped == [] + + +def test_survivors_keep_the_grid_order(config, enabled, session, monkeypatch): + """The screen prunes; it does not get to decide what EXPLORE tries first.""" + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 10.5, "v1": 9.0, "v2": 20.0, "v3": 9.5})) + kept, _ = screen_variants(variants(), config, session_dir=session) + assert [v.name for v in kept] == ["v0", "v1", "v3"] + + +def test_margin_is_configurable(config, enabled, session, monkeypatch): + """A wider margin trusts the screen less: v3, cut at the default, survives.""" + monkeypatch.setenv(ENV_MARGIN_PCT, "30") + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 14.0, "v1": 9.0, "v2": 11.0, "v3": 11.1})) + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert [v.name for v in kept] == ["v1", "v2", "v3"] + assert [d["name"] for d in dropped] == ["v0"] + + +def test_nonsense_margin_falls_back_to_the_measured_one(config, enabled, session, monkeypatch): + monkeypatch.setenv(ENV_MARGIN_PCT, "not-a-number") + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 14.0, "v1": 9.0, "v2": 11.0, "v3": 10.5})) + kept, _ = screen_variants(variants(), config, session_dir=session) + assert [v.name for v in kept] == ["v1", "v2", "v3"] + + +def test_unreadable_variant_is_kept_not_dropped(config, enabled, session, monkeypatch): + """A probe that fails for one variant must not decide against it.""" + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 90.0, "v1": 3.0, "v2": 7.0})) # v3 unreadable + kept, dropped = screen_variants(variants(), config, session_dir=session) + assert "v3" in {v.name for v in kept} + assert "v3" not in {d["name"] for d in dropped} + + +def test_dropped_records_carry_the_measurement(config, enabled, session, monkeypatch): + monkeypatch.setattr(_explore_screen, "_probe", fake_probe({"v0": 14.0, "v1": 9.0, "v2": 7.0, "v3": 1.0})) + _, dropped = screen_variants(variants(), config, session_dir=session) + assert dropped[0]["detail"] == "probe decode 14.000 ms vs baseline 10.000 ms (+40%)" + + +# --- the probe carries the variant's own levers ----------------------------- + + +def test_probe_command_applies_variant_flags_and_env(config, enabled, tmp_path): + with open(config) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + variant = GridVariant( + name="aiter_off", + extra_server_args="--max-num-seqs 512 --enable-chunked-prefill", + extra_envs={"VLLM_ROCM_USE_AITER": "0"}, + ) + cmd = _explore_screen._probe_command(variant, bench, "/tmp/out.json", None) + assert "--server-args=--max-num-seqs 512 --enable-chunked-prefill" in cmd + assert "VLLM_ROCM_USE_AITER=0" in cmd + # The probe runs at the deployment's target TP but on fewer GPUs, which is + # where the saving comes from. + assert cmd[cmd.index("--tp") + 1] == "8" + assert cmd[cmd.index("--benchmark-gpus") + 1] == "1" + assert cmd[cmd.index("--batches") + 1] == "32" + + +def test_probe_gpu_count_and_depth_are_configurable(config, enabled, monkeypatch): + monkeypatch.setenv(ENV_GPUS, "2") + monkeypatch.setenv(ENV_LAYERS, "8") + with open(config) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + cmd = _explore_screen._probe_command(GridVariant(name="v"), bench, "/tmp/o.json", None) + assert cmd[cmd.index("--benchmark-gpus") + 1] == "2" + assert cmd[cmd.index("--num-hidden-layers") + 1] == "8" + + +def test_probe_never_asks_for_more_gpus_than_the_target_has(config, enabled, monkeypatch): + monkeypatch.setenv(ENV_GPUS, "16") + with open(config) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + cmd = _explore_screen._probe_command(GridVariant(name="v"), bench, "/tmp/o.json", None) + assert cmd[cmd.index("--benchmark-gpus") + 1] == "8" + + +def test_probe_reads_decode_and_kernels_from_its_own_run(config, enabled, tmp_path, monkeypatch): + """The real _probe, with the subprocess replaced by a canned run.""" + artifact = {"sweep": [{"batch": 32, "decode_ms": 4.25}]} + + class Done: + returncode = 0 + stdout = KERNEL_LOG + stderr = "" + + def fake_run(cmd, **kwargs): + save_path = cmd[cmd.index("--save") + 1] + with open(save_path, "w") as fh: + json.dump(artifact, fh) + return Done() + + monkeypatch.setattr(_explore_screen.subprocess, "run", fake_run) + with open(config) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + reading, kernels = _explore_screen._probe(GridVariant(name="v"), bench, 60, None) + assert reading == pytest.approx(4.25) + assert kernels == MATCHING_KERNELS + + +def test_probe_failure_returns_no_reading(config, enabled, monkeypatch): + class Failed: + returncode = 1 + stdout = "" + stderr = "boom" + + monkeypatch.setattr(_explore_screen.subprocess, "run", lambda cmd, **kw: Failed()) + with open(config) as fh: + import yaml + + bench = yaml.safe_load(fh)["benchmark"] + reading, _ = _explore_screen._probe(GridVariant(name="v"), bench, 60, None) + assert reading is None diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_settled_dedup.py b/src/hyperloom/inference_optimizer/tests/test_explore_settled_dedup.py new file mode 100644 index 0000000000..542835fd57 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_explore_settled_dedup.py @@ -0,0 +1,111 @@ +"""The cross-round dedup guard must never cost the search an optimization. + +Skipping a variant the ledger has already settled saves a server boot and a full +benchmark. It is also the one kind of pruning that can silently make Hyperloom +worse, because a variant that lost on its own can win once something it composes +with has landed. So most of these tests are about what the guard refuses to skip: +anything whose stack moved, whose workload moved, that never reached a verdict, +or that landed close enough to the KEEP threshold that noise could have decided +it. +""" + +from __future__ import annotations + +from hyperloom.orchestrator.actions.executors.explore import ( + REPEAT_NOISE_PCT, + SETTLED_MARGIN, + _settled_against_same_stack, +) + +WS = "workload-sig-abc" +BASE = 15235.83 +THRESH = 1.0 + +# Distance from the threshold beyond which a repeat cannot plausibly flip. +BAND = REPEAT_NOISE_PCT * SETTLED_MARGIN + + +def _prior(**over): + """A conclusive prior result, far below the KEEP threshold.""" + d = { + "workload_signature": WS, + "status": "succeeded", + "base_tput": BASE, + "gain_pct": THRESH - BAND - 0.5, + "outcome": "REVERT", + "round_id": "explore-001", + } + d.update(over) + return d + + +def test_settled_repeat_is_skipped(): + """Same config, same stack, same workload, verdict nowhere near the line.""" + assert _settled_against_same_stack(_prior(), WS, BASE, THRESH) is True + + +def test_a_moved_stack_is_retested(): + """A KEEP advanced the running baseline, so the variant may compose + differently now and has to be measured again.""" + assert _settled_against_same_stack(_prior(), WS, BASE * 1.04, THRESH) is False + + +def test_a_drifted_baseline_is_retested(): + """A baseline that merely drifted on re-measurement is indistinguishable + from a moved stack, so it falls the safe way.""" + assert _settled_against_same_stack(_prior(), WS, BASE + 0.01, THRESH) is False + + +def test_a_different_workload_is_retested(): + assert _settled_against_same_stack(_prior(), "other-sig", BASE, THRESH) is False + + +def test_a_variant_that_never_got_a_verdict_is_retested(): + """Crashed, killed on overtime, or failed at warmup: the ledger holds no + result to reuse, only the fact that it did not finish.""" + for status in ("failed", "killed_overtime", ""): + assert _settled_against_same_stack(_prior(status=status, gain_pct=None), WS, BASE, THRESH) is False + + +def test_a_result_near_the_threshold_is_retested(): + """Within a couple of noise envelopes of the KEEP line, the verdict may have + been decided by the measurement rather than by the variant. That deserves a + second sample, not a skip.""" + for gain in (THRESH - BAND * 0.5, THRESH + BAND * 0.5, THRESH, THRESH - BAND): + assert _settled_against_same_stack(_prior(gain_pct=gain), WS, BASE, THRESH) is False + + +def test_a_clear_winner_is_also_settled(): + """The guard is symmetric: a variant that won by a wide margin is as settled + as one that lost by one, and re-running it re-derives a known number.""" + assert _settled_against_same_stack(_prior(gain_pct=THRESH + BAND + 5.0, outcome="KEEP"), WS, BASE, THRESH) is True + + +def test_missing_fields_are_retested(): + for missing in ("base_tput", "gain_pct"): + assert _settled_against_same_stack(_prior(**{missing: None}), WS, BASE, THRESH) is False + + +def test_no_baseline_yet_is_retested(): + assert _settled_against_same_stack(_prior(), WS, 0.0, THRESH) is False + + +def test_only_the_decisive_real_session_variants_are_settled(): + """The audited session tested six variants against a stack that never moved, + and their gains were clustered just under the 1% line. Only the two that lost + by more than the noise band are settled; the three that landed inside it are + re-measured, which is the guard declining to prune on a number it cannot + distinguish from noise.""" + settled = {"chunked-prefill-8192": -0.812, "sched-conservativeness-03": -0.157} + inside_band = {"attn-aiter": 0.045, "cuda-graph-max-bs-256": 0.241, "mem-frac-092": 0.165} + for name, g in settled.items(): + assert _settled_against_same_stack(_prior(gain_pct=g), WS, BASE, THRESH) is True, name + for name, g in inside_band.items(): + assert _settled_against_same_stack(_prior(gain_pct=g), WS, BASE, THRESH) is False, name + + +def test_the_variant_that_failed_at_warmup_is_not_skipped(): + """One of the six never produced a number at all. It is not settled.""" + assert ( + _settled_against_same_stack(_prior(status="failed", gain_pct=None, outcome="REVERT"), WS, BASE, THRESH) is False + ) diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_short_warmup.py b/src/hyperloom/inference_optimizer/tests/test_explore_short_warmup.py new file mode 100644 index 0000000000..210fb5eb1b --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_explore_short_warmup.py @@ -0,0 +1,109 @@ +"""The warmup round has to warm, not measure. + +Every explore variant pays for two full benchmarks. The first one boots the +server, runs the accuracy gate, and has its throughput read only for +success/failure before being discarded -- yet it runs at the same length as the +round that is kept, which ``_workload_envs`` sizes at five to ten waves of the +concurrency. + +Cutting it to one wave is the largest safe saving in an explore round, so these +tests pin the two things that make it safe: it must be short enough to matter, +and it must fall back to the long warmup rather than to a broken one whenever +the concurrency cannot be established. +""" + +from __future__ import annotations + +import textwrap + +import pytest + +from hyperloom.orchestrator.actions.executors.explore import ( + WARMUP_WAVES, + _short_warmup_enabled, + _warmup_num_prompts, +) + + +def _cfg(tmp_path, body: str): + p = tmp_path / "bench.yaml" + p.write_text(textwrap.dedent(body), encoding="utf-8") + return p + + +def test_warmup_is_one_wave_of_the_concurrency(tmp_path): + cfg = _cfg( + tmp_path, + """ + benchmark: + envs: + TP: 8 + CONC: 64 + ISL: 1024 + OSL: 1024 + """, + ) + assert _warmup_num_prompts(cfg) == 64 * WARMUP_WAVES + + +def test_warmup_is_shorter_than_the_measured_round(tmp_path): + """A measured round at ISL+OSL=2048 is CONC*5; the warmup must be well under.""" + cfg = _cfg( + tmp_path, + """ + benchmark: + envs: + CONC: 64 + ISL: 1024 + OSL: 1024 + """, + ) + measured_num_prompts = 64 * 5 + assert _warmup_num_prompts(cfg) < measured_num_prompts + + +@pytest.mark.parametrize( + "body", + [ + # No concurrency key at all. + "benchmark:\n envs:\n TP: 8\n", + # Concurrency present but unusable. + "benchmark:\n envs:\n CONC: 0\n", + "benchmark:\n envs:\n CONC: 'not-a-number'\n", + # No benchmark section. + "something_else: 1\n", + # Empty file. + "", + ], +) +def test_unreadable_concurrency_keeps_the_long_warmup(tmp_path, body): + """Falling back must mean the warmup we already run, never a shorter one.""" + assert _warmup_num_prompts(_cfg(tmp_path, body)) is None + + +def test_missing_file_keeps_the_long_warmup(tmp_path): + assert _warmup_num_prompts(tmp_path / "does-not-exist.yaml") is None + + +def test_short_warmup_is_on_by_default(monkeypatch): + monkeypatch.delenv("INFERENCE_OPTIMIZER_EXPLORE_SHORT_WARMUP", raising=False) + assert _short_warmup_enabled() is True + + +@pytest.mark.parametrize("val", ["0", "false", "no", "off", "FALSE", "Off"]) +def test_short_warmup_can_be_turned_off(monkeypatch, val): + monkeypatch.setenv("INFERENCE_OPTIMIZER_EXPLORE_SHORT_WARMUP", val) + assert _short_warmup_enabled() is False + + +@pytest.mark.parametrize("val", ["1", "true", "yes", "on"]) +def test_short_warmup_stays_on_for_truthy_values(monkeypatch, val): + monkeypatch.setenv("INFERENCE_OPTIMIZER_EXPLORE_SHORT_WARMUP", val) + assert _short_warmup_enabled() is True + + +def test_warmup_never_shorter_than_one_full_batch(tmp_path): + """Under-filling the batch would warm a different shape than we measure.""" + for conc in (1, 8, 64, 256, 1024): + cfg = _cfg(tmp_path, f"benchmark:\n envs:\n CONC: {conc}\n") + assert _warmup_num_prompts(cfg) >= conc diff --git a/src/hyperloom/inference_optimizer/tests/test_extrapolation_notes.py b/src/hyperloom/inference_optimizer/tests/test_extrapolation_notes.py new file mode 100644 index 0000000000..fedc827d33 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_extrapolation_notes.py @@ -0,0 +1,65 @@ +"""The projection has to say when it is working outside its evidence. + +A wrong number that announces itself is a different class of problem from a +wrong number that does not, and the search consumes these silently. Context is +the case that matters: Hyperloom sweeps it to 65,536 while nothing validated the +model past about 1,500, so crossing the boundary is routine rather than exotic. +""" + +from __future__ import annotations + +import pytest + +from hyperloom.orchestrator.actions.executors.inferasim_bridge import ( + SINGLE_NODE_GPUS, + VALIDATED_CONTEXT_TOKENS, + ServingSpec, + extrapolation_notes, +) + + +def spec(isl: int = 1024, osl: int = 128, **kw) -> ServingSpec: + return ServingSpec(framework="vllm", model_path="/models/gpt-oss-120b", isl=isl, osl=osl, **kw) + + +def test_inside_the_validated_box_says_nothing(): + notes = extrapolation_notes(spec(isl=1024, osl=128), replica_gpus=8, calibrated=True) + assert notes == [] + + +def test_context_past_the_validated_range_is_flagged(): + notes = extrapolation_notes(spec(isl=65536, osl=4096), replica_gpus=8, calibrated=True) + assert any("context" in n for n in notes) + assert any(str(65536 + 4096) in n for n in notes) + + +def test_the_boundary_itself_is_not_flagged(): + """Exactly at the validated edge is still inside it.""" + notes = extrapolation_notes(spec(isl=VALIDATED_CONTEXT_TOKENS, osl=0), replica_gpus=8, calibrated=True) + assert not any("context" in n for n in notes) + + +def test_context_counts_what_the_run_will_hold_not_just_the_prompt(): + """A short prompt decoding for a long time still ends up at long context.""" + notes = extrapolation_notes(spec(isl=512, osl=131072), replica_gpus=8, calibrated=True) + assert any("context" in n for n in notes) + + +def test_crossing_a_node_boundary_is_flagged(): + notes = extrapolation_notes(spec(), replica_gpus=SINGLE_NODE_GPUS * 2, calibrated=True) + assert any("nodes" in n for n in notes) + + +def test_uncalibrated_is_flagged(): + notes = extrapolation_notes(spec(), replica_gpus=8, calibrated=False) + assert any("simulation" in n for n in notes) + + +def test_notes_accumulate_rather_than_shadowing_each_other(): + notes = extrapolation_notes(spec(isl=131072, osl=1024), replica_gpus=32, calibrated=False) + assert len(notes) == 3 + + +@pytest.mark.parametrize("isl,osl", [(0, 0), (None, None)]) +def test_missing_lengths_do_not_raise(isl, osl): + assert extrapolation_notes(spec(isl=isl, osl=osl), replica_gpus=8, calibrated=True) == [] diff --git a/src/hyperloom/inference_optimizer/tests/test_grid_runner_kill_and_branches_unit.py b/src/hyperloom/inference_optimizer/tests/test_grid_runner_kill_and_branches_unit.py index b5d57fbf3f..d14c262295 100644 --- a/src/hyperloom/inference_optimizer/tests/test_grid_runner_kill_and_branches_unit.py +++ b/src/hyperloom/inference_optimizer/tests/test_grid_runner_kill_and_branches_unit.py @@ -39,6 +39,19 @@ def test_kill_stale_servers_noop_in_multi_node(monkeypatch): assert slept == [] +def test_kill_stale_servers_noop_against_external_server(monkeypatch): + """The engine behind an external endpoint must survive the pre-run reap. + + Its cmdline matches the vLLM/SGLang kill patterns, so without this gate + every Magpie invocation would restart the server it is measuring. + """ + monkeypatch.setenv("HYPERLOOM_MN_EXT_SERVICE_URL", "http://infera:8000") + slept: list = [] + monkeypatch.setattr("time.sleep", lambda *_a: slept.append(True)) + gr._kill_stale_servers() + assert slept == [] + + def _proc_open_factory( cmdlines: dict[str, bytes], maps: dict[str, str], diff --git a/src/hyperloom/inference_optimizer/tests/test_inferasim_backend.py b/src/hyperloom/inference_optimizer/tests/test_inferasim_backend.py new file mode 100644 index 0000000000..f23d246c40 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_inferasim_backend.py @@ -0,0 +1,485 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for the inferasim benchmark backend + projection bridge. + +No GPU and no Infera install: the projection call is monkeypatched, so these +verify backend selection, argv construction, benchmark-spec parsing, model +preset resolution, the metrics->report mapping, and that a simulated run flows +through Hyperloom's measurement extractor unchanged. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +import yaml + +from hyperloom.orchestrator.actions.executors import benchmark_backend as bb +from hyperloom.orchestrator.actions.executors import inferasim_bridge as ib +from hyperloom.orchestrator.actions.executors import inferasim_runner +from hyperloom.orchestrator.actions.executors.benchmark_result import ( + extract_benchmark_measurement, + is_valid_measurement, +) + + +def test_inferasim_backend_selected(monkeypatch): + monkeypatch.setenv(bb.BENCHMARK_BACKEND_ENV, "inferasim") + assert bb.resolve_backend_name() == "inferasim" + backend = bb.resolve_backend() + assert backend.name == "inferasim" + cmd = backend.build_command( + python_exe="PY", + config_path=Path("/cfg.yaml"), + output_dir=Path("/out"), + ) + assert cmd == [ + "PY", + "-m", + "hyperloom.orchestrator.actions.executors.inferasim_runner", + "benchmark", + "--benchmark-config", + "/cfg.yaml", + "--output-dir", + "/out", + "--run-mode", + "local", + ] + + +def test_inferasim_backend_lifecycle_ineligible(): + backend = bb.InferasimBackend() + verdict = backend.lifecycle_eligibility({"framework": "sglang"}) + assert verdict is not None + assert verdict["eligible"] is False + + +def test_inferasim_interpreter_prefers_env(monkeypatch): + monkeypatch.setenv("HYPERLOOM_INFERASIM_PYTHON", "/opt/infera/bin/python") + assert bb.InferasimBackend().resolve_interpreter() == "/opt/infera/bin/python" + + +def test_spec_from_benchmark_parses_envs(monkeypatch): + monkeypatch.delenv(ib.ENV_EP, raising=False) + monkeypatch.delenv(ib.ENV_PP, raising=False) + bench = { + "framework": "vllm", + "model": "/models/Qwen-Qwen3-14B", + "precision": "fp8", + "envs": {"TP": 4, "CONC": 128, "ISL": 2048, "OSL": 256}, + } + spec = ib.spec_from_benchmark(bench) + assert spec.framework == "vllm" + assert spec.tp == 4 + assert spec.conc == 128 + assert spec.isl == 2048 + assert spec.osl == 256 + assert spec.weight_dtype == "fp8" + + +def test_spec_parses_ep_from_server_args(monkeypatch): + monkeypatch.delenv(ib.ENV_EP, raising=False) + bench = { + "framework": "sglang", + "model": "/models/gpt-oss-120b", + "envs": {"TP": 8, "EXTRA_SGLANG_ARGS": "--ep-size 8 --foo 1"}, + } + spec = ib.spec_from_benchmark(bench) + assert spec.ep == 8 + + +@pytest.mark.parametrize( + "flag,expected", + [ + ("--kv-cache-dtype fp8_e4m3", "fp8"), + ("--kv-cache-dtype fp8", "fp8"), + ("--kv-cache-dtype=fp8_e5m2", "fp8"), + ("--kv-cache-dtype auto", "bf16"), # auto follows the weights + ("--foo 1", "bf16"), # absent + ], +) +def test_spec_parses_kv_cache_dtype_from_server_args(monkeypatch, flag, expected): + """An explore grid changes the KV dtype by passing this flag and nothing else. + + The projection prices KV dtype perfectly well, so dropping the flag made a + lever the model *can* see look like one it cannot, and scored an fp8 + candidate as bf16 -- a candidate whose whole point is halving KV traffic. + """ + monkeypatch.delenv(ib.ENV_KV_DTYPE, raising=False) + spec = ib.spec_from_benchmark( + { + "framework": "vllm", + "model": "/models/gpt-oss-120b", + "envs": {"TP": 8, "EXTRA_VLLM_ARGS": flag}, + } + ) + assert spec.kv_cache_dtype == expected + + +def test_kv_dtype_env_overrides_the_server_arg(monkeypatch): + monkeypatch.setenv(ib.ENV_KV_DTYPE, "bf16") + spec = ib.spec_from_benchmark( + { + "framework": "vllm", + "model": "/models/gpt-oss-120b", + "envs": {"TP": 8, "EXTRA_VLLM_ARGS": "--kv-cache-dtype fp8_e4m3"}, + } + ) + assert spec.kv_cache_dtype == "bf16" + + +def test_max_num_seqs_caps_the_running_batch(monkeypatch): + """A scheduler cap below the offered load is the batch the step actually runs.""" + spec = ib.spec_from_benchmark( + { + "framework": "vllm", + "model": "/models/gpt-oss-120b", + "envs": {"TP": 8, "CONC": 64, "EXTRA_VLLM_ARGS": "--max-num-seqs 32"}, + } + ) + assert spec.conc == 32 + + +def test_max_num_seqs_above_the_load_changes_nothing(monkeypatch): + """Raising a cap nobody reaches is a no-op, and must not be reported as a win.""" + spec = ib.spec_from_benchmark( + { + "framework": "vllm", + "model": "/models/gpt-oss-120b", + "envs": {"TP": 8, "CONC": 64, "EXTRA_VLLM_ARGS": "--max-num-seqs 512"}, + } + ) + assert spec.conc == 64 + + +def test_resolve_preset_heuristics_and_override(monkeypatch): + monkeypatch.delenv(ib.ENV_MODEL, raising=False) + assert ib.resolve_preset("/models/gpt-oss-120b") == "gpt_oss_120B" + assert ib.resolve_preset("/data/Qwen-Qwen3-14B") == "qwen3_14B" + assert ib.resolve_preset("/some/unknown-model") is None + monkeypatch.setenv(ib.ENV_MODEL, "custom_preset") + assert ib.resolve_preset("/models/gpt-oss-120b") == "custom_preset" + + +def _fake_metrics() -> ib.ProjMetrics: + return ib.ProjMetrics( + output_throughput=9000.0, + request_throughput=8.78, + total_token_throughput=18000.0, + ttft_ms=25.0, + tpot_ms=6.5, + itl_ms=6.5, + e2el_ms=6650.0, + decode_tps_per_gpu=9000.0, + memory_per_gpu_gb=70.0, + max_concurrency=64, + calibrated=False, + replica_gpus=1, + ) + + +def test_raw_result_from_metrics_shape(): + spec = ib.ServingSpec(framework="sglang", model_path="/m", tp=1, conc=64, isl=1024, osl=1024) + raw = ib.raw_result_from_metrics(spec, _fake_metrics()) + assert raw["output_throughput"] == 9000.0 + assert raw["mean_ttft_ms"] == 25.0 + assert raw["mean_tpot_ms"] == 6.5 + assert raw["mean_e2el_ms"] == 6650.0 + assert raw["total_output_tokens"] == 64 * 1024 + assert raw["inferasim_decode_tps_per_gpu"] == 9000.0 + + +def _write_bench(tmp_path: Path) -> Path: + cfg = { + "benchmark": { + "framework": "sglang", + "model": "/models/gpt-oss-120b", + "precision": "bf16", + "run_mode": "local", + "envs": {"TP": 1, "CONC": 64, "ISL": 1024, "OSL": 1024}, + } + } + path = tmp_path / "bench.yaml" + path.write_text(yaml.safe_dump(cfg), encoding="utf-8") + return path + + +def test_runner_end_to_end_with_mocked_projection(tmp_path, monkeypatch): + """The runner writes a Magpie-compatible report from projected metrics.""" + monkeypatch.setattr(ib, "project", lambda spec: _fake_metrics()) + + cfg_path = _write_bench(tmp_path) + rc = inferasim_runner.run_benchmark(cfg_path, tmp_path / "out") + assert rc == 0 + + workspaces = list((tmp_path / "out").glob("benchmark_sglang_*")) + assert len(workspaces) == 1 + ws = workspaces[0] + report = json.loads((ws / "benchmark_report.json").read_text(encoding="utf-8")) + assert report["success"] is True + assert report["bypass_analysis"]["backend"] == "inferasim" + + m = extract_benchmark_measurement(report, workspace=ws) + assert is_valid_measurement(m) is True + assert m["output_throughput"] == 9000.0 + assert m["ttft_mean_ms"] == 25.0 + + +def test_runner_projection_failure_emits_failed_report(tmp_path, monkeypatch): + def boom(spec): + raise ib.InferasimBridgeError("no preset resolvable") + + monkeypatch.setattr(ib, "project", boom) + cfg_path = _write_bench(tmp_path) + rc = inferasim_runner.run_benchmark(cfg_path, tmp_path / "out") + assert rc == 1 + + ws = sorted((tmp_path / "out").glob("benchmark_sglang_*"))[-1] + report = json.loads((ws / "benchmark_report.json").read_text(encoding="utf-8")) + assert report["success"] is False + assert any("no preset resolvable" in e for e in report["errors"]) + + +def test_runner_cli_rejects_non_local(tmp_path): + rc = inferasim_runner.main( + [ + "benchmark", + "--benchmark-config", + str(tmp_path / "c.yaml"), + "--output-dir", + str(tmp_path / "o"), + "--run-mode", + "docker", + ] + ) + assert rc == 2 + + +def test_runner_server_phase_is_noop_success(tmp_path): + rc = inferasim_runner.main( + [ + "benchmark", + "--benchmark-config", + str(tmp_path / "c.yaml"), + "--output-dir", + str(tmp_path / "o"), + "--phase", + "server", + ] + ) + assert rc == 0 + + +def test_resolve_workload_prefers_explicit_env(tmp_path, monkeypatch): + wl = tmp_path / "custom_workload.yaml" + wl.write_text("work_group: t\n", encoding="utf-8") + monkeypatch.setenv(ib.ENV_WORKLOAD, str(wl)) + spec = ib.ServingSpec(framework="sglang", model_path="/m") + workload, extra_env = ib._resolve_workload_and_env(spec) + assert workload == str(wl.resolve()) + assert "INFERASIM_MODEL" not in extra_env + + +def test_resolve_workload_uses_template_for_preset(monkeypatch): + monkeypatch.delenv(ib.ENV_WORKLOAD, raising=False) + monkeypatch.setenv(ib.ENV_MODEL, "gpt_oss_120B") + spec = ib.ServingSpec(framework="sglang", model_path="/models/gpt-oss-120b") + workload, extra_env = ib._resolve_workload_and_env(spec) + assert Path(workload).name == "inferasim_workload.yaml" + assert extra_env["INFERASIM_MODEL"] == "gpt_oss_120B" + + +def _write_anchor( + path: Path, *, model: str, real_weights: bool, decode_ms: float, quant=None, kv="bf16", aiter=True +) -> None: + """Minimal benchmark artifact in the shape benchmark_vllm.py emits.""" + path.write_text( + json.dumps( + { + "backend": "vllm", + "measured": {"model": {"prefill_ms": 10.0, "decode_ms": decode_ms}}, + "sweep": [{"batch": 16, "prefill_ms": 10.0, "decode_ms": decode_ms}], + "meta": { + "model": model, + "batch": 16, + "input_len": 1024, + "tp": 1, + "quantization": quant, + "kv_cache_dtype": kv, + "use_aiter": aiter, + "real_weights": real_weights, + "load_format": "auto" if real_weights else "dummy", + }, + } + ), + encoding="utf-8", + ) + + +def test_anchor_is_real_weights_detection(tmp_path): + real, dummy = tmp_path / "r.json", tmp_path / "d.json" + _write_anchor(real, model="m", real_weights=True, decode_ms=9.0) + _write_anchor(dummy, model="m", real_weights=False, decode_ms=5.0) + assert ib._anchor_is_real_weights(str(real)) is True + assert ib._anchor_is_real_weights(str(dummy)) is False + assert ib._anchor_is_real_weights(str(tmp_path / "missing.json")) is False + + +def test_recipe_from_spec_extracts_attention_backend(): + spec = ib.ServingSpec( + framework="sglang", + model_path="/models/x", + extra_server_args="--attention-backend aiter --max-num-seqs 64", + ) + recipe = ib.recipe_from_spec(spec) + assert recipe["attention_backend"] == "aiter" + assert recipe["weight_dtype"] == "bf16" + + +def test_select_anchor_prefers_explicit_env(tmp_path, monkeypatch): + a = tmp_path / "explicit.json" + _write_anchor(a, model="m", real_weights=True, decode_ms=9.0) + monkeypatch.setenv(ib.ENV_ANCHOR, str(a)) + choice = ib.select_anchor(ib.ServingSpec(framework="vllm", model_path="m")) + assert choice is not None + assert choice.path == str(a) + assert choice.regime_distance == 0 + + +def test_select_anchor_none_without_store(monkeypatch): + monkeypatch.delenv(ib.ENV_ANCHOR, raising=False) + monkeypatch.delenv(ib.ENV_ANCHOR_STORE, raising=False) + assert ib.select_anchor(ib.ServingSpec(framework="vllm", model_path="m")) is None + + +def _write_curve(path: Path, points: list[tuple[int, float]]) -> None: + """Artifact carrying an explicit decode-vs-batch curve.""" + path.write_text( + json.dumps( + { + "backend": "vllm", + "sweep": [{"batch": b, "prefill_ms": 10.0, "decode_ms": d} for b, d in points], + "meta": {"model": "m", "tp": 1, "input_len": 1024}, + } + ) + ) + + +@pytest.mark.parametrize( + "points, sane", + [ + ([(1, 4.0), (8, 6.4), (32, 12.0)], True), # ordinary rising curve + ([(16, 12.0)], True), # single point: narrow, valid + ([(8, 6.0), (16, 5.7)], True), # -5%: run-to-run noise + ([(16, 16.3), (64, 1.4)], False), # differencing degenerated + ([(4, 11.6), (32, 9.5)], False), # decode faster at 8x batch + ([(8, 0.0)], False), # non-positive timing + ([], False), # nothing measured + ], +) +def test_anchor_curve_sanity_gate(tmp_path, points, sane): + p = tmp_path / "curve.json" + _write_curve(p, points) + assert ib.anchor_curve_is_sane(str(p)) is sane + + +def test_anchor_curve_sanity_gate_missing_file(tmp_path): + assert ib.anchor_curve_is_sane(str(tmp_path / "nope.json")) is False + + +@pytest.mark.parametrize("payload", ["[]", '[{"batch": 1}]', '"text"', "12"]) +def test_anchor_curve_sanity_gate_rejects_non_object_json(tmp_path, payload): + """A JSON file that is not an artifact is rejected, not a crash. + + The anchor store sits next to analysis output, so the gate is pointed at + whatever JSON is on disk; a list or scalar must not raise past the caller. + """ + art = tmp_path / "not_an_artifact.json" + art.write_text(payload) + assert ib.anchor_curve_is_sane(str(art)) is False + + +@pytest.mark.parametrize( + "args, expected", + [ + ("", (None, 0)), + ("--attention-backend triton", (None, 0)), + ('--speculative-config \'{"method": "deepseek_mtp", "num_speculative_tokens": 3}\'', ("deepseek_mtp", 3)), + ("--speculative-algorithm NEXTN --speculative-num-steps 3", ("NEXTN", 3)), + ("--speculative-algorithm EAGLE3", ("EAGLE3", 1)), + ("--method mtp --num-speculative-tokens 3", ("mtp", 3)), + ("--method fp8", (None, 0)), + ], +) +def test_parse_speculative_across_frameworks(args, expected): + assert ib.parse_speculative(args) == expected + + +def test_recipe_marks_speculative_candidates_apart(): + """A speculating candidate must not share a regime with a plain one. + + Speculation changes how many tokens a step emits, so reusing a + non-speculative anchor for it silently under-predicts throughput. + """ + plain = ib.recipe_from_spec(ib.ServingSpec(framework="vllm", model_path="m")) + mtp = ib.recipe_from_spec( + ib.ServingSpec( + framework="atom", + model_path="m", + extra_server_args="--method mtp --num-speculative-tokens 3", + ) + ) + assert plain["speculative"] == "off" + assert mtp["speculative"] == "spec:3" + assert plain["speculative"] != mtp["speculative"] + + +def test_recipe_names_the_serving_engine(): + """Two engines serving one checkpoint are two regimes, not one. + + InferaSim scores an axis as matching when it is absent on either side, so + leaving the engine off the recipe lets a vLLM anchor price an SGLang run. + """ + engines = ("vllm", "sglang", "atom") + recipes = [ib.recipe_from_spec(ib.ServingSpec(framework=fw, model_path="m")) for fw in engines] + assert [r["engine"] for r in recipes] == list(engines) + assert len({r["engine"] for r in recipes}) == len(engines) + + +def test_argv_names_the_serving_engine(): + """Tell InferaSim the engine too, so it can refuse a cross-engine anchor. + + Nothing in the analytical model reads it, so the projected number does not + move; it decides which measured anchor calibration may read. + """ + for fw in ("vllm", "sglang", "atom"): + argv = ib._build_argv(ib.ServingSpec(framework=fw, model_path="m"), "w.yaml") + assert argv[argv.index("--serving-engine") + 1] == fw + + +def test_argv_omits_the_engine_when_unknown(): + """An unnamed framework leaves the axis unset rather than emitting a blank.""" + argv = ib._build_argv(ib.ServingSpec(framework="", model_path="m"), "w.yaml") + assert "--serving-engine" not in argv + + +def test_select_anchor_rejects_insane_anchor(tmp_path, monkeypatch): + """A corrupt curve is worse than no anchor: fall back to pure analysis. + + Applies even to an operator-pinned anchor, which is the path most likely to + point at a hand-picked artifact nobody re-validated. + """ + bad = tmp_path / "bad.json" + _write_curve(bad, [(16, 16.3), (64, 1.4)]) + monkeypatch.setenv(ib.ENV_ANCHOR, str(bad)) + monkeypatch.delenv(ib.ENV_ANCHOR_STORE, raising=False) + assert ib.select_anchor(ib.ServingSpec(framework="vllm", model_path="m")) is None + + good = tmp_path / "good.json" + _write_curve(good, [(16, 12.0), (64, 20.0)]) + monkeypatch.setenv(ib.ENV_ANCHOR, str(good)) + choice = ib.select_anchor(ib.ServingSpec(framework="vllm", model_path="m")) + assert choice is not None and choice.path == str(good) diff --git a/src/hyperloom/inference_optimizer/tests/test_kernel_amdahl_gate.py b/src/hyperloom/inference_optimizer/tests/test_kernel_amdahl_gate.py new file mode 100644 index 0000000000..4c7629572a --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_kernel_amdahl_gate.py @@ -0,0 +1,139 @@ +"""An integrate run that cannot clear its bar is GPU time spent on a known answer. + +A kernel worth ``f`` of GPU time, made ``S`` times faster, cannot make the whole +deployment more than ``1/((1-f) + f/S)`` faster. When that ceiling is under the +threshold the integrate run has to beat, the run is arithmetically incapable of +passing -- and it is a full end-to-end serving benchmark. + +Skipping it is the rare kind of pruning that cannot cost an optimization, but +only if the bound is applied in the safe direction. So these tests are mostly +about the cases where the gate must decline to fire: missing inputs, unusable +inputs, and anything close enough to the bar that the trace's own error in +``gpu_pct`` could account for the gap. +""" + +from __future__ import annotations + +import pytest + +from hyperloom.orchestrator.kernel._kernel_decisions import ( + AMDAHL_GPU_PCT_MARGIN, + _amdahl_gate_enabled, + amdahl_e2e_ceiling_pct, + integrate_cannot_pass, +) + + +# ── the arithmetic ──────────────────────────────────────────────────────────── + + +def test_ceiling_matches_amdahl(): + """10% of GPU time made 2x faster caps the whole thing at 1/(0.9+0.05).""" + got = amdahl_e2e_ceiling_pct(10.0, 2.0) + assert got == pytest.approx((1.0 / 0.95 - 1.0) * 100.0, rel=1e-9) + + +def test_a_kernel_that_is_everything_inherits_its_own_speedup(): + assert amdahl_e2e_ceiling_pct(100.0, 1.5) == pytest.approx(50.0, rel=1e-9) + + +def test_no_speedup_is_no_gain(): + assert amdahl_e2e_ceiling_pct(30.0, 1.0) == pytest.approx(0.0, abs=1e-9) + + +def test_ceiling_rises_with_share_and_with_speedup(): + assert amdahl_e2e_ceiling_pct(20.0, 1.5) > amdahl_e2e_ceiling_pct(10.0, 1.5) + assert amdahl_e2e_ceiling_pct(10.0, 2.0) > amdahl_e2e_ceiling_pct(10.0, 1.5) + + +@pytest.mark.parametrize( + "pct,spd", + [ + (0.0, 1.5), + (-5.0, 1.5), + (101.0, 1.5), + (10.0, 0.0), + (10.0, -1.0), + (None, 1.5), + (10.0, None), + ("x", 1.5), + (10.0, "x"), + ], +) +def test_unusable_inputs_give_no_ceiling(pct, spd): + assert amdahl_e2e_ceiling_pct(pct, spd) is None + + +# ── the gate ────────────────────────────────────────────────────────────────── + + +def test_the_documented_dead_zone_is_caught(): + """The minimum-share, minimum-speedup kernel cannot clear a 1% bar. + + A kernel at the 10% dispatch floor that just clears the 1.10x micro bar + reaches 1/(0.9 + 0.1/1.1) = 0.92% end to end. Hyperloom would run a full + serving benchmark to find that out. + """ + assert amdahl_e2e_ceiling_pct(10.0, 1.10) < 1.0 + + +def test_gate_declines_when_the_margin_rescues_it(): + """At 10%/1.10x the raw ceiling fails, but the gate inflates the share first.""" + raw = amdahl_e2e_ceiling_pct(10.0, 1.10) + inflated = amdahl_e2e_ceiling_pct(10.0 * AMDAHL_GPU_PCT_MARGIN, 1.10) + assert raw < 1.0 < inflated + assert integrate_cannot_pass(10.0, 1.10, 1.0) is False + + +def test_gate_fires_only_when_even_the_inflated_share_fails(): + # 2% of GPU time at 1.10x: even at 3% the ceiling stays far under 1%. + assert integrate_cannot_pass(2.0, 1.10, 1.0) is True + + +def test_a_big_kernel_always_runs(): + assert integrate_cannot_pass(40.0, 1.5, 1.0) is False + + +def test_a_large_speedup_on_a_small_kernel_still_runs_when_it_could_pass(): + # 5% at 3x -> inflated to 7.5%, ceiling 5.4%, comfortably over the bar. + assert integrate_cannot_pass(5.0, 3.0, 1.0) is False + + +@pytest.mark.parametrize("pct,spd", [(0.0, 1.5), (None, 1.5), (10.0, None), (10.0, 0.0)]) +def test_gate_never_fires_on_inputs_it_cannot_read(pct, spd): + """Unknown inputs must mean 'run the benchmark', never 'skip it'.""" + assert integrate_cannot_pass(pct, spd, 1.0) is False + + +def test_gate_never_fires_without_a_threshold(): + assert integrate_cannot_pass(1.0, 1.01, 0.0) is False + assert integrate_cannot_pass(1.0, 1.01, -1.0) is False + + +def test_a_higher_bar_prunes_more(): + """Raising the threshold can only turn a run off, never on.""" + for thresh in (1.0, 2.0, 5.0): + prev = integrate_cannot_pass(8.0, 1.2, thresh) + assert prev in (True, False) + assert integrate_cannot_pass(8.0, 1.2, 5.0) is True + assert integrate_cannot_pass(8.0, 1.2, 1.0) is False + + +def test_gate_is_monotone_in_speedup(): + """A faster kernel must never be pruned when a slower one was kept.""" + assert integrate_cannot_pass(6.0, 1.05, 1.0) is True + assert integrate_cannot_pass(6.0, 1.60, 1.0) is False + + +# ── the switch ──────────────────────────────────────────────────────────────── + + +def test_gate_is_on_by_default(monkeypatch): + monkeypatch.delenv("HYPERLOOM_KERNEL_AMDAHL_GATE", raising=False) + assert _amdahl_gate_enabled() is True + + +@pytest.mark.parametrize("val", ["0", "false", "no", "off", "OFF"]) +def test_gate_can_be_turned_off(monkeypatch, val): + monkeypatch.setenv("HYPERLOOM_KERNEL_AMDAHL_GATE", val) + assert _amdahl_gate_enabled() is False diff --git a/src/hyperloom/inference_optimizer/tests/test_multi_node_scripts.py b/src/hyperloom/inference_optimizer/tests/test_multi_node_scripts.py index 5b6e9233ea..fc72925df4 100644 --- a/src/hyperloom/inference_optimizer/tests/test_multi_node_scripts.py +++ b/src/hyperloom/inference_optimizer/tests/test_multi_node_scripts.py @@ -1217,6 +1217,29 @@ def test_magpie_remote_env_empty_for_single_node(tmp_path, monkeypatch): assert mne.magpie_remote_env() == {} +def test_magpie_remote_env_external_single_node(tmp_path, monkeypatch): + """A single-node run targets a handed-over endpoint client-only. + + Magpie's client-only path is env-driven, not multi-node specific, so an + Infera-hosted server fronting the engine is reachable without pretending + the run spans two nodes. + """ + _write_mn_state(tmp_path, monkeypatch, {"backend": "rayjob", "nodes": 1}) + monkeypatch.setenv("INFERENCE_OPTIMIZER_NODES", "1") + monkeypatch.setattr(mne, "external_service_url", lambda: "http://infera:8000") + monkeypatch.setattr( + "hyperloom.orchestrator.actions.executors.benchmark_backend.resolve_benchmark_interpreter", + lambda: "/opt/venv/bin/python", + ) + + env = mne.magpie_remote_env() + + assert env["MAGPIE_RUN_PHASE"] == "client" + assert env["BENCHMARK_BASE_URL"] == "http://infera:8000" + assert env["MAGPIE_EVAL_PYTHON"] == "/opt/venv/bin/python" + + +# --------------------------------------------------------------------------- # _write_rayjob_meta sidecar JSON tests. diff --git a/src/hyperloom/inference_optimizer/tests/test_recover_executor.py b/src/hyperloom/inference_optimizer/tests/test_recover_executor.py index 6b5e262409..23644c0753 100644 --- a/src/hyperloom/inference_optimizer/tests/test_recover_executor.py +++ b/src/hyperloom/inference_optimizer/tests/test_recover_executor.py @@ -119,6 +119,29 @@ def _should_not_be_called(): assert out["killed_pids"] == [] +@pytest.mark.asyncio +async def test_external_server_skips_kill_stage(tmp_path, monkeypatch): + """An external endpoint's engine survives recovery even with cleanup forced. + + ``OWNER_PATTERNS`` match by cmdline, so a pgrep-driven kill would take down + a server we never launched and restart what the benchmark is measuring. + """ + workspace = tmp_path / "ws" + workspace.mkdir() + exe = RecoverExecutor() + + monkeypatch.setenv("HYPERLOOM_MN_EXT_SERVICE_URL", "http://infera:8000") + monkeypatch.setattr(exe, "_probe_gpu_free_mb", _healthy_probe) + calls: list[str] = [] + monkeypatch.setattr(exe, "_kill_stale_owners", lambda: calls.append("kill") or []) + + out = await exe(_ctx(workspace, params={"force_gpu_cleanup": True})) + + assert calls == [] + assert out["killed_pids"] == [] + assert out["state"] == "succeeded" + + @pytest.mark.asyncio async def test_kills_stale_owners_and_recovers(tmp_path, monkeypatch): workspace = tmp_path / "ws" diff --git a/src/hyperloom/orchestrator/actions/executors/_explore_screen.py b/src/hyperloom/orchestrator/actions/executors/_explore_screen.py new file mode 100644 index 0000000000..930f7090ce --- /dev/null +++ b/src/hyperloom/orchestrator/actions/executors/_explore_screen.py @@ -0,0 +1,359 @@ +"""Drop EXPLORE's hopeless variants on a cheap probe before benchmarking them. + +EXPLORE benchmarks every proposed variant on the deployment configuration, which +for a TP=8 target means eight GPUs held for the whole run. Most of those runs buy +nothing: the variant loses, and the only thing the benchmark had to establish was +that it loses. + +A screen answers that question far more cheaply by running the SAME engine with +the SAME flags on fewer GPUs. It is a real engine launch, not a projection, so +kernel-level levers -- attention backend, aiter, cudagraph -- actually execute +and actually show up; an analytical model reports 0.0% for every one of them. +What it needs to transfer is the ORDER, not the latency. + +On the one grid this has been measured against, it does not. Kernel dispatch +(below) is part of the reason but not all of it: a reduced-parallelism server +resolves exactly the deployment's kernels and the order still does not hold. +What is left is that fewer GPUs is a different machine, and the levers do not +keep their order across it. Hence the default below is off, and turning it on +wants evidence from the deployment in front of you. + +The catch, and the reason for most of the code below: the probe does not run the +deployment's kernels just because it was handed the deployment's flags. vLLM +resolves an attention and a MoE implementation per engine, and the offline engine +the probe builds resolves them differently from the server EXPLORE benchmarks. +Measured on gpt-oss-120b/MI355X at the SAME TP=8, same model, same flags: the +server settles on ROCM_AITER_UNIFIED_ATTN with a Triton MXFP4 MoE, the probe on +ROCM_AITER_FA with an AITER one. Across that gap the screen is not noisy, it is +confidently wrong: it ranks a stack the deployment never runs. + +So the screen does not assume its regime, it checks it: both sides say in their +own logs which kernels they resolved, the deployment's from a benchmark the +session has already paid for, and nothing is pruned unless they agree. The target +backend is pinned onto the probe to give that check a chance of passing; it is +pinned first, so a variant naming the backend itself still overrides it. + +A matching regime is necessary and not sufficient, which is why this is off by +default. Forcing the probe all the way onto the server's kernels lines one lever +up and lines the next one up by deleting it: with the MoE implementation pinned, +VLLM_ROCM_USE_AITER_MOE=0 becomes a no-op in the probe and reads flat, where on +the server it is not. Kernel dispatch is therefore verified and not forced past +the attention backend, and a screen that cannot reach the deployment's regime +returns the grid untouched rather than a ranking of a different stack. + +The screen only prunes, never promotes, and a variant whose probe fails is kept: +the cost of a wasted benchmark is minutes, the cost of silently discarding the +round's winner is the session. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import subprocess +import tempfile +from pathlib import Path +from typing import Any + +import yaml + +from ._grid_base import GridVariant + +log = logging.getLogger(__name__) + +ENV_ENABLED = "HYPERLOOM_EXPLORE_SCREEN" +ENV_MARGIN_PCT = "HYPERLOOM_EXPLORE_SCREEN_MARGIN_PCT" +ENV_GPUS = "HYPERLOOM_EXPLORE_SCREEN_GPUS" +ENV_MODEL = "HYPERLOOM_EXPLORE_SCREEN_MODEL" +ENV_LAYERS = "HYPERLOOM_EXPLORE_SCREEN_LAYERS" +ENV_BACKEND = "HYPERLOOM_EXPLORE_SCREEN_ATTENTION_BACKEND" +ENV_TIMEOUT = "HYPERLOOM_EXPLORE_SCREEN_TIMEOUT_SEC" +ENV_INFERA_ROOT = "HYPERLOOM_INFERASIM_ROOT" + +BENCH_REL = "infera/projection/core/projection/inference_projection/benchmark_vllm.py" +# A decode step is differenced between a K and a K/2 run, so a short K leaves +# call-to-call jitter in the estimate. Measured on MI355X: K=256 repeats to 18% +# sd on an identical config, K=1024 to 3.5%. +DECODE_STEPS = 1024 +SEEDS = "0,1,2" +# The screen may only act on gaps far wider than its own noise. Measured on eight +# independent replicas, a gap this wide is one every replica reproduces and +# narrower ones are not, which makes this a floor on what may be pruned rather +# than a width at which pruning becomes safe. +DEFAULT_MARGIN_PCT = 10.0 +DEFAULT_TIMEOUT_SEC = 1800 +# The reading the survivors are judged against: the stack as it stands, probed +# the same way on the same device, so device and drift are common to both sides. +BASELINE = "__screen_baseline__" +BACKEND_FLAG = "--attention-backend" +# vLLM names every kernel family it settled on as it builds the engine, whether +# it was told which to use or fell back to one. These are the lines that say +# which stack a run actually measured. +_KERNEL_RES = { + "attention": re.compile(r"Using (\w+) backend|Overriding with (\w+)"), + "moe": re.compile(r"Using '([\w]+)' Mxfp4 MoE backend"), +} + + +def screen_enabled() -> bool: + """Off unless explicitly turned on.""" + return str(os.environ.get(ENV_ENABLED, "0")).strip().lower() in ("1", "true", "yes") + + +def kernels_from_log(text: str) -> dict[str, str]: + """The kernel families a vLLM log says the run actually resolved to. + + This is read from both sides -- the deployment's server log and the probe's + own output -- because it is the only honest way to know they are the same + stack. They are not the same by construction: the offline engine the probe + builds and the server EXPLORE benchmarks select different kernels from the + identical model and flags (measured on gpt-oss-120b/MI355X at TP8: the probe + resolves ROCM_AITER_FA with an AITER MXFP4 MoE, the server + ROCM_AITER_UNIFIED_ATTN with a Triton one). Ranked across that gap the screen + is measuring a stack the deployment never runs. + """ + found = {} + for family, pattern in _KERNEL_RES.items(): + match = pattern.search(text or "") + if match: + found[family] = next(g for g in match.groups() if g) + return found + + +def _target_kernels(session_dir: Path | None) -> dict[str, str]: + """The kernels the DEPLOYMENT runs, read from a benchmark already paid for. + + EXPLORE has booted the stack at full parallelism before it proposes anything, + so the answer is already on disk and costs nothing to look up. + """ + if session_dir is None: + return {} + logs = sorted( + Path(session_dir).rglob("server.log"), key=lambda p: p.stat().st_mtime if p.exists() else 0, reverse=True + ) + for path in logs[:8]: + try: + kernels = kernels_from_log(path.read_text(errors="ignore")) + except OSError: + continue + if kernels: + log.info("explore screen: target kernels %s (from %s)", kernels, path) + return kernels + return {} + + +def _target_backend(bench: dict[str, Any], session_dir: Path | None) -> str | None: + """The attention backend to pin onto the probe, or None to leave it alone.""" + override = os.environ.get(ENV_BACKEND, "").strip() + if override: + return override + + args = str((bench.get("envs") or {}).get("EXTRA_VLLM_ARGS") or "") + tokens = args.split() + for i, tok in enumerate(tokens): + if tok == BACKEND_FLAG and i + 1 < len(tokens): + return tokens[i + 1] + if tok.startswith(BACKEND_FLAG + "="): + return tok.split("=", 1)[1] + + return _target_kernels(session_dir).get("attention") + + +def _probe_command(variant: GridVariant, bench: dict[str, Any], out_path: str, backend: str | None) -> list[str]: + """The ``benchmark_vllm`` invocation that screens one variant.""" + envs = bench.get("envs") or {} + root = os.environ.get(ENV_INFERA_ROOT, "") + model = os.environ.get(ENV_MODEL) or str(bench.get("model") or envs.get("MODEL", "")) + layers = os.environ.get(ENV_LAYERS, "").strip() + + cmd = [ + "python", + str(Path(root) / BENCH_REL), + "--model", + model, + "--tp", + str(_as_int(envs.get("TP"), 1)), + "--benchmark-gpus", + str(_probe_gpus(envs)), + "--batches", + str(_as_int(envs.get("CONC"), 32)), + "--input-len", + str(_as_int(envs.get("ISL"), 1024)), + "--decode-steps", + str(DECODE_STEPS), + "--seeds", + SEEDS, + "--load-format", + "auto", + "--routing-dist", + "none", + # The screen is a ranking probe, not an anchor, and it depends on things + # only the offline entrypoint offers: truncated layers, a fixed decode + # step count and a seed sweep. Anchors take the serving default instead. + "--offline", + "--save", + out_path, + ] + if layers: + cmd += ["--num-hidden-layers", layers] + # The pin goes first so a variant that names the backend itself still wins: + # it holds the rest of the stack at the target's regime, it does not override + # the lever under test. + pin = [BACKEND_FLAG, backend] if backend else [] + server_args = " ".join([*pin, variant.extra_server_args or ""]).strip() + if server_args: + cmd += ["--server-args=" + server_args] + for key, value in (variant.extra_envs or {}).items(): + cmd += ["--env", f"{key}={value}"] + return cmd + + +DEFAULT_PROBE_GPUS = 1 + + +def _probe_gpus(envs: dict[str, Any]) -> int: + """How many GPUs the probe runs on, never more than the target's TP. + + One by default: this is where nearly all the saving is, and the regime guard + above is what makes taking it safe. + """ + target = _as_int(envs.get("TP"), 1) + return min(target, _as_int(os.environ.get(ENV_GPUS), DEFAULT_PROBE_GPUS)) + + +def _as_int(value: Any, default: int) -> int: + try: + return int(str(value).strip()) + except (TypeError, ValueError): + return default + + +def _probe( + variant: GridVariant, bench: dict[str, Any], timeout_sec: int, backend: str | None +) -> tuple[float | None, dict[str, str]]: + """One variant's decode step latency (ms) and the kernels that produced it. + + A reading of None means the probe could not answer, which is always resolved + in the variant's favour by the caller. + """ + with tempfile.TemporaryDirectory(prefix="explore-screen-") as tmp: + out_path = os.path.join(tmp, "probe.json") + cmd = _probe_command(variant, bench, out_path, backend) + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_sec) + except (OSError, subprocess.SubprocessError) as exc: + log.warning("explore screen: probe for %r did not run (%s)", variant.name, exc) + return None, {} + kernels = kernels_from_log((proc.stdout or "") + (proc.stderr or "")) + if proc.returncode != 0 or not os.path.exists(out_path): + # vLLM raises rather than falling back when a pinned backend is not + # valid for the probe's shape, so this is also how a screen that + # could not hold the target's regime fails. + log.warning( + "explore screen: probe for %r failed rc=%s: %s", + variant.name, + proc.returncode, + (proc.stderr or "")[-400:], + ) + return None, kernels + try: + with open(out_path) as fh: + sweep = json.load(fh)["sweep"] + return float(sweep[0]["decode_ms"]), kernels + except (OSError, ValueError, KeyError, IndexError) as exc: + log.warning("explore screen: probe for %r produced no reading (%s)", variant.name, exc) + return None, kernels + + +def screen_variants( + variants: list[GridVariant], + config_path: Path, + *, + session_dir: Path | None = None, + margin_pct: float | None = None, +) -> tuple[list[GridVariant], list[dict[str, str]]]: + """Return the variants still worth benchmarking, plus a record of what was cut. + + A variant is cut only when the screen puts it more than ``margin_pct`` behind + the screened baseline -- a gap far wider than the screen's own noise. The + screen is not asked to pick a winner: the differences EXPLORE keeps on are a + few percent, which is inside its error. It is asked to recognise the variants + that are decisively worse. + + Nothing is cut unless the probe ran in the target's regime, and anything the + probe cannot read is kept, so a broken or mismatched screen degrades to + today's behaviour rather than to a silently smaller grid. + """ + if not screen_enabled() or len(variants) < 3: + return list(variants), [] + if not os.environ.get(ENV_INFERA_ROOT): + log.warning("explore screen: %s is not set; skipping the screen", ENV_INFERA_ROOT) + return list(variants), [] + + try: + with open(config_path) as fh: + bench = (yaml.safe_load(fh) or {}).get("benchmark") or {} + except (OSError, yaml.YAMLError) as exc: + log.warning("explore screen: could not read %s (%s); skipping", config_path, exc) + return list(variants), [] + + target = _target_kernels(session_dir) + if not target: + log.warning( + "explore screen: no benchmark in this session says which kernels " + "the deployment runs, so a probe cannot be checked against it; " + "skipping the screen" + ) + return list(variants), [] + + timeout_sec = _as_int(os.environ.get(ENV_TIMEOUT), DEFAULT_TIMEOUT_SEC) + backend = _target_backend(bench, session_dir) + baseline, probed = _probe(GridVariant(name=BASELINE), bench, timeout_sec, backend) + if baseline is None: + log.warning("explore screen: baseline probe failed; benchmarking the full grid") + return list(variants), [] + + mismatch = {k: (v, probed.get(k)) for k, v in target.items() if probed.get(k) != v} + if mismatch: + log.warning( + "explore screen: the probe is not running the deployment's kernels " + "(%s), so its ordering is about a different stack; skipping the screen", + ", ".join(f"{k}: target {t}, probe {p}" for k, (t, p) in mismatch.items()), + ) + return list(variants), [] + + margin = margin_pct if margin_pct is not None else _margin_pct() + cut_at = baseline * (1.0 + margin / 100.0) + + survivors, dropped = [], [] + for variant in variants: + reading, _ = _probe(variant, bench, timeout_sec, backend) + if reading is not None and reading > cut_at: + dropped.append( + { + "name": variant.name, + "reason": "screen_decisively_slower", + "detail": f"probe decode {reading:.3f} ms vs baseline " + f"{baseline:.3f} ms (+{(reading / baseline - 1) * 100:.0f}%)", + } + ) + else: + survivors.append(variant) + + log.info( + "explore screen: %d/%d variants forwarded to benchmark; cut %s", + len(survivors), + len(variants), + ", ".join(d["name"] for d in dropped) or "nothing", + ) + return survivors, dropped + + +def _margin_pct() -> float: + try: + margin = float(os.environ.get(ENV_MARGIN_PCT, DEFAULT_MARGIN_PCT)) + except (TypeError, ValueError): + return DEFAULT_MARGIN_PCT + return margin if margin > 0 else DEFAULT_MARGIN_PCT diff --git a/src/hyperloom/orchestrator/actions/executors/_grid_runner.py b/src/hyperloom/orchestrator/actions/executors/_grid_runner.py index 395716f2b9..19b491f6bc 100644 --- a/src/hyperloom/orchestrator/actions/executors/_grid_runner.py +++ b/src/hyperloom/orchestrator/actions/executors/_grid_runner.py @@ -658,11 +658,24 @@ def _read_pid_gpu_mask(pid: int) -> tuple[list[int], bool] | None: def _kill_stale_servers() -> None: - """Deep-clean any lingering inference server processes + shared memory.""" - from ._multi_node_env import is_multi_node + """Deep-clean any lingering inference server processes + shared memory. + + Reaps vLLM::Worker / EngineCore children that escape Magpie's pgrp-leader + cleanup. Called before every Magpie invocation; uses a /proc scan (not + pgrep) to avoid clashing with test subprocess mocks. No-op in multi-node + mode (servers live in RayJob pods) and against an external server (the + engine behind it is not ours to kill -- reaping it would restart the very + server the client-only run is measuring). + + Note: + Side-effecting and best-effort: it sends signals to matching processes + and unlinks stale shared-memory segments, swallowing errors. Returns + nothing. + """ + from ._multi_node_env import is_multi_node, uses_external_server from ...bus.gpu_pool import _visible_device_mask - if is_multi_node(): + if is_multi_node() or uses_external_server(): return import signal diff --git a/src/hyperloom/orchestrator/actions/executors/_multi_node_env.py b/src/hyperloom/orchestrator/actions/executors/_multi_node_env.py index 7fdfb8c008..d5b0136888 100644 --- a/src/hyperloom/orchestrator/actions/executors/_multi_node_env.py +++ b/src/hyperloom/orchestrator/actions/executors/_multi_node_env.py @@ -1,7 +1,20 @@ # SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. # SPDX-License-Identifier: MIT -"""Helper that bridges the multi-node CLI state into Magpie subprocesses.""" +"""Helper that bridges the multi-node CLI state into Magpie subprocesses. + +Lives in the executors package (the dependency edge stays one-way: executors +import this; ``multi_node/`` knows nothing about them). + +Reads ``$INFERENCE_OPTIMIZER_NODES`` + ``$MULTI_NODE_STATE_FILE``. Single node +(< 2): returns ``{}``. Multi-node (>= 2) with a ``service_url``: returns +``MAGPIE_RUN_PHASE=client`` + ``BENCHMARK_BASE_URL=`` so Magpie +skips its own server launch and points ``benchmark_serving`` at the head pod. +``$HYPERLOOM_MN_EXT_SERVICE_URL`` selects that same client-only path at any node +count (:func:`uses_external_server`). +:func:`export_ray_address_to_os` also copies ``ray_address`` into +``RAY_ADDRESS`` for kernel-agent ``ray.init``. +""" from __future__ import annotations @@ -52,6 +65,21 @@ def is_multi_node() -> bool: return env_n >= 2 +def uses_external_server() -> bool: + """True when benchmarks target a server Hyperloom does not manage. + + ``HYPERLOOM_MN_EXT_SERVICE_URL`` hands over an OpenAI-compatible endpoint -- + a platform cluster, or an Infera router fronting the vLLM/SGLang engine. + Independent of node count: Magpie's client-only path is env-driven, so a + single-node run can target an externally hosted engine. Callers must then + neither launch, reuse, nor reap a local server. + + Returns: + True when an external endpoint was handed over, else False. + """ + return bool(external_service_url()) + + def resolve_kb_topology() -> dict[str, Any]: """Resolve the node/GPU and PD-disaggregation topology for the KB hardware suffix.""" state = _read_state() @@ -238,10 +266,23 @@ def _remote_client_env(service_url: str) -> dict[str, str]: def magpie_remote_env() -> dict[str, str]: - """Return env vars to inject into a Magpie ``benchmark`` subprocess.""" - # External mode: point benchmarks at the env-provided endpoint when multi-node. + """Return env vars to inject into a Magpie ``benchmark`` subprocess. + + An external endpoint (:func:`uses_external_server`) wins at any node count: + ``MAGPIE_RUN_PHASE=client`` + ``BENCHMARK_BASE_URL`` so Magpie skips its + local server launch and points ``benchmark_serving`` at that endpoint, plus + ``MAGPIE_EVAL_PYTHON`` (see :func:`_remote_client_env`). Otherwise + multi-node resolves the same env from the state file's ``service_url``, and + single-node returns ``{}`` (Magpie's ``--run-mode local`` untouched). + Multi-node without a state file: ``{}`` + WARN (the local-launch failure + surfaces clearly). + + Returns: + Env vars to inject into the Magpie subprocess, or ``{}`` for the + single-node path or when no service URL is available. + """ ext = external_service_url() - if ext and is_multi_node(): + if ext: return _remote_client_env(ext) if not is_multi_node(): return {} @@ -306,4 +347,5 @@ def log_mn_banner( "ray_gcs_address_from_state", "rayjob_id_from_state", "resolve_kb_topology", + "uses_external_server", ] diff --git a/src/hyperloom/orchestrator/actions/executors/_server_lifecycle.py b/src/hyperloom/orchestrator/actions/executors/_server_lifecycle.py index 9c0474d3f4..84e7fe9987 100644 --- a/src/hyperloom/orchestrator/actions/executors/_server_lifecycle.py +++ b/src/hyperloom/orchestrator/actions/executors/_server_lifecycle.py @@ -126,11 +126,14 @@ def resolve_lifecycle_params(materialized_config_path: Path) -> dict[str, Any]: info["reason"] = "scriptable framework (server-less; no server_lifecycle)" return info - from ._multi_node_env import is_multi_node + from ._multi_node_env import is_multi_node, uses_external_server if is_multi_node(): info["reason"] = "multi-node (server_lifecycle is local-only)" return info + if uses_external_server(): + info["reason"] = "external server (client-only; nothing local to boot or reuse)" + return info script_name = Path(str(bench.get("benchmark_script") or "")).name if script_name not in MAGPIE_BUILTIN_SCRIPTS: diff --git a/src/hyperloom/orchestrator/actions/executors/benchmark_backend.py b/src/hyperloom/orchestrator/actions/executors/benchmark_backend.py index ced02ee33b..9534828316 100644 --- a/src/hyperloom/orchestrator/actions/executors/benchmark_backend.py +++ b/src/hyperloom/orchestrator/actions/executors/benchmark_backend.py @@ -1,7 +1,25 @@ # SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. # SPDX-License-Identifier: MIT -"""Benchmark backend seam.""" +"""Benchmark backend seam. + +Central place that builds the benchmark subprocess command line so the +optimizer can run against different benchmark engines without every executor +knowing which engine is active. The default backend is Magpie, whose command is +``python -m Magpie -v benchmark --benchmark-config CFG --output-dir OUT +--run-mode local``. + +The bypass backend implements the same contract (same input YAML, same +workspace/report artifacts) and is selected via +HYPERLOOM_BENCHMARK_BACKEND=bypass without touching the executors. + +The inferasim backend implements the same contract but produces the report from +Infera's ``inferasim`` serving projection (analytical / anchor-calibrated, no +GPU) instead of a real server + client. It is selected via +HYPERLOOM_BENCHMARK_BACKEND=inferasim and lets an entire optimization session +run without a GPU, reserving real GPU time for the final validation. See +:mod:`inferasim_runner`. +""" from __future__ import annotations @@ -12,7 +30,7 @@ # Backend selection env var. BENCHMARK_BACKEND_ENV = "HYPERLOOM_BENCHMARK_BACKEND" DEFAULT_BENCHMARK_BACKEND = "magpie" -KNOWN_BENCHMARK_BACKENDS = frozenset({"magpie", "bypass"}) +KNOWN_BENCHMARK_BACKENDS = frozenset({"magpie", "bypass", "inferasim"}) class BenchmarkBackend(Protocol): @@ -93,12 +111,18 @@ def lifecycle_eligibility(self, bench: dict) -> dict | None: except (TypeError, ValueError): port = 8888 verdict = {"eligible": False, "framework": framework, "port": port, "reason": ""} - # The reuse protocol boots a local server and re-attaches a client round to it; that only holds single-node. - from ._multi_node_env import is_multi_node + # The reuse protocol boots a local server and re-attaches a client + # round to it; that only holds single-node. Mirror the Magpie path's + # multi-node gate (see _server_lifecycle.resolve_lifecycle_params), + # which our non-None verdict would otherwise short-circuit past. + from ._multi_node_env import is_multi_node, uses_external_server if is_multi_node(): verdict["reason"] = "multi-node (server_lifecycle is local-only)" return verdict + if uses_external_server(): + verdict["reason"] = "external server (client-only; nothing local to boot or reuse)" + return verdict if framework not in self._LIFECYCLE_FRAMEWORKS: verdict["reason"] = f"framework {framework!r} is not a serving framework" return verdict @@ -131,6 +155,68 @@ def build_command( ] +class InferasimBackend: + """InferaSim backend: projects serving metrics instead of running a server. + + Accepts the same CLI flags as Magpie/bypass and writes the same + workspace/report contract, but every measurement comes from Infera's + analytical (optionally anchor-calibrated) serving projection, so no server + is booted and no GPU is used. See :mod:`inferasim_runner`. + """ + + name = "inferasim" + + def resolve_interpreter(self) -> str: + """Return the interpreter used to run the InferaSim projection. + + Prefers ``HYPERLOOM_INFERASIM_PYTHON`` (an interpreter that can import + ``infera``), then the current interpreter, then a PATH ``python3``. + InferaSim is analytical and never needs Magpie's canonical venv. + """ + import shutil + import sys + + explicit = (os.environ.get("HYPERLOOM_INFERASIM_PYTHON") or "").strip() + if explicit: + return explicit + return sys.executable or shutil.which("python3") or "python3" + + def lifecycle_eligibility(self, bench: dict) -> dict | None: + """A projection has no persistent server, so lifecycle reuse is off. + + Returning an ineligible verdict routes run_grid through single-shot + ``phase=all`` calls (the projection is cheap and stateless), mirroring + the Magpie non-lifecycle path. + """ + return { + "eligible": False, + "framework": str(bench.get("framework") or "").lower(), + "port": 0, + "reason": "inferasim projection has no server to reuse", + } + + def build_command( + self, + *, + python_exe: str, + config_path: Path, + output_dir: Path, + ) -> list[str]: + """Return the InferaSim runner argv mirroring Magpie's flags.""" + return [ + python_exe, + "-m", + "hyperloom.orchestrator.actions.executors.inferasim_runner", + "benchmark", + "--benchmark-config", + str(config_path), + "--output-dir", + str(output_dir), + "--run-mode", + "local", + ] + + def resolve_backend_name() -> str: """Resolve the active backend name from the environment.""" raw = (os.environ.get(BENCHMARK_BACKEND_ENV) or "").strip().lower() @@ -140,10 +226,20 @@ def resolve_backend_name() -> str: def resolve_backend() -> BenchmarkBackend: - """Resolve the active benchmark backend instance.""" + """Resolve the active benchmark backend instance. + + ``bypass`` selects the Hyperloom runner; ``inferasim`` selects the analytical + projection runner; ``magpie`` (the default) and any unknown value fall back + to Magpie so a typo cannot silently disable benchmarking. + + Returns: + The selected BenchmarkBackend implementation. + """ name = resolve_backend_name() if name == "bypass": return BypassBackend() + if name == "inferasim": + return InferasimBackend() return MagpieBackend() diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py index 573832f3cc..12d60d419f 100644 --- a/src/hyperloom/orchestrator/actions/executors/explore.py +++ b/src/hyperloom/orchestrator/actions/executors/explore.py @@ -57,6 +57,7 @@ from . import _framework_switch_manifest as _switch_manifest from ._canonical_fingerprint import workload_signature from ._proposal_identity import effective_fingerprint, normalize_proposal +from ._explore_screen import screen_variants from ._grid_runner import ( DEFAULT_KEEP_THRESHOLD_PCT, _MN_BACKENDS_PRIORITY, @@ -94,9 +95,105 @@ log = logging.getLogger(__name__) +# Run-to-run spread on throughput for the same configuration on the same server. +# Whether a repeat could flip a verdict depends on this and not on the threshold: +# a variant measured at 0.05% gain against a 1% threshold is only one noise +# envelope from crossing it, and a repeat of that is a genuine second opinion. A +# variant at -0.8% is nearly four envelopes away and a repeat only re-derives the +# same answer. The audited session's warmup and decision rounds of one +# configuration landed 0.06% apart; 0.5% is the conservative envelope around it. +REPEAT_NOISE_PCT = 0.5 + +# How many noise envelopes a prior result must sit clear of the KEEP threshold +# before a repeat is treated as settled rather than as a second opinion. +SETTLED_MARGIN = 2.0 + + +# The warmup round exists to leave the server hot so the decision round is +# measured warm, and to run the accuracy gate. Its throughput number is read for +# success/failure and then discarded. It nevertheless runs a full-length +# benchmark: ``_workload_envs`` sizes ``NUM_PROMPTS`` at ``CONC`` times a factor +# of 10/5/3/2 depending on sequence length, so the round Arbor throws away is +# five to ten waves of the concurrency. +# +# One wave is enough to warm. It fills every slot and decodes a full OSL, and +# measurements taken immediately after a four-token warmup show no ordered +# difference from ones taken after a full benchmark (see Infera's +# ``warmup_cost.py``). The accuracy gate is a separate workload and does not +# read ``NUM_PROMPTS``, so it is unaffected. +# +# Set ``INFERENCE_OPTIMIZER_EXPLORE_SHORT_WARMUP=0`` to restore the full-length +# warmup round. +WARMUP_WAVES = 1 + + +def _short_warmup_enabled() -> bool: + raw = os.environ.get("INFERENCE_OPTIMIZER_EXPLORE_SHORT_WARMUP") + return (raw if raw is not None else "1").strip().lower() not in {"0", "false", "no", "off"} + + +def _warmup_num_prompts(config_path: Any) -> int | None: + """How many prompts a warmup round needs: one wave of the concurrency. + + Returns ``None`` when the concurrency cannot be read, which leaves the + warmup round at whatever length ``_workload_envs`` would have chosen. That + is the safe direction: the failure mode is the warmup we already run. + """ + try: + with Path(config_path).open(encoding="utf-8") as fp: + cfg = yaml.safe_load(fp) or {} + envs = (cfg.get("benchmark") or {}).get("envs") or {} + conc = int(envs.get("CONC", 0) or 0) + except Exception: # noqa: BLE001 — best-effort; fall back to the long warmup + return None + if conc <= 0: + return None + return max(conc * WARMUP_WAVES, conc) + + _now_iso = functools.partial(now_iso, "auto") +def _settled_against_same_stack( + prior: dict[str, Any], + ws_sig: str, + base_tput: float, + keep_threshold_pct: float, +) -> bool: + """Has this exact variant already been benchmarked against this exact stack? + + Re-running it then costs a server boot and a full benchmark to re-derive a + number the ledger already holds. But skipping is only safe while nothing the + variant composes with has moved: a flag that loses alone can win once a patch + it interacts with has landed, and dropping it for that reason would be the + search quietly narrowing itself, which is the one outcome worth more than the + GPU time. + + So this asks for an exact match on three things. The workload signature, so a + different shape is always re-measured. The baseline the prior run was scored + against -- the running baseline only advances when a KEEP changes the stack, + so an unchanged number means an unchanged stack. And a conclusive prior + outcome sitting at least :data:`SETTLED_MARGIN` noise envelopes clear of the + KEEP threshold, because a result that landed near it may have been decided by + noise rather than by the variant. + + Every other case returns False and the variant is benchmarked again, + including a baseline that merely drifted on re-measurement. The failure mode + is a repeated run, never a missed optimization. + """ + if str(prior.get("workload_signature") or "") != ws_sig: + return False + if str(prior.get("status") or "") != "succeeded": + return False + prior_base = prior.get("base_tput") + gain = prior.get("gain_pct") + if prior_base is None or gain is None or base_tput <= 0: + return False + if float(prior_base) != float(base_tput): + return False + return abs(float(gain) - keep_threshold_pct) > REPEAT_NOISE_PCT * SETTLED_MARGIN + + def _initial_explore_search_state() -> dict[str, Any]: """Empty :attr:`SharedState.explore_search` ledger.""" return { @@ -782,6 +879,21 @@ async def _run_explore(self, ctx) -> dict[str, Any]: } ) continue + prior = (tested_dict or {}).get(fp) + if isinstance(prior, dict) and _settled_against_same_stack(prior, ws_sig, base_tput, keep_threshold_pct): + skipped_dup.append( + { + "name": gv.name, + "fingerprint": fp, + "reason": "settled_prior_round", + "detail": ( + f"tested in {prior.get('round_id') or 'an earlier round'} " + f"against the same stack: {float(prior.get('gain_pct')):.2f}% " + f"gain, {prior.get('outcome') or 'no outcome'}" + ), + } + ) + continue unique_in_round[fp] = gv runnable: list[GridVariant] = list(unique_in_round.values()) @@ -789,13 +901,15 @@ async def _run_explore(self, ctx) -> dict[str, Any]: # Re-proposals are still benchmarked; the tested ledger already carries # each prior outcome and is rendered in full, so this only counts them. re_proposed = sum(1 for fp in unique_in_round if isinstance(tested_dict.get(fp), dict)) + _settled = sum(1 for d in skipped_dup if d.get("reason") == "settled_prior_round") log.info( - "explore dedup: payload=%d → runnable=%d (round_dup=%d re_proposed=%d)", + "explore dedup: payload=%d → runnable=%d (round_dup=%d re_proposed=%d settled=%d)", len(grid), len(runnable), - len(skipped_dup), + len(skipped_dup) - _settled, re_proposed, + _settled, ) # Multi-node grid shaping. @@ -828,6 +942,12 @@ async def _run_explore(self, ctx) -> dict[str, Any]: priority_tags=_MN_PARAMS_PRIORITY + _MN_BACKENDS_PRIORITY, ) + # Drop the variants a cheap reduced-scale probe puts decisively behind + # the stack. Off by default; a no-op when the probe cannot run, or when + # it cannot be held in the deployment's kernel regime. + runnable, screened_out = screen_variants(runnable, config_path, session_dir=_resolve_session_dir()) + skipped_dup.extend(screened_out) + round_id_seed = int(search.get("cursor") or 0) + 1 round_id = f"explore-{round_id_seed:03d}" @@ -987,6 +1107,26 @@ def _stopped_by_the_run(result: Any, *, variant: GridVariant, idx: int, round_la if use_warm_decision: warmup_slot = slot / "warmup_round" warmup_slot.mkdir(parents=True, exist_ok=True) + # The warmup's throughput is discarded below, so it only + # has to reach steady state and run the accuracy gate. + # Shorten it to one wave of the concurrency instead of + # the five to ten a measured round would use. + warmup_prompts = _warmup_num_prompts(config_path) if _short_warmup_enabled() else None + if warmup_prompts is not None: + warmup_envs = dict(getattr(warmup_gv, "extra_envs", {}) or {}) + warmup_envs["NUM_PROMPTS"] = str(warmup_prompts) + warmup_gv = _carry_variant_metadata( + warmup_gv, + GridVariant( + name=warmup_gv.name, + extra_server_args=warmup_gv.extra_server_args, + extra_envs=warmup_envs, + note=warmup_gv.note, + remove_args=list(getattr(warmup_gv, "remove_args", []) or []), + unset_envs=list(getattr(warmup_gv, "unset_envs", []) or []), + args_mode=str(getattr(warmup_gv, "args_mode", "append") or "append"), + ), + ) warmup_results = await run_grid( base_yaml_path=config_path, base_extra_args=stack_extra_args, diff --git a/src/hyperloom/orchestrator/actions/executors/inferasim_bridge.py b/src/hyperloom/orchestrator/actions/executors/inferasim_bridge.py new file mode 100644 index 0000000000..4904d54490 --- /dev/null +++ b/src/hyperloom/orchestrator/actions/executors/inferasim_bridge.py @@ -0,0 +1,867 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""InferaSim projection bridge. + +Maps a Hyperloom benchmark spec (the ``benchmark`` block of a materialized +Magpie YAML: framework/model/precision + TP/CONC/ISL/OSL envs) onto Infera's +``inferasim`` serving projection and returns the same throughput/latency +measurements a real serving benchmark would produce -- without booting a +server or touching a GPU. + +This is the analytical inner-loop that lets the optimizer simulate candidate +serving configs and spend real GPU time only on the final validation, which is +the GPU-time reduction projected in the deck. + +Design notes +------------ +* Infera is an *optional* dependency. Everything here imports it lazily so the + Hyperloom base install is unaffected; a missing/broken Infera surfaces as a + structured error the runner turns into a failed report (never a crash). +* The projection is driven through Infera's own CLI plumbing + (``build_parser().parse_known_args`` -> ``launch_projection_from_cli``) so we + inherit its argument defaults and stay forward-compatible with new flags + instead of hand-constructing its config dataclasses. +* Model selection is deliberately explicit: the operator points us at an + InferaSim model preset (``HYPERLOOM_INFERASIM_MODEL``) or a full workload YAML + (``HYPERLOOM_INFERASIM_WORKLOAD``); a best-effort heuristic maps common HF + model paths to presets so the common cases work with zero extra config. +""" + +from __future__ import annotations + +import json +import os +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +# Env knobs (all optional unless noted). Documented in the module docstring and +# the runner --help. +ENV_ROOT = "HYPERLOOM_INFERASIM_ROOT" # path to the Infera checkout (added to sys.path) +ENV_WORKLOAD = "HYPERLOOM_INFERASIM_WORKLOAD" # explicit InferaSim workload YAML +ENV_MODEL = "HYPERLOOM_INFERASIM_MODEL" # InferaSim model preset name (e.g. gpt_oss_120B) +ENV_GPU_ARCH = "HYPERLOOM_INFERASIM_GPU_ARCH" # e.g. mi355x (default) +ENV_HBM_GB = "HYPERLOOM_INFERASIM_HBM_GB" # per-GPU HBM capacity, GB +ENV_EP = "HYPERLOOM_INFERASIM_EP" # expert parallelism override +ENV_PP = "HYPERLOOM_INFERASIM_PP" # pipeline parallelism override +ENV_KV_DTYPE = "HYPERLOOM_INFERASIM_KV_DTYPE" # kv-cache dtype override +ENV_ANCHOR = "HYPERLOOM_INFERASIM_ANCHOR" # single GPU anchor JSON (calibration) +ENV_ANCHOR_SCALING = "HYPERLOOM_INFERASIM_ANCHOR_SCALING" # comma-sep TP-scaling anchors +ENV_ANCHOR_STORE = "HYPERLOOM_INFERASIM_ANCHOR_STORE" # dir of warmup anchors (auto-select) +ENV_SERVING_MODEL = "HYPERLOOM_INFERASIM_SERVING_MODEL" # continuous (default) | static + +_DEFAULT_GPU_ARCH = "mi355x" +# Per-GPU HBM by arch (GB); only used when HBM is not supplied explicitly. +_ARCH_HBM_GB = {"mi300x": 192.0, "mi325x": 256.0, "mi355x": 288.0} + +# Best-effort HF-path/name substring -> InferaSim megatron preset. First match +# wins; extend freely. Override any time with HYPERLOOM_INFERASIM_MODEL. +_MODEL_HEURISTICS: tuple[tuple[str, str], ...] = ( + ("gpt-oss-120b", "gpt_oss_120B"), + ("gpt-oss-20b", "gpt_oss_20B"), + ("gpt_oss_120b", "gpt_oss_120B"), + ("gpt_oss_20b", "gpt_oss_20B"), + ("minimax-m2.5", "minimax_m2.5"), + ("minimax_m2.5", "minimax_m2.5"), + ("qwen3-235b", "qwen3_235B_A22B"), + ("qwen3-32b", "qwen3_32B"), + ("qwen3-30b", "qwen3_30B_A3B"), + ("qwen3-14b", "qwen3_14B"), + ("qwen3-4b", "qwen3_4B"), + ("qwen2.5-72b", "qwen2.5_72B"), + ("qwen2.5-32b", "qwen2.5_32B"), + ("qwen2.5-14b", "qwen2.5_14B"), + ("qwen2.5-7b", "qwen2.5_7B"), + ("llama3.1-70b", "llama3.1_70B"), + ("llama3.1-8b", "llama3.1_8B"), + ("llama3.3-70b", "llama3.3_70B"), + # Longest/most specific needles first: "deepseek-v2-lite" must not be + # swallowed by "deepseek-v2", and R1 is the V3 architecture at 671B so it + # resolves to the V3 config rather than a config of its own. + ("deepseek-v2-lite", "deepseek_v2_lite"), + ("deepseek-r1", "deepseek_v3_671b-fp8"), + ("deepseek-v3", "deepseek_v3"), + ("deepseek-v2", "deepseek_v2"), + ("minimax-m2", "minimax_m2.5"), + ("mixtral-8x22b", "mixtral_8x22B_v0.1"), + ("mixtral-8x7b", "mixtral_8x7B_v0.1"), +) + +# Bundled env-driven workload template used when only a preset name is known. +_TEMPLATE_WORKLOAD = ( + Path(__file__).resolve().parents[3] / "inference_optimizer" / "assets" / "inferasim" / "inferasim_workload.yaml" +) + + +class InferasimBridgeError(RuntimeError): + """Raised for any recoverable bridge failure (bad config, import, etc.).""" + + +@dataclass +class ServingSpec: + """Normalized serving request extracted from a Hyperloom benchmark block.""" + + framework: str + model_path: str + tp: int = 1 + ep: int = 1 + pp: int = 1 + conc: int = 64 + isl: int = 1024 + osl: int = 1024 + weight_dtype: str = "bf16" + kv_cache_dtype: str = "bf16" + extra_server_args: str = "" + + +# Context, in tokens, out to which the projection has actually been checked +# against real vLLM measurements. A controlled sweep -- prefix caching off, three +# seeds, prompt length from 1k to 65,536 -- scores 6.0% across that range, which +# covers everything Hyperloom searches. Beyond it the decode step has never been +# compared to hardware, and the last extrapolation past a measured range hid a +# 46.3% error until someone ran the sweep. +VALIDATED_CONTEXT_TOKENS = 65536 + +# Above this the deployment spans nodes, and no inference measurement behind the +# projection ever crossed a node boundary. +SINGLE_NODE_GPUS = 8 + + +def extrapolation_notes(spec: "ServingSpec", replica_gpus: int, calibrated: bool) -> list[str]: + """Where this projection is being asked to work outside its evidence. + + Returned on every projection so a search cannot quietly trust a number that + was produced by extrapolating along an axis nothing validated. These are not + error bars -- the size of the error is not known, which is the point. + """ + notes: list[str] = [] + context = int(spec.isl or 0) + int(spec.osl or 0) + if context > VALIDATED_CONTEXT_TOKENS: + notes.append( + f"context {context} tokens exceeds the {VALIDATED_CONTEXT_TOKENS} " + "this model has been checked against, and the decode step is known " + "to be too flat in context; use for ranking, and anchor near the " + "target context before believing the absolute latency" + ) + if replica_gpus > SINGLE_NODE_GPUS: + notes.append( + f"{replica_gpus} GPUs spans nodes; the inter-node collective terms " + "are derived, not measured, and nothing has checked them" + ) + if not calibrated: + notes.append( + "no warmup anchor matched this model, so this is pure simulation with no measurement pinning its scale" + ) + return notes + + +@dataclass +class ProjMetrics: + """Projected serving metrics, mapped onto benchmark measurement fields.""" + + output_throughput: float # aggregate output tok/s (Magpie headline) + request_throughput: float + total_token_throughput: float + ttft_ms: float + tpot_ms: float + itl_ms: float + e2el_ms: float + decode_tps_per_gpu: float + memory_per_gpu_gb: float + max_concurrency: int + calibrated: bool = False + replica_gpus: int = 0 + extras: dict[str, Any] = field(default_factory=dict) + + +def _as_int(value: Any, default: int) -> int: + try: + return int(str(value).strip()) + except (TypeError, ValueError): + return default + + +def _first_env_or(bench_envs: dict, key: str, default: Any) -> Any: + """Ambient env wins over YAML envs (Magpie/bypass convention), then default.""" + val = os.environ.get(key) + if val is None or str(val).strip() == "": + val = bench_envs.get(key) + if val is None or str(val).strip() == "": + return default + return val + + +def _precision_to_weight_dtype(precision: str) -> str: + p = (precision or "").strip().lower() + if p in ("fp8", "e4m3", "e5m2", "fp8_hybrid", "hybrid"): + return "fp8" + if p in ("mxfp4", "fp4"): + return "mxfp4" + return "bf16" + + +def _parse_server_arg_int(server_args: str, *flags: str) -> int | None: + """Pull an int value for any of ``flags`` out of a server-arg string.""" + if not server_args: + return None + toks = server_args.split() + for i, tok in enumerate(toks): + for flag in flags: + if tok == flag and i + 1 < len(toks): + return _as_int(toks[i + 1], 0) or None + if tok.startswith(flag + "="): + return _as_int(tok.split("=", 1)[1], 0) or None + return None + + +def _parse_server_arg_str(server_args: str, *flags: str) -> str | None: + """Pull a string value for any of ``flags`` out of a server-arg string.""" + if not server_args: + return None + toks = server_args.split() + for i, tok in enumerate(toks): + for flag in flags: + if tok == flag and i + 1 < len(toks): + return toks[i + 1] + if tok.startswith(flag + "="): + return tok.split("=", 1)[1] + return None + + +def _parse_kv_cache_dtype(server_args: str) -> str | None: + """KV-cache dtype from a framework's server args, normalised for InferaSim. + + vLLM spells it ``--kv-cache-dtype fp8_e4m3`` (or ``fp8``, ``fp8_e5m2``) and + sglang ``--kv-cache-dtype fp8_e4m3``; both mean the cache is stored in a + single byte per element, which is what the projection needs to know. ``auto`` + means "follow the weights", which the caller's default already does. + """ + raw = _parse_server_arg_str(server_args, "--kv-cache-dtype", "--kv_cache_dtype") + if not raw: + return None + v = raw.strip().strip("'\"").lower() + if v in ("auto", ""): + return None + if v.startswith("fp8"): + return "fp8" + if v in ("bf16", "bfloat16"): + return "bf16" + if v in ("fp16", "float16", "half"): + return "fp16" + return v + + +def parse_speculative(server_args: str) -> tuple[str | None, int]: + """Speculative-decoding ``(method, k)`` from a framework's server args. + + Each serving framework spells this differently, and Hyperloom's EXPLORE grid + emits all three, so all three are recognised: + + * vLLM ``--speculative-config '{"method": "deepseek_mtp", + "num_speculative_tokens": 3}'`` + * SGLang ``--speculative-algorithm NEXTN --speculative-num-steps 3`` + * atom ``--method mtp --num-speculative-tokens 3`` + + Returns ``(None, 0)`` when the args request no speculation. ``k`` defaults to + 1 when a method is named without a token count, matching every framework's + own default. + """ + if not server_args: + return None, 0 + + method = _parse_server_arg_str(server_args, "--speculative-algorithm") + k = _parse_server_arg_int( + server_args, + "--speculative-num-steps", + "--num-speculative-tokens", + "--speculative-num-draft-tokens", + "--speculative-tokens", + ) + + raw = _parse_server_arg_str(server_args, "--speculative-config") + if raw: + # The JSON is usually quoted as a single shell token, but a bare + # unquoted blob would have been split on spaces by the tokenizer; fall + # back to a regex over the whole string in that case. + try: + cfg = json.loads(raw.strip("'\"")) + except (ValueError, TypeError): + cfg = {} + m = re.search(r'"method"\s*:\s*"([^"]+)"', server_args) + if m: + cfg["method"] = m.group(1) + m = re.search(r'"num_speculative_tokens"\s*:\s*(\d+)', server_args) + if m: + cfg["num_speculative_tokens"] = int(m.group(1)) + if isinstance(cfg, dict): + method = cfg.get("method") or method + k = int(cfg.get("num_speculative_tokens") or 0) or k + + # atom spells the method as ``--method mtp``; only honour that spelling when + # it names a speculative method, since ``--method`` is a generic flag name. + if not method: + generic = _parse_server_arg_str(server_args, "--method") + if generic and generic.strip().lower() in ("mtp", "eagle", "eagle3", "nextn"): + method = generic + + if not method: + return None, 0 + return str(method), max(1, int(k or 0)) + + +def _parse_server_arg_int(server_args: str, *flags: str) -> int: + """Integer value for any of ``flags``, or 0 when absent/non-numeric.""" + raw = _parse_server_arg_str(server_args, *flags) + try: + return int(str(raw)) + except (TypeError, ValueError): + return 0 + + +def spec_from_benchmark(bench: dict) -> ServingSpec: + """Extract a :class:`ServingSpec` from a Magpie ``benchmark`` block.""" + bench = bench or {} + envs = dict(bench.get("envs") or {}) + framework = str(bench.get("framework") or "sglang").lower() + model_path = str(bench.get("model") or os.environ.get("MODEL", "")) + extra_key = { + "sglang": "EXTRA_SGLANG_ARGS", + "vllm": "EXTRA_VLLM_ARGS", + "atom": "EXTRA_ATOM_ARGS", + }.get(framework, "") + extra_args = str(_first_env_or(envs, extra_key, "")) if extra_key else "" + + tp = _as_int(_first_env_or(envs, "TP", 1), 1) + # EP/PP: explicit bridge env, else parse from server args, else 1. + ep = ( + _as_int(os.environ.get(ENV_EP) or "", 0) + or _parse_server_arg_int(extra_args, "--ep-size", "--expert-parallel-size", "--moe-ep-size") + or 1 + ) + pp = ( + _as_int(os.environ.get(ENV_PP) or "", 0) + or _parse_server_arg_int(extra_args, "--pp-size", "--pipeline-parallel-size") + or 1 + ) + + weight_dtype = _precision_to_weight_dtype(str(bench.get("precision") or "bf16")) + # KV dtype: explicit bridge env wins, else the server arg the variant sets, + # else bf16. Reading the arg matters because an explore grid changes the KV + # dtype by passing this flag and nothing else -- and the projection prices KV + # dtype perfectly well, so ignoring the flag made a lever the model *can* + # see look like one it cannot, and projected an fp8 candidate as bf16. + kv_dtype = str(os.environ.get(ENV_KV_DTYPE) or _parse_kv_cache_dtype(extra_args) or "bf16").lower() + + conc = max(1, _as_int(_first_env_or(envs, "CONC", 64), 64)) + # A running batch cannot exceed the scheduler's cap on concurrent sequences, + # so a variant that lowers it lowers the batch the decode step actually runs. + max_seqs = _parse_server_arg_int(extra_args, "--max-num-seqs", "--max-running-requests") + if max_seqs and max_seqs > 0: + conc = min(conc, max_seqs) + + return ServingSpec( + framework=framework, + model_path=model_path, + tp=max(1, tp), + ep=max(1, ep), + pp=max(1, pp), + conc=conc, + isl=max(1, _as_int(_first_env_or(envs, "ISL", 1024), 1024)), + osl=max(1, _as_int(_first_env_or(envs, "OSL", 1024), 1024)), + weight_dtype=weight_dtype, + kv_cache_dtype=kv_dtype, + extra_server_args=extra_args, + ) + + +def resolve_preset(model_path: str) -> str | None: + """Best-effort map a model path/name to an InferaSim preset name.""" + explicit = os.environ.get(ENV_MODEL) + if explicit and explicit.strip(): + return explicit.strip() + key = re.sub(r"[^a-z0-9.]+", "-", (model_path or "").lower()) + for needle, preset in _MODEL_HEURISTICS: + if needle in key: + return preset + return None + + +@dataclass +class AnchorChoice: + """The warmup anchor selected for a candidate, plus why it was chosen. + + ``regime_distance`` is the Hamming distance over InferaSim's regime-defining + axes (model/dtypes/attention-backend/cudagraph/aiter). Distance 0 means the + candidate only moves along *transport* axes (TP/EP/PP, batch, concurrency, + sequence lengths) and is fully reconstructable from this anchor -- i.e. no + new GPU benchmark is needed. A non-zero distance means the candidate changes + the kernel regime and warrants a fresh warmup anchor. + """ + + path: str + regime_distance: int + model: str | None = None + needs_warmup: bool = False + real_weights: bool = False + served: bool = False + + +def recipe_from_spec(spec: ServingSpec) -> dict[str, Any]: + """Canonical InferaSim recipe dict for a Hyperloom serving spec.""" + attn = _parse_server_arg_str(spec.extra_server_args, "--attention-backend") + method, k = parse_speculative(spec.extra_server_args) + return { + "model": spec.model_path or None, + # The engines differ in scheduler, paging and kernel selection, so one + # engine's anchor cannot price another's run. Omitting this axis leaves + # it missing on our side, and regime_distance skips an axis missing on + # either side -- a vLLM and an SGLang anchor would both score 0 here. + "engine": spec.framework or None, + "weight_dtype": spec.weight_dtype, + "kv_cache_dtype": spec.kv_cache_dtype, + "moe_expert_dtype": None, + "attention_backend": attn, + "cudagraph": None, + "aiter": None, + # A speculating candidate emits >1 token per step, so an anchor measured + # without speculation cannot price it. Putting this on the recipe is what + # makes the regime signature reject such an anchor instead of silently + # reporting a plain-decode number for it. + "speculative": f"spec:{k}" if method else "off", + "tp": spec.tp, + "pp": spec.pp, + "ep": spec.ep, + "batch": spec.conc, + "concurrency": spec.conc, + "input_len": spec.isl, + "output_len": spec.osl, + } + + +def select_anchor(spec: ServingSpec) -> AnchorChoice | None: + """Pick the closest in-regime warmup anchor for ``spec``. + + Precedence: an explicit ``HYPERLOOM_INFERASIM_ANCHOR`` always wins; otherwise + an anchor store directory is searched for the nearest anchor in regime space. + Returns ``None`` when neither is configured (pure-analytical projection). + """ + explicit = os.environ.get(ENV_ANCHOR) + if explicit and Path(explicit).is_file(): + # Pinned by an operator, but still gated: a corrupt curve would silently + # propagate into every projection made from it. + if not anchor_curve_is_sane(explicit): + return None + return AnchorChoice(path=explicit, regime_distance=0, model=spec.model_path) + + store_root = os.environ.get(ENV_ANCHOR_STORE) + if not store_root or not Path(store_root).is_dir(): + return None + + _ensure_infera_importable() + try: + from infera.projection.core.projection.inference_projection.search.anchor_store import ( + AnchorStore, + ) + except Exception as exc: # noqa: BLE001 + raise InferasimBridgeError(f"cannot import InferaSim AnchorStore: {exc}") from exc + + store = AnchorStore(store_root) + recipe = recipe_from_spec(spec) + + from infera.projection.core.projection.inference_projection.search import regime + + # Only this model's warmups may calibrate this model. The names arrive in + # different spellings -- the spec carries a checkout path or a preset, the + # artifact carries a HuggingFace id -- so they are compared loosely rather + # than for equality, which never held and left the filter inert. + # + # An anchor with no recorded model is allowed: a structural warmup names no + # checkpoint, and the regime axes still have to agree before it is used. + # But when the store holds anchors and none of them are for this model, the + # answer is that there is no anchor. This used to fall back to using any + # anchor at all, which calibrated qwen3 against gpt-oss and returned the + # same decode step for every model in the sweep, labelled "calibrated". + # A target has two names and either one identifies it. The checkout path + # ("/models/DeepSeek-R1") usually shares a spelling with the artifact's id, + # while the preset names an architecture and can legitimately cover several + # checkpoints -- deepseek_v3 is the preset for DeepSeek-R1, and matching only + # on that would reject R1's own warmup. + names = [n for n in (spec.model_path, resolve_preset(spec.model_path)) if n] + entries = [ + e for e in store.entries() if not e.get("model") or any(regime.models_match(n, e["model"]) for n in names) + ] + if not entries: + return None + + def rank(entry: dict[str, Any]) -> tuple[int, int, int, float]: + """Sort key: regime distance, fidelity, provenance, transport closeness. + + Fidelity matters as much as regime here: a dummy-weight anchor runs the + same kernels but with synthetic MoE routing, so its decode curve is much + flatter than a real-weights run. Ranking it below a real-weights anchor + in the same regime is the difference between ~2% and ~30% error against + measured serving. Provenance is the same argument one level down -- an + offline anchor does not even run the served kernels. + """ + dist = regime.regime_distance(recipe, dict(entry.get("regime") or {})) + real = _anchor_is_real_weights(entry["path"]) + served = _anchor_is_served(entry["path"]) + transport = entry.get("transport") or {} + gap = 0.0 + for axis in ("tp", "ep", "pp"): + av, rv = transport.get(axis), recipe.get(axis) + if av and rv: + gap += abs(float(av) - float(rv)) + return (dist, 0 if real else 1, 0 if served else 1, gap) + + usable = [e for e in entries if anchor_curve_is_sane(e["path"])] + if not usable: + return None + best = min(usable, key=rank) + dist, fidelity_rank, served_rank, _ = rank(best) + return AnchorChoice( + path=best["path"], + regime_distance=int(dist), + model=best.get("model"), + needs_warmup=bool(dist), + real_weights=(fidelity_rank == 0), + served=(served_rank == 0), + ) + + +# A decode step at a larger batch does strictly more work, so measured decode +# latency must not fall as batch rises. Small drops are ordinary run-to-run +# noise; a large one means the measurement itself is broken (the harness times +# decode by differencing two generate() calls, which degenerates when the longer +# call is not actually longer). Such an artifact is not merely imprecise -- it +# poisons every projection that anchors on it, so it is rejected outright. +_ANCHOR_MONOTONIC_TOLERANCE = 0.15 + + +def anchor_curve_is_sane(path: str) -> bool: + """False when an artifact's measured decode curve is physically impossible.""" + try: + with open(path) as fh: + doc = json.load(fh) or {} + except (OSError, ValueError): + return False + if not isinstance(doc, dict): + return False + points = [] + for entry in doc.get("sweep") or []: + try: + batch, decode_ms = int(entry["batch"]), float(entry["decode_ms"]) + except (KeyError, TypeError, ValueError): + continue + if decode_ms <= 0.0: + return False + points.append((batch, decode_ms)) + if not points: + return False + # A single-batch anchor is narrow, not invalid: there is no monotonicity to + # check, and it still calibrates the one operating point it measured. + points.sort() + peak = points[0][1] + for _, decode_ms in points[1:]: + if decode_ms < peak * (1.0 - _ANCHOR_MONOTONIC_TOLERANCE): + return False + peak = max(peak, decode_ms) + return True + + +def _anchor_is_served(path: str) -> bool: + """True when an anchor was measured against a real server, not offline vLLM. + + The distinction is not cosmetic. Given identical flags the offline ``LLM()`` + entrypoint and ``vllm serve`` resolve different attention and MoE kernels, + and the served decode step ran 1.9x the offline one at concurrency 8 and + 5.5x at 128. Calibrating a served target from an offline anchor scored no + better than not calibrating at all, where a served anchor scored 2.2%. + """ + try: + with open(path) as fh: + meta = (json.load(fh) or {}).get("meta") or {} + except (OSError, ValueError): + return False + return "serving" in str(meta.get("derived_from") or "").lower() + + +def _anchor_is_real_weights(path: str) -> bool: + """True when an anchor artifact was measured with real checkpoint weights.""" + try: + with open(path) as fh: + meta = (json.load(fh) or {}).get("meta") or {} + except (OSError, ValueError): + return False + if meta.get("real_weights") is not None: + return bool(meta["real_weights"]) + return str(meta.get("load_format") or "").lower() not in ("dummy", "") + + +def _resolve_workload_and_env(spec: ServingSpec) -> tuple[str, dict[str, str]]: + """Return (workload_yaml_path, extra_env) for the projection. + + Precedence: an explicit ``HYPERLOOM_INFERASIM_WORKLOAD`` wins; otherwise a + resolved preset name is fed to the bundled env-driven template via + ``INFERASIM_MODEL``. + """ + extra_env: dict[str, str] = {} + explicit = os.environ.get(ENV_WORKLOAD) + if explicit and Path(explicit).is_file(): + return str(Path(explicit).resolve()), extra_env + + preset = resolve_preset(spec.model_path) + if not preset: + raise InferasimBridgeError( + f"could not resolve an InferaSim model preset for model={spec.model_path!r}; " + f"set {ENV_MODEL}= or {ENV_WORKLOAD}=" + ) + if not _TEMPLATE_WORKLOAD.is_file(): + raise InferasimBridgeError(f"bundled workload template missing: {_TEMPLATE_WORKLOAD}") + # The template reads INFERASIM_MODEL/TP/PP/EP; parallelism is *also* forced via + # CLI overrides below so an explicit workload YAML is honored too. + extra_env["INFERASIM_MODEL"] = preset + return str(_TEMPLATE_WORKLOAD), extra_env + + +def _purge_foreign_infera(root: str) -> None: + """Drop cached ``infera*`` modules not originating from ``root``. + + Another ``infera`` checkout may already be importable on the default path + (and may lack the ``projection`` subpackage). Once imported it is cached in + ``sys.modules``, so a later ``sys.path`` insert cannot override the + top-level package. Purge any cached ``infera`` whose file is outside our + root so the re-import resolves against ``HYPERLOOM_INFERASIM_ROOT``. + """ + root_resolved = str(Path(root).resolve()) + for name in list(sys.modules): + if name != "infera" and not name.startswith("infera."): + continue + mod = sys.modules.get(name) + origin = getattr(mod, "__file__", None) or "" + paths = list(getattr(mod, "__path__", []) or []) + located = origin or (paths[0] if paths else "") + if not located or not str(Path(located).resolve()).startswith(root_resolved): + sys.modules.pop(name, None) + + +def _ensure_infera_importable() -> None: + """Make ``infera.projection`` importable, honoring HYPERLOOM_INFERASIM_ROOT. + + When ``HYPERLOOM_INFERASIM_ROOT`` is set it takes precedence over any other + ``infera`` on the path so the pinned Infera checkout is the one projected + against. + """ + root = os.environ.get(ENV_ROOT) + if root and Path(root).is_dir(): + if sys.path[:1] != [root]: + while root in sys.path: + sys.path.remove(root) + sys.path.insert(0, root) + _purge_foreign_infera(root) + + try: + import infera.projection # noqa: F401 + + return + except Exception as exc: # noqa: BLE001 + raise InferasimBridgeError( + f"cannot import Infera 'infera.projection' (set {ENV_ROOT} to the Infera " + f"checkout or pip install amd-infera[projection]): {exc}" + ) from exc + + +def _build_argv(spec: ServingSpec, workload: str, anchor: AnchorChoice | None = None) -> list[str]: + """Build the ``inferasim inference`` argv for this serving spec.""" + gpu_arch = str(os.environ.get(ENV_GPU_ARCH) or _DEFAULT_GPU_ARCH).lower() + hbm_gb = os.environ.get(ENV_HBM_GB) or _ARCH_HBM_GB.get(gpu_arch) + serving_model = str(os.environ.get(ENV_SERVING_MODEL) or "continuous").lower() + + argv: list[str] = [ + "inference", + "--config", + workload, + "--inference-mode", + "both", + "--profiling-mode", + "simulate", + "--serving-model", + serving_model, + "--input-len", + str(spec.isl), + "--output-len", + str(spec.osl), + "--inference-batch-size", + str(spec.conc), + "--max-concurrency", + str(spec.conc), + "--weight-dtype", + spec.weight_dtype, + "--kv-cache-dtype", + spec.kv_cache_dtype, + "--gpu-arch", + gpu_arch, + ] + # Name the engine so InferaSim refuses a cross-engine anchor on its own, + # rather than leaving ``select_anchor`` as the only thing standing between + # a vLLM measurement and an SGLang candidate. Simulate mode is analytical + # and returns the same number whichever engine is named, so this only ever + # gates calibration -- which is what ``--load-benchmark`` below asks for. + if spec.framework: + argv += ["--serving-engine", spec.framework] + if hbm_gb: + argv += ["--hbm-capacity-gb", str(hbm_gb)] + + if anchor is not None and Path(anchor.path).is_file(): + argv += ["--load-benchmark", anchor.path] + argv += ["--profiling-mode", "both"] # calibrate + report source + scaling = os.environ.get(ENV_ANCHOR_SCALING) + if scaling: + for path in [p.strip() for p in scaling.split(",") if p.strip()]: + argv += ["--load-benchmark-scaling", path] + + # Force parallelism via config overrides so an explicit workload YAML is + # honored regardless of its baked-in values. + argv += [ + f"tensor_model_parallel_size={spec.tp}", + f"expert_model_parallel_size={spec.ep}", + f"pipeline_model_parallel_size={spec.pp}", + ] + return argv + + +def project(spec: ServingSpec) -> ProjMetrics: + """Run the InferaSim projection for ``spec`` and return mapped metrics.""" + _ensure_infera_importable() + workload, extra_env = _resolve_workload_and_env(spec) + + from infera.projection.cli import build_parser + from infera.projection.core.projection.inference_projection import ( + launch_projection_from_cli, + ) + + anchor = select_anchor(spec) + argv = _build_argv(spec, workload, anchor) + # Template reads INFERASIM_* env; also expose TP/PP/EP for template default + # interpolation (overrides above still win for explicit workloads). + prev_env: dict[str, str | None] = {} + inject = dict(extra_env) + inject.setdefault("INFERASIM_TP", str(spec.tp)) + inject.setdefault("INFERASIM_PP", str(spec.pp)) + inject.setdefault("INFERASIM_EP", str(spec.ep)) + for key, val in inject.items(): + prev_env[key] = os.environ.get(key) + os.environ[key] = val + try: + args, overrides = build_parser().parse_known_args(argv) + results = launch_projection_from_cli(args, overrides) + except InferasimBridgeError: + raise + except Exception as exc: # noqa: BLE001 + raise InferasimBridgeError(f"InferaSim projection failed: {exc}") from exc + finally: + for key, val in prev_env.items(): + if val is None: + os.environ.pop(key, None) + else: + os.environ[key] = val + + perf = results.get("performance") + if perf is None: + raise InferasimBridgeError("InferaSim returned no performance projection") + mem = results.get("memory") + + return _metrics_from_results(spec, perf, mem, anchor) + + +def _metrics_from_results(spec: ServingSpec, perf: Any, mem: Any, anchor: AnchorChoice | None = None) -> ProjMetrics: + """Map InferaSim result objects onto benchmark measurement fields.""" + output_tps = float(getattr(perf, "decode_throughput_tps", 0.0) or 0.0) + osl = max(1, spec.osl) + isl = max(1, spec.isl) + request_tps = output_tps / osl if osl else 0.0 + total_tps = output_tps * (isl + osl) / osl if osl else output_tps + + mem_gb = 0.0 + if mem is not None: + total_bytes = float(getattr(mem, "total_bytes", 0) or 0) + mem_gb = total_bytes / (1024.0**3) + extras = dict(getattr(perf, "extras", {}) or {}) + max_conc = int(extras.get("concurrency_used", 0) or extras.get("concurrency", 0) or spec.conc) + if anchor is not None: + # Provenance so a session can audit which warmup anchor served this + # candidate and whether it stayed inside the anchor's regime. + extras["anchor_path"] = anchor.path + extras["anchor_regime_distance"] = anchor.regime_distance + extras["anchor_needs_warmup"] = anchor.needs_warmup + extras["anchor_real_weights"] = anchor.real_weights + extras["anchor_served"] = anchor.served + + replica_gpus = int(getattr(perf, "replica_gpus", 0) or 0) + extras["extrapolation"] = extrapolation_notes(spec, replica_gpus, bool(extras.get("benchmark_calibrated", 0.0))) + + return ProjMetrics( + output_throughput=output_tps, + request_throughput=request_tps, + total_token_throughput=total_tps, + ttft_ms=float(getattr(perf, "ttft_ms", 0.0) or 0.0), + tpot_ms=float(getattr(perf, "itl_ms", 0.0) or 0.0), + itl_ms=float(getattr(perf, "itl_ms", 0.0) or 0.0), + e2el_ms=float(getattr(perf, "request_latency_ms", 0.0) or 0.0), + decode_tps_per_gpu=float(getattr(perf, "decode_throughput_tps_per_gpu", 0.0) or 0.0), + memory_per_gpu_gb=mem_gb, + max_concurrency=max_conc, + calibrated=bool(extras.get("benchmark_calibrated", 0.0)), + replica_gpus=replica_gpus, + extras=extras, + ) + + +def raw_result_from_metrics(spec: ServingSpec, m: ProjMetrics) -> dict[str, Any]: + """Build an InferenceX-style flat result dict from projected metrics. + + Keys mirror what ``bypass_report.build_report`` / Magpie's result parser + read, so the simulated run flows through Hyperloom's collectors unchanged. + Percentiles are set to the mean (the point projection has no distribution; + use the DES path for tails). + """ + completed = max(1, m.max_concurrency) + duration = (m.e2el_ms / 1000.0) if m.e2el_ms else 0.0 + return { + "model_id": spec.model_path, + "request_throughput": m.request_throughput, + "output_throughput": m.output_throughput, + "total_token_throughput": m.total_token_throughput, + "completed": completed, + "total_input_tokens": completed * spec.isl, + "total_output_tokens": completed * spec.osl, + "duration": duration, + "mean_ttft_ms": m.ttft_ms, + "median_ttft_ms": m.ttft_ms, + "p99_ttft_ms": m.ttft_ms, + "std_ttft_ms": 0.0, + "mean_tpot_ms": m.tpot_ms, + "median_tpot_ms": m.tpot_ms, + "p99_tpot_ms": m.tpot_ms, + "std_tpot_ms": 0.0, + "mean_itl_ms": m.itl_ms, + "median_itl_ms": m.itl_ms, + "p99_itl_ms": m.itl_ms, + "std_itl_ms": 0.0, + "mean_e2el_ms": m.e2el_ms, + "median_e2el_ms": m.e2el_ms, + "p99_e2el_ms": m.e2el_ms, + "std_e2el_ms": 0.0, + # Non-Magpie diagnostics, carried for inspection/reporting. + "inferasim_decode_tps_per_gpu": m.decode_tps_per_gpu, + "inferasim_memory_per_gpu_gb": m.memory_per_gpu_gb, + "inferasim_calibrated": m.calibrated, + "inferasim_extrapolation": list(m.extras.get("extrapolation") or []), + "inferasim_replica_gpus": m.replica_gpus, + "inferasim_tp": spec.tp, + "inferasim_ep": spec.ep, + "inferasim_pp": spec.pp, + } diff --git a/src/hyperloom/orchestrator/actions/executors/inferasim_runner.py b/src/hyperloom/orchestrator/actions/executors/inferasim_runner.py new file mode 100644 index 0000000000..4e39c9edb4 --- /dev/null +++ b/src/hyperloom/orchestrator/actions/executors/inferasim_runner.py @@ -0,0 +1,170 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""InferaSim benchmark runner (CLI). + +A simulate-only, GPU-free stand-in for ``python -m Magpie -v benchmark ... +--run-mode local``. It accepts the same CLI flags and writes the same +Magpie-compatible workspace + ``benchmark_report.json`` (via +:mod:`bypass_report`), but the numbers come from Infera's ``inferasim`` serving +projection instead of a real server + client. + +Selected with ``HYPERLOOM_BENCHMARK_BACKEND=inferasim``. Because it emits the +same report contract as Magpie/bypass, every executor, collector, and the +optimizer's gain math consume simulated runs unchanged -- so an entire +optimization session can run without a GPU, and real GPU time is spent only on +the final validation. That is the GPU-time reduction projected in the deck. + +Lifecycle/server flags (``--phase``, ``--server-lifecycle-*``) are accepted for +drop-in compatibility and ignored: a projection has no server to persist. +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +from pathlib import Path +from typing import Any + +import yaml + +from . import bypass_report +from . import inferasim_bridge + + +def run_benchmark(config_path: Path, output_dir: Path) -> int: + """Project a serving config with InferaSim and write a Magpie-style report. + + Args: + config_path: Materialized benchmark config YAML (the Magpie contract). + output_dir: Output root for the benchmark workspace. + + Returns: + Process exit code (0 on success). + """ + start = time.time() + try: + cfg = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {} + except (OSError, yaml.YAMLError) as exc: + return _emit_failure(output_dir, "unknown", "", f"cannot read benchmark config: {exc}", start) + + bench = cfg.get("benchmark") or {} + framework = str(bench.get("framework") or "sglang").lower() + model = str(bench.get("model") or "") + + try: + spec = inferasim_bridge.spec_from_benchmark(bench) + metrics = inferasim_bridge.project(spec) + except inferasim_bridge.InferasimBridgeError as exc: + return _emit_failure(output_dir, framework, model, str(exc), start) + except Exception as exc: # noqa: BLE001 - never crash the optimizer loop + return _emit_failure(output_dir, framework, model, f"unexpected InferaSim error: {exc}", start) + + workspace = bypass_report.create_workspace(output_dir, framework) + _snapshot_config(workspace, cfg) + raw = inferasim_bridge.raw_result_from_metrics(spec, metrics) + # Persist the raw InferenceX-style result so the workspace matches a real + # bypass/Magpie run (collectors that rescan raw json stay consistent). + try: + (workspace / "inferencex_result.json").write_text(json.dumps(raw, indent=2), encoding="utf-8") + except OSError: + # Best-effort artifact: keep the run successful even if this auxiliary + # file cannot be written; benchmark_report.json remains the source of truth. + pass + + report = bypass_report.build_report( + raw, + framework=framework, + model=model, + success=True, + workspace_dir=str(workspace), + execution_time=time.time() - start, + errors=[], + analysis={ + "backend": "inferasim", + "source": "calibrated" if metrics.calibrated else "simulation", + "extrapolation": list(metrics.extras.get("extrapolation") or []), + "decode_tps_per_gpu": metrics.decode_tps_per_gpu, + "memory_per_gpu_gb": metrics.memory_per_gpu_gb, + "max_concurrency": metrics.max_concurrency, + "replica_gpus": metrics.replica_gpus, + "tp": spec.tp, + "ep": spec.ep, + "pp": spec.pp, + "isl": spec.isl, + "osl": spec.osl, + }, + profiling_enabled=False, + ) + bypass_report.write_report(workspace, report) + return 0 + + +def _snapshot_config(workspace: Path, cfg: dict[str, Any]) -> None: + """Persist the effective config into the workspace (best-effort).""" + try: + (workspace / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8") + except OSError: + # Best-effort snapshot: reporting must continue even if config persistence fails. + pass + + +def _emit_failure(output_dir: Path, framework: str, model: str, error: str, start: float) -> int: + """Emit a failing report + workspace for a pre-projection error.""" + workspace = bypass_report.create_workspace(output_dir, framework) + report = bypass_report.build_report( + None, + framework=framework, + model=model, + success=False, + workspace_dir=str(workspace), + execution_time=time.time() - start, + errors=[error], + profiling_enabled=False, + ) + bypass_report.write_report(workspace, report) + return 1 + + +def _build_arg_parser() -> argparse.ArgumentParser: + """Build the InferaSim runner parser (Magpie-compatible flags).""" + parser = argparse.ArgumentParser(prog="hyperloom-inferasim-benchmark") + sub = parser.add_subparsers(dest="mode", required=True) + bench = sub.add_parser("benchmark", help="Project a serving config with InferaSim") + bench.add_argument("--benchmark-config", required=True) + bench.add_argument("--output-dir", required=True) + bench.add_argument("--run-mode", default="local") + # Accepted for drop-in parity with Magpie/bypass; ignored (no server). + bench.add_argument("--phase", default="all", choices=["all", "server", "client"]) + bench.add_argument("--server-lifecycle-pid-dir", default=None) + bench.add_argument("--server-lifecycle-cleanup", default="true") + return parser + + +def main(argv: list[str] | None = None) -> int: + """CLI entry point mirroring the bypass runner. + + Args: + argv: Optional argument vector (defaults to ``sys.argv[1:]``). + + Returns: + Process exit code. + """ + args = _build_arg_parser().parse_args(argv) + if args.mode != "benchmark": + print(f"unsupported mode: {args.mode}", file=sys.stderr) + return 2 + if args.run_mode != "local": + print(f"inferasim runner supports --run-mode local only, got {args.run_mode}", file=sys.stderr) + return 2 + if args.phase == "server": + # No persistent server exists for a projection; a lone server phase is a + # no-op success so lifecycle-driven callers don't stall. + return 0 + return run_benchmark(Path(args.benchmark_config), Path(args.output_dir)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/hyperloom/orchestrator/actions/executors/recover.py b/src/hyperloom/orchestrator/actions/executors/recover.py index bc5be902be..031f13d8e9 100644 --- a/src/hyperloom/orchestrator/actions/executors/recover.py +++ b/src/hyperloom/orchestrator/actions/executors/recover.py @@ -98,11 +98,19 @@ async def __call__(self, ctx: RunnerContext) -> dict[str, Any]: pre = await asyncio.to_thread(self._probe_gpu_free_mb) # 2) Soft cleanup — TERM/KILL stale owners. + from ._multi_node_env import uses_external_server + killed: list[dict[str, Any]] = [] - if force_cleanup: - killed = await asyncio.to_thread(self._kill_stale_owners) - else: + if not force_cleanup: log.info("recover_executor: force_gpu_cleanup=false; skipping kill stage") + elif uses_external_server(): + # OWNER_PATTERNS match by cmdline, so the engine behind an external + # endpoint would be TERM/KILLed here even though we never launched + # it -- restarting the server the benchmark is measuring. The GPU + # probe above still applies: single-node, those GPUs are local. + log.info("recover_executor: external server; skipping kill stage (engine is not ours)") + else: + killed = await asyncio.to_thread(self._kill_stale_owners) # 3) Probe after kills. mid = await asyncio.to_thread(self._probe_gpu_free_mb) diff --git a/src/hyperloom/orchestrator/kernel/_kernel_decisions.py b/src/hyperloom/orchestrator/kernel/_kernel_decisions.py index 1c630a1f0f..3fdd7d55d7 100644 --- a/src/hyperloom/orchestrator/kernel/_kernel_decisions.py +++ b/src/hyperloom/orchestrator/kernel/_kernel_decisions.py @@ -102,6 +102,75 @@ def _kernel_integration_id( return f"kernel-integration:{digest}" +# A kernel that is ``f`` of GPU time and got ``S`` times faster cannot make the +# whole deployment more than ``1/((1-f) + f/S)`` faster. That is arithmetic, not +# a model, and it is an upper bound twice over: it assumes the kernel's time is +# entirely on the critical path, and that nothing else got slower. +# +# So when the bound is below the threshold the integrate run has to clear, the +# run cannot pass, and it is a full end-to-end serving benchmark spent to +# rediscover that. Skipping it cannot lose an optimization -- there was none to +# lose. +# +# The margin is what makes this safe in practice rather than only on paper. The +# GPU share comes from a trace and a kernel speedup can relieve second-order +# pressure the bound does not model, so the share is inflated before the bound +# is taken and only a comfortable failure is skipped. +AMDAHL_GPU_PCT_MARGIN = 1.5 + +# Set to 0 to queue every micro-KEEP for integrate regardless of its ceiling. +_AMDAHL_GATE_ENV = "HYPERLOOM_KERNEL_AMDAHL_GATE" + + +def amdahl_e2e_ceiling_pct(gpu_pct: float, micro_speedup: float) -> float | None: + """Best end-to-end gain a kernel speedup could produce, in percent. + + Args: + gpu_pct: The kernel's share of GPU time, in percent. + micro_speedup: Measured microbenchmark speedup, as a ratio. + + Returns: + The ceiling as a percentage gain, or ``None`` when either input is + missing or out of range -- in which case no gate should be applied. + """ + try: + f = float(gpu_pct) / 100.0 + s = float(micro_speedup) + except (TypeError, ValueError): + return None + if not (0.0 < f <= 1.0) or s <= 0.0: + return None + denom = (1.0 - f) + f / s + if denom <= 0.0: + return None + return (1.0 / denom - 1.0) * 100.0 + + +def integrate_cannot_pass( + gpu_pct: float, + micro_speedup: float, + keep_threshold_pct: float, + *, + margin: float = AMDAHL_GPU_PCT_MARGIN, +) -> bool: + """Is the integrate run arithmetically incapable of clearing its bar? + + Answers ``False`` -- run it -- whenever the inputs do not support a + confident answer, so the failure mode is the benchmark we already run. + """ + if keep_threshold_pct <= 0: + return False + ceiling = amdahl_e2e_ceiling_pct(float(gpu_pct or 0.0) * margin, micro_speedup) + if ceiling is None: + return False + return ceiling < keep_threshold_pct + + +def _amdahl_gate_enabled() -> bool: + raw = os.environ.get(_AMDAHL_GATE_ENV) + return (raw if raw is not None else "1").strip().lower() not in {"0", "false", "no", "off"} + + def _queue_kernel_keep( state, *, @@ -139,6 +208,27 @@ def _queue_kernel_keep( ) if decision != "KEEP" and not promoted_needs_review: return None + # Amdahl gate: a full end-to-end serving benchmark that cannot arithmetically + # clear its own bar is GPU time spent to confirm a foregone conclusion. + gate_threshold = float(getattr(state, "kernel_integrate_keep_threshold_pct", None) or 1.0) + if _amdahl_gate_enabled() and integrate_cannot_pass(entry.get("last_gpu_pct", 0.0), micro_speedup, gate_threshold): + ceiling = amdahl_e2e_ceiling_pct( + float(entry.get("last_gpu_pct", 0.0) or 0.0) * AMDAHL_GPU_PCT_MARGIN, + micro_speedup, + ) + log.info( + "kernel %s: skipping integrate; %.1f%% of GPU time at %.2fx caps the " + "end-to-end gain at %.2f%%, under the %.2f%% bar even after a %.1fx " + "margin on the share", + kernel_id, + float(entry.get("last_gpu_pct", 0.0) or 0.0), + micro_speedup, + ceiling if ceiling is not None else float("nan"), + gate_threshold, + AMDAHL_GPU_PCT_MARGIN, + ) + entry["last_integrate_skip_reason"] = "amdahl_ceiling_below_threshold" + return None artifact_path = str(entry.get("last_artifact_path") or "") artifact_bundle = dict(entry.get("last_artifact_bundle") or {}) source_file = str(entry.get("last_source_file") or "") diff --git a/src/hyperloom/orchestrator/phases/machine_state.py b/src/hyperloom/orchestrator/phases/machine_state.py index 3d3e75b145..9b07d636dd 100644 --- a/src/hyperloom/orchestrator/phases/machine_state.py +++ b/src/hyperloom/orchestrator/phases/machine_state.py @@ -324,6 +324,17 @@ def is_valid_phase_exit_reason(value: str) -> bool: import os as _os_env # noqa: E402 + +def _plateau_round_window_enabled() -> bool: + """Whether the EXPLORE plateau reads recent rounds instead of recent winners. + + Set ``HYPERLOOM_EXPLORE_PLATEAU_ROUND_WINDOW=0`` to restore the window over + the last ``lookback`` winners and the no-proposal streak. + """ + raw = (_os_env.environ.get("HYPERLOOM_EXPLORE_PLATEAU_ROUND_WINDOW") or "").strip() + return (raw if raw else "1").lower() not in {"0", "false", "no", "off"} + + # FRAMEWORK per-candidate plateau: after this many consecutive resolved candidates without a KEEP (including # non-benchmarked terminal outcomes), the source arm is dry. DEFAULT_FRAMEWORK_PLATEAU_NO_KEEP_STREAK: int = 5 @@ -1035,12 +1046,39 @@ def compute_plateau_explore( return False, {"reason": "lookback_disabled"} keep_gain_threshold_pct = float(keep_gain_threshold_pct or 0.0) empty_streak_threshold = int(empty_streak_threshold or 0) + round_window = _plateau_round_window_enabled() explore_search = getattr(state, "explore_search", None) or {} if not isinstance(explore_search, dict): explore_search = {} winners_history = _rows_for_current_cycle(explore_search.get("winners_history") or [], state) - recent_winners = list(winners_history[-lookback:]) + specialist_rounds = _rows_for_current_cycle(getattr(state, "specialist_rounds", None) or [], state) + + # Scope the gain window to the last ``lookback`` *rounds* rather than the + # last ``lookback`` winners. A winner is in this ledger only because it + # cleared the KEEP threshold, so a window over winners holds entries that + # are each >= that threshold and sums to more than the plateau floor no + # matter how long ago they were found. Taken literally that makes the gain + # arm unsatisfiable from the first KEEP onward, and since the trigger is an + # AND, one early win disables the plateau exit for the rest of the cycle. + # Scoped to rounds the sum falls to zero once recent rounds stop keeping + # anything, which is the question the arm was written to ask. + recent_round_ids = { + str(row.get("round_id")) + for row in specialist_rounds[-lookback:] + if isinstance(row, dict) and row.get("round_id") is not None + } + winners_carry_rounds = any(isinstance(w, dict) and w.get("round_id") is not None for w in winners_history) + # A ledger whose winners predate round attribution would read as an empty + # window and plateau the phase on missing data, so that case keeps the + # winner-window behaviour: the failure mode is exploring longer. + scoped_by_round = bool(round_window and recent_round_ids and (winners_carry_rounds or not winners_history)) + if scoped_by_round: + recent_winners = [ + w for w in winners_history if isinstance(w, dict) and str(w.get("round_id")) in recent_round_ids + ] + else: + recent_winners = list(winners_history[-lookback:]) recent_keep_gain = 0.0 for w in recent_winners: if not isinstance(w, dict): @@ -1051,7 +1089,14 @@ def compute_plateau_explore( except (TypeError, ValueError): continue - specialist_rounds = _rows_for_current_cycle(getattr(state, "specialist_rounds", None) or [], state) + def _kept_count(row: dict[str, Any]) -> int: + """KEEPs recorded by a specialist-round summary.""" + try: + return int( + row.get("proposals_kept") if row.get("proposals_kept") is not None else row.get("kept_count") or 0, + ) + except (TypeError, ValueError): + return 0 def _round_is_empty(row: Any) -> bool: """Return True when a specialist-round summary produced no work.""" @@ -1066,18 +1111,26 @@ def _round_is_empty(row: Any) -> bool: ) except (TypeError, ValueError): proposals = 0 - try: - kept = int( - row.get("proposals_kept") if row.get("proposals_kept") is not None else row.get("kept_count") or 0, - ) - except (TypeError, ValueError): - kept = 0 - return proposals == 0 and kept == 0 + return proposals == 0 and _kept_count(row) == 0 + + def _round_kept_nothing(row: Any) -> bool: + """Return True when a round ended without keeping anything. + + The streak arm is asking whether EXPLORE has stopped finding wins, and a + round that proposed a full grid and kept none of it answers that as + plainly as a round that proposed nothing. Counting only the latter means + a proposer that keeps generating losing candidates holds the streak at + zero indefinitely. + """ + if not isinstance(row, dict): + return False + return _kept_count(row) == 0 - # Walk from newest to oldest counting the trailing-empty streak. + # Walk from newest to oldest counting the trailing unproductive streak. + unproductive = _round_kept_nothing if round_window else _round_is_empty streak = 0 for row in reversed(specialist_rounds): - if _round_is_empty(row): + if unproductive(row): streak += 1 else: break @@ -1088,6 +1141,8 @@ def _round_is_empty(row: Any) -> bool: "keep_gain_threshold_pct": keep_gain_threshold_pct, "empty_streak": int(streak), "empty_streak_threshold": empty_streak_threshold, + "empty_streak_basis": "kept_nothing" if round_window else "no_proposals", + "gain_window": "recent_rounds" if scoped_by_round else "recent_winners", "lookback": int(lookback), "winners_seen": len(recent_winners), "specialist_rounds_seen": len(specialist_rounds),