fix: judge explore variants and warm recipes by what they change about the running launch

scripts/code_metrics_baseline.txt

@@ -708,7 +708,7 @@ module-lines src/hyperloom/orchestrator/actions/executors/_grid_runner.py 2052
 module-lines src/hyperloom/orchestrator/actions/executors/_kv_metrics.py 1591
 module-lines src/hyperloom/orchestrator/actions/executors/_workload_envs.py 1685
 module-lines src/hyperloom/orchestrator/actions/executors/baseline.py 4052
-module-lines src/hyperloom/orchestrator/actions/executors/explore.py 1845
+module-lines src/hyperloom/orchestrator/actions/executors/explore.py 1793
 module-lines src/hyperloom/orchestrator/actions/executors/profile.py 1388
 module-lines src/hyperloom/orchestrator/actions/executors/report.py 1342
 module-lines src/hyperloom/orchestrator/actions/executors/roofline.py 1306

src/hyperloom/inference_optimizer/tests/test_longrun_phase2.py

@@ -197,6 +197,18 @@ def test_acceptance_threshold_advisory_lists_unblocked(cyclic_coordinator):
     assert "Historical results are evidence only" in block
 
 
+def test_a_row_answered_without_measuring_is_no_negative_ledger_pressure(cyclic_coordinator):
+    c = cyclic_coordinator
+    c.shared_state.explore_search = {
+        "tested": {
+            "fp1": {"name": "v_revert", "outcome": "REVERT", "domain": "serving_specialist", "cycle": 0},
+            "fp2": {"name": "v_restates", "outcome": "SKIPPED_DEDUP", "domain": "serving_specialist", "cycle": 0},
+        },
+        "rejected": [],
+    }
+    assert c.phase_macro_cycle._negative_ledger_domain_counts() == {"serving_specialist": 1}
+
+
 def test_acceptance_threshold_advisory_empty_first_cycle(cyclic_coordinator):
     c = cyclic_coordinator
     st = c.shared_state

src/hyperloom/inference_optimizer/tests/test_shared_state_renderers.py

@@ -139,6 +139,22 @@ def test_prompt_summary_renders_current_best_without_stack_or_measurement():
     assert "stacked" not in line and "launch_evidence" not in line
 
 
+def _launch_flags_line(state: SharedState) -> str:
+    return next(ln for ln in state.to_prompt_summary().splitlines() if ln.startswith("current_launch_flags"))
+
+
+def test_prompt_summary_renders_the_flags_the_current_best_launched_with():
+    """Orchestration cannot read the session config, so the summary is its only account of the running launch."""
+    st = SharedState()
+    assert _launch_flags_line(st) == "current_launch_flags(already passed by the running stack)=(unknown)"
+    st.current_best_measurement = {
+        "launch_evidence": {"requested_server_args": "--max-num-seqs 64 --kv-cache-dtype fp8 --max-num-seqs 128"}
+    }
+    assert _launch_flags_line(st) == (
+        "current_launch_flags(already passed by the running stack)=--max-num-seqs 128 --kv-cache-dtype fp8"
+    )
+
+
 def test_format_last_trace_analyze():
     st = SharedState()
     assert st._format_trace_analyze_blob(st.last_trace_analyze) == "(none)"

src/hyperloom/inference_optimizer/tests/test_warm_replay.py

@@ -763,6 +763,35 @@ async def test_warm_replay_skips_a_recipe_that_restates_the_baseline(tmp_path):
     assert persisted["warm_replay_attempted"] is True
 
 
+@pytest.mark.asyncio
+async def test_warm_replay_skips_a_recipe_an_agentx_baseline_already_launches_with(tmp_path):
+    """AgentX carries the baseline's server args in its launch overrides, not in ``benchmark.envs``."""
+    recipe = _warm_recipe_t1(extra_server_args="--kv-cache-dtype fp8", extra_envs={"SGLANG_TEST": "1"})
+    coord = _make_coord(tmp_path, warm_start_recipe=recipe)
+    config = tmp_path / "baseline_config.with_envs.yaml"
+    launch = {"version": 1, "append_args": ["--kv-cache-dtype", "fp8"], "env": {"SGLANG_TEST": "1"}}
+    config.write_text(
+        json.dumps(
+            {
+                "benchmark": {
+                    "framework": "sglang",
+                    "model": "Qwen/Qwen3-0.6B",
+                    "agentx": {"enabled": True, "mode": "canonical", "launch_overrides": launch},
+                    "envs": {"TP": "1", "EP_SIZE": "1", "CONC": "8"},
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+    coord.shared_state.baseline_config_path = str(config)
+
+    assert await coord.phase_prelude.maybe_enqueue_warm_replay(baseline_tput=600.0) is None
+
+    assert coord.tasks.calls == []
+    outcome = coord.shared_state.warm_replay_outcome
+    assert (outcome["status"], outcome["reason"]) == ("skipped", "restates_baseline")
+
+
 @pytest.mark.asyncio
 async def test_warm_replay_runs_a_recipe_that_changes_the_baseline(tmp_path):
     """One setting the baseline does not hold is enough for the replay to run."""

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

@@ -1,30 +1,30 @@
 # SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc.
 # SPDX-License-Identifier: MIT
 
-"""Whether a launch request only restates the baseline launch it would be appended to.
+"""What a candidate launch changes about the launch it is composed onto.
 
-Explore and the prelude warm replay both append a request to the baseline's
-server argv and envs. When every setting and env the request names already
-holds that value in the baseline, the launch is the baseline's own, and
-measuring it spends a whole benchmark round to reproduce the baseline number.
+Both launches are read back from configs the launch's own materializer built, so everything the server is
+started with -- argv, envs, removals, overlays, AgentX launch overrides -- is on one side or the other, and a
+candidate whose launch equals the stack's measures the stack again.
 """
 
 from __future__ import annotations
 
-import logging
+import json
 import shlex
-from collections.abc import Mapping
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 import yaml
 
+from hyperloom.common.benchmark_config import is_magpie_agentx
+from hyperloom.common.coerce import to_str_list
 from hyperloom.common.launch_log_evidence import launch_flag_setting_name
 from hyperloom.inference_optimizer.grid_server_args import server_args_env_name
 
-from ._server_argv import ConfigUnreadable, config_server_argv
-
-log = logging.getLogger(__name__)
+from ._server_argv import config_server_argv
 
 
 def config_envs(config_path: Path) -> dict[str, str]:
@@ -44,20 +44,35 @@ def config_envs(config_path: Path) -> dict[str, str]:
     return {str(k): str(v) for k, v in envs.items() if str(k) != args_env and v is not None}
 
 
+_NEGATION = "no_"
+_NEGATED = "false"
+
+
 def named_settings(argv_text: str, framework: str) -> dict[str, list[str]]:
-    """Map each setting ``argv_text`` names to its value tokens, last occurrence winning like the engine."""
-    try:
-        tokens = shlex.split(argv_text)
-    except ValueError:
-        return {}
+    """Map each setting ``argv_text`` names to its value tokens as the engine resolves them.
+
+    Raises:
+        ValueError: When the argv does not parse or opens on a positional token.
+    """
+    return _settings_of_tokens(shlex.split(argv_text), framework)
+
+
+def _settings_of_tokens(tokens: Sequence[str], framework: str) -> dict[str, list[str]]:
+    # Last occurrence wins, and a bare ``--no-x`` sets ``x`` to false, so both spellings land on one setting.
+    if tokens and not str(tokens[0]).startswith("-"):
+        raise ValueError(f"server argv opens on a positional token: {tokens[0]!r}")
     settings: dict[str, list[str]] = {}
     current = ""
-    for token in tokens:
+    for token in map(str, tokens):
         if token.startswith("-"):
             name, separator, attached = token.partition("=")
             current = launch_flag_setting_name(name, framework)
-            settings[current] = [attached] if separator else []
-        elif current:
+            if current.startswith(_NEGATION) and not separator:
+                current = current[len(_NEGATION) :]
+                settings[current] = [_NEGATED]
+            else:
+                settings[current] = [attached] if separator else []
+        else:
             settings[current].append(token)
     return settings
 
@@ -73,14 +88,18 @@ def observed_launch_after(
 
     The observed launch is what the server ran when the round opened. A KEEP that changed a setting, an env, a removal
     or an unset (added or cleared) leaves the observed value of that one, and of its negated spelling, stale; every
-    other setting is still what the server runs. Both stacks are ``measured_against`` shaped, with their args
+    other setting is still what the server runs. Nothing is known when either stack's args cannot be read. Both stacks are ``measured_against`` shaped, with their args
     composed as launched, so a change of args mode alone moves nothing.
     """
-    before = named_settings(str(opening.get("extra_server_args") or ""), framework)
-    after = named_settings(str(now.get("extra_server_args") or ""), framework)
-    stale = {name for name in before.keys() | after.keys() if before.get(name) != after.get(name)}
     removed = set(now.get("remove_args") or ()) ^ set(opening.get("remove_args") or ())
-    stale |= named_settings(" ".join(sorted(removed)), framework).keys()
+    try:
+        before = named_settings(str(opening.get("extra_server_args") or ""), framework)
+        after = named_settings(str(now.get("extra_server_args") or ""), framework)
+        removed_settings = named_settings(" ".join(sorted(removed)), framework).keys()
+    except ValueError:
+        return {}, {}
+    stale = {name for name in before.keys() | after.keys() if before.get(name) != after.get(name)}
+    stale |= removed_settings
     stale |= {_opposite_setting(name) for name in stale}
     envs_before, envs_after = opening.get("extra_envs") or {}, now.get("extra_envs") or {}
     stale_envs = {
@@ -98,53 +117,101 @@ def _opposite_setting(name: str) -> str:
     return name[3:] if name.startswith("no_") else f"no_{name}"
 
 
-def _names_only_flags(argv_text: str) -> bool:
-    """Whether ``argv_text`` parses and opens on a flag, so ``named_settings`` accounts for every token."""
-    try:
-        tokens = shlex.split(argv_text)
-    except ValueError:
-        return False
-    return not tokens or tokens[0].startswith("-")
+@dataclass(frozen=True)
+class Launch:
+    """The server a materialized config starts: its settings, its envs, and the controls neither of them shows."""
 
+    settings: Mapping[str, list[str]]
+    envs: Mapping[str, str]
+    controls: Mapping[str, Any]
 
-def restates_the_baseline(
-    argv_text: str,
-    envs: Mapping[str, Any],
-    base_argv: str,
-    base_envs: Mapping[str, str],
-    framework: str,
-) -> bool:
-    """Whether appending ``argv_text`` and ``envs`` to the baseline launch would change nothing.
-
-    True only when every setting the request names is already in ``base_argv``
-    with the same value tokens, and every env already carries that value in
-    ``base_envs``. A request this cannot read in full -- an unparsable argv, a
-    positional token, a flag whose negated spelling the baseline also names --
-    is never a restatement, and neither is an empty one.
+    @property
+    def identity(self) -> str:
+        """One string per distinct server launch."""
+        return json.dumps([self.settings, self.envs, self.controls], sort_keys=True)
+
+
+def read_launch(config_path: Path, unset_envs: Iterable[str] = ()) -> Launch:
+    """Read the launch a config built by the grid materializer starts, whichever layout it carries.
+
+    ``unset_envs`` are the names the run also pops from the inherited process env, which no config records.
+
+    Raises:
+        ConfigUnreadable, OSError, ValueError, yaml.YAMLError: When the config or its argv cannot be read.
     """
-    text = str(argv_text or "").strip()
-    requested_envs = {str(k): str(v) for k, v in (envs or {}).items()}
-    if (not text and not requested_envs) or not _names_only_flags(text):
-        return False
-    base_settings = named_settings(base_argv, framework)
-    for name, values in named_settings(text, framework).items():
-        if base_settings.get(name) != values or _opposite_setting(name) in base_settings:
-            return False
-    return all(base_envs.get(name) == value for name, value in requested_envs.items())
+    bench = yaml.safe_load(config_path.read_text(encoding="utf-8"))["benchmark"]
+    framework = str(bench.get("framework") or "")
+    controls: dict[str, Any] = {"unset_envs": sorted(set(to_str_list(unset_envs)))}
+    if not is_magpie_agentx(bench):
+        argv = config_server_argv(config_path).text
+        return Launch(named_settings(argv, framework), config_envs(config_path), controls)
+    overrides = (bench.get("agentx") or {}).get("launch_overrides") or {}
+    envs = {**config_envs(config_path), **{str(k): str(v) for k, v in (overrides.get("env") or {}).items()}}
+    controls.update(
+        remove_args=sorted(overrides.get("remove_args") or []),
+        replace_args=bool(overrides.get("replace_args")),
+        unset_env=sorted(overrides.get("unset_env") or []),
+        executable=str(overrides.get("executable") or ""),
+        source_files=dict(overrides.get("source_files") or {}),
+    )
+    return Launch(_settings_of_tokens(list(overrides.get("append_args") or []), framework), envs, controls)
 
 
-def config_launch_restated_by(config_path: str, argv_text: str, envs: Mapping[str, Any], framework: str) -> bool:
-    """Whether appending ``argv_text`` and ``envs`` to the config at ``config_path`` would change nothing.
+def launch_change(
+    base: Launch, launch: Launch, observed: tuple[Mapping[str, Any], Mapping[str, str]]
+) -> dict[str, Any]:
+    """What ``launch`` changes about ``base``, by kind; empty when both start the same server.
 
-    False when the config is absent or unreadable: the launch to compare
-    against is then unknown.
+    A setting or env set to the value the observed server already runs with is not a change, since the engine
+    resolves defaults no argv names. Removals and controls always are.
     """
-    if not str(config_path or "").strip():
+    observed_config, observed_env = observed
+    change = {
+        "settings": {
+            name: values
+            for name, values in launch.settings.items()
+            if base.settings.get(name) != values and not _already_resolved(name, values, observed_config)
+        },
+        "removed": sorted(base.settings.keys() - launch.settings.keys()),
+        "envs": {
+            name: value
+            for name, value in launch.envs.items()
+            if base.envs.get(name) != value and observed_env.get(name) != value
+        },
+        "unset_envs": sorted(base.envs.keys() - launch.envs.keys()),
+        "controls": sorted(
+            name
+            for name in base.controls.keys() | launch.controls.keys()
+            if base.controls.get(name) != launch.controls.get(name)
+        ),
+    }
+    return {kind: diff for kind, diff in change.items() if diff}
+
+
+def _already_resolved(name: str, values: list[str], observed_config: Mapping[str, Any]) -> bool:
+    """Whether the observed server already resolved setting ``name`` to ``values``."""
+    if name in observed_config and setting_already_holds(values, observed_config[name]):
+        return True
+    return values == [_NEGATED] and observed_config.get(_NEGATION + name) is True
+
+
+def setting_already_holds(proposed: list[str], observed: Any) -> bool:
+    """Whether ``observed`` is already the value ``proposed`` asks for."""
+    if not proposed:
+        # A bare flag asks for the feature on; anything else is a change.
+        return observed is True
+    if isinstance(observed, (list, tuple)):
+        return [str(item) for item in observed] == proposed
+    if len(proposed) != 1:
         return False
-    try:
-        base_argv = config_server_argv(config_path).text
-        base_envs = config_envs(Path(config_path))
-    except (ConfigUnreadable, OSError, ValueError, yaml.YAMLError) as exc:
-        log.warning("baseline config %s unreadable for the restatement check (%s)", config_path, exc)
+    text = proposed[0]
+    if isinstance(observed, bool):
+        return text.lower() == str(observed).lower()
+    if isinstance(observed, (int, float)):
+        try:
+            return float(text) == float(observed)
+        except ValueError:
+            return False
+    if observed is None:
         return False
-    return restates_the_baseline(argv_text, envs, base_argv, base_envs, framework)
+    return text == str(observed)