From 567d2cec4d87a1620c6edf903dfc3a5b80826247 Mon Sep 17 00:00:00 2001
From: lishuoshuo-amd <shuoli@amd.com>
Date: Tue, 27 Oct 2026 14:32:10 +0800
Subject: [PATCH 1/6] fix(explore): never judge a stack revalidation as a
 restatement of the stack

The baseline-noop filter exempted a variant whose provenance was
resume_stack_revalidate, a source #1739 renamed to stack_revalidate, so the
exemption never matched. A revalidation row re-passes the stack's own args,
which the observed launch already holds, so the round dropped it as a
baseline noop and the stack was never re-measured.

Exempt the round by its purpose (is_stack_revalidation) at both noop checks
and drop the dead provenance name.
---
 .../tests/test_explore_executor.py            | 24 +++++++++++++++++--
 .../orchestrator/actions/executors/explore.py | 20 +++++-----------
 2 files changed, 28 insertions(+), 16 deletions(-)

diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
index 40cf75e320..b19be0408b 100644
--- a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
+++ b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
@@ -684,7 +684,9 @@ def _fake_run(cmd, *args, **kwargs):
     assert rejected_provenance == {"llm_direct"}
 
 
-async def _explore_on_an_observed_fp8_stack(sub_agent_runner, tmp_path, monkeypatch, grid, tputs, inherited=""):
+async def _explore_on_an_observed_fp8_stack(
+    sub_agent_runner, tmp_path, monkeypatch, grid, tputs, inherited="", **params
+):
     """Run one explore round whose stack was observed launching with ``--kv-cache-dtype fp8``."""
     sub, tr, _ = sub_agent_runner
     _force_cold_decision(monkeypatch)
@@ -710,7 +712,13 @@ def _fake_run(cmd, *args, **kwargs):
     task, _ = await tr.create_or_return_existing(
         TaskSpec(
             kind="explore",
-            params={"config_path": str(base), "output_dir": str(tmp_path / "out"), "base_tput": 800.0, "grid": grid},
+            params={
+                "config_path": str(base),
+                "output_dir": str(tmp_path / "out"),
+                "base_tput": 800.0,
+                "grid": grid,
+                **params,
+            },
             idempotency_key="ex-noop-stack",
         ),
     )
@@ -723,6 +731,18 @@ def _fake_run(cmd, *args, **kwargs):
     return res.result, launched
 
 
+@pytest.mark.asyncio
+async def test_a_stack_revalidation_runs_although_it_restates_the_stack(sub_agent_runner, tmp_path, monkeypatch):
+    """A stack revalidation exists to re-measure the stack, so restating it is the point."""
+    grid = [{"name": "stack_revalidate", "extra_args": "--kv-cache-dtype fp8", "provenance": "stack_revalidate"}]
+    out, launched = await _explore_on_an_observed_fp8_stack(
+        sub_agent_runner, tmp_path, monkeypatch, grid, {"stack_revalidate": 800.0}, source="stack_revalidate"
+    )
+
+    assert launched == ["stack_revalidate"]
+    assert out["skipped_dup"] == []
+
+
 @pytest.mark.asyncio
 async def test_a_variant_that_was_the_stack_runs_once_a_keep_has_moved_the_stack(
     sub_agent_runner, tmp_path, monkeypatch
diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py
index f83ffdbd91..842e1c0ab0 100644
--- a/src/hyperloom/orchestrator/actions/executors/explore.py
+++ b/src/hyperloom/orchestrator/actions/executors/explore.py
@@ -519,21 +519,10 @@ def _touches_a_multi_value_flag(argv_text: str) -> bool:
     return any(flag in argv_text for flag in _MULTI_VALUE_FLAGS)
 
 
-# Separate from _CONFIG_REPLAY_PROVENANCE (shared with filter_operator_pinned_envs):
-# a revalidation re-measures its own config, so it must always run.
-_NOOP_FILTER_ALWAYS_RUNS = frozenset({"resume_stack_revalidate"})
-
 #: Names the throwaway variant that materializes the unchanged baseline to diff against.
 _NOOP_PROBE_BASE_NAME = "__baseline_noop_probe_base__"
 
 
-def _is_noop_filter_exempt(variant: GridVariant) -> bool:
-    """Whether ``variant`` must skip the baseline-noop probe outright and always run."""
-    if _is_config_replay_variant(variant):
-        return True
-    return str(getattr(variant, "provenance", "") or "").strip() in _NOOP_FILTER_ALWAYS_RUNS
-
-
 def _materialized_launch(recipe: GridRecipe, gv: GridVariant, output_subdir: Path) -> tuple[str, dict[str, str]]:
     """The server argv and benchmark envs ``run_grid`` would launch *gv* with."""
     out_path = _build_variant_yaml(recipe, gv, output_subdir=output_subdir)
@@ -626,7 +615,7 @@ def filter_baseline_noop_variants(
             return list(grid), []
         for idx, gv in enumerate(grid):
             args = str(getattr(gv, "extra_server_args", "") or "")
-            if _is_noop_filter_exempt(gv) or _subtracts_from_the_launch(gv) or _touches_a_multi_value_flag(args):
+            if _is_config_replay_variant(gv) or _subtracts_from_the_launch(gv) or _touches_a_multi_value_flag(args):
                 kept.append(gv)
                 continue
             try:
@@ -1161,7 +1150,7 @@ def _drop_leading_noops(self) -> None:
         # They change nothing while the stack stays as observed, and can only launch on it, so they go before a
         # serving lease is taken; any other is judged at its turn, against the stack a KEEP has moved.
         before = len(self.runnable)
-        if self.runnable:
+        if self.runnable and not is_stack_revalidation(self.params):
             kept, dropped = filter_baseline_noop_variants(
                 self.runnable,
                 framework=self.framework,
@@ -1237,8 +1226,11 @@ def _restates_stack(self, gv: GridVariant) -> str:
         """Why ``gv`` only restates the launch the stack runs as it stands now, or ``""`` when it changes it.
 
         The observed launch is what the server ran when the round opened; a KEEP since then has made the observed
-        value of whatever it changed stale, and ``observed_launch_after`` leaves that out.
+        value of whatever it changed stale, and ``observed_launch_after`` leaves that out. A stack revalidation
+        re-measures the stack on purpose, so it never restates it.
         """
+        if is_stack_revalidation(self.params):
+            return ""
         _, restated = filter_baseline_noop_variants(
             [gv],
             framework=self.framework,

From 4f44696b3099bad4a24eb29b11ded7e1d5f85801 Mon Sep 17 00:00:00 2001
From: lishuoshuo-amd <shuoli@amd.com>
Date: Tue, 27 Oct 2026 14:44:51 +0800
Subject: [PATCH 2/6] fix(explore): judge a variant by the whole launch it
 would start

The baseline-noop probe materialized the baseline and each variant, then
compared only the settings a variant names. It could not see a removal,
a list-valued flag or an empty delta, so it exempted the first two and
fell back to matching the request text for the third. Under AgentX the
probe read benchmark.envs, while the launch lives in
agentx.launch_overrides, so every variant looked like the baseline and
only the text fallback kept restatements from being dropped; none were.

read_launch reads the materialized config back whole, in either layout:
settings as the engine resolves them (last spelling wins, --no-x sets x
to false), envs, and the controls neither shows (process-level unsets,
AgentX removals, replace mode, executable, source files). launch_change
diffs two such launches and discounts settings and envs the observed
server already resolved. A variant with no change is a baseline noop,
and a variant starting a server this round already launched is a
round_dup, whatever its text. The exemptions and the text fallback go.
---
 scripts/code_metrics_baseline.txt             |   2 +-
 .../tests/test_explore_executor.py            | 122 +++++++++++--
 .../actions/executors/_launch_restatement.py  | 160 ++++++++++++++--
 .../orchestrator/actions/executors/explore.py | 172 +++++-------------
 4 files changed, 290 insertions(+), 166 deletions(-)

diff --git a/scripts/code_metrics_baseline.txt b/scripts/code_metrics_baseline.txt
index 37bcd994aa..4154adbbf6 100644
--- a/scripts/code_metrics_baseline.txt
+++ b/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 1750
 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
diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
index b19be0408b..c0b60ecabc 100644
--- a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
+++ b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
@@ -43,7 +43,10 @@
     observed_launch_from_state,
 )
 from hyperloom.orchestrator.actions.executors._launch_restatement import (
+    Launch,
     config_launch_restated_by,
+    launch_change,
+    named_settings,
     restates_the_baseline,
 )
 from hyperloom.orchestrator.state.shared_state import SharedState
@@ -197,10 +200,7 @@ def _noop_filter_kwargs(tmp_path: Path, **overrides: object) -> dict:
         yaml.safe_dump({"benchmark": {"framework": "sglang", "envs": {**_RECIPE_ENV, **_BASE_ARGS_ENV}}}),
         encoding="utf-8",
     )
-    kwargs = {
-        "framework": "sglang",
-        "base_yaml_path": base_yaml,
-    }
+    kwargs = {"base_yaml_path": base_yaml}
     kwargs.update(overrides)
     kwargs["recipe"] = GridRecipe(
         base_yaml_path=kwargs.pop("base_yaml_path"),
@@ -287,19 +287,61 @@ def test_variant_that_is_not_known_to_be_the_stack_is_kept(tmp_path, variant, ov
     assert dropped == []
 
 
-def test_a_removal_is_never_a_restatement_of_the_stack(tmp_path):
-    """chaojhou 3: the materialized config cannot show a subtraction, so it must not read as a noop.
-
-    ``unset_envs`` also pops the process env at run time, which the probe's
-    YAML never sees; ``remove_args`` leaves no trace in the merged argv.
-    """
+def test_a_removal_changes_the_launch_only_when_the_stack_passes_what_it_removes(tmp_path):
+    """The materialized argv shows a removal; an unset also pops the process env, which no config shows."""
     grid = [
-        GridVariant("removes-an-arg", remove_args=["--enable-dp-attention"]),
+        GridVariant("removes-a-passed-arg", remove_args=["--mem-fraction-static"]),
+        GridVariant("removes-an-absent-arg", remove_args=["--enable-dp-attention"]),
         GridVariant("unsets-an-env", unset_envs=["HIP_FORCE_DEV_KERNARG"]),
     ]
     kept, dropped = filter_baseline_noop_variants(grid, **_noop_filter_kwargs(tmp_path))
-    assert [gv.name for gv in kept] == ["removes-an-arg", "unsets-an-env"]
-    assert dropped == []
+    assert [gv.name for gv in kept] == ["removes-a-passed-arg", "unsets-an-env"]
+    assert dropped == [("removes-an-absent-arg", "launches the server the stack already runs (baseline noop)")]
+
+
+def _agentx_noop_filter_kwargs(tmp_path: Path) -> dict:
+    workload = tmp_path / "agentx.yaml"
+    workload.write_text(
+        yaml.safe_dump(
+            {
+                "benchmark": {
+                    "framework": "sglang",
+                    "model": "Qwen/Qwen3-0.6B",
+                    "agentx": {"enabled": True, "mode": "canonical"},
+                    "envs": {"TP": "1", "EP_SIZE": "1", "CONC": "8"},
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+    kwargs = _noop_filter_kwargs(
+        tmp_path, observed_server_config={"kv_cache_dtype": "fp8", "max_running_requests": 512}
+    )
+    kwargs["recipe"] = GridRecipe(
+        base_yaml_path=workload,
+        base=LaunchLayer(extra_server_args="--kv-cache-dtype fp8", extra_envs={"SGLANG_TEST": "1"}),
+        target=BenchmarkTarget(model_path=None, gpu_type=None, benchmark_script=None),
+    )
+    return kwargs
+
+
+def test_an_agentx_variant_is_judged_on_its_launch_overrides(tmp_path):
+    """AgentX carries the server argv in ``launch_overrides``, not in ``benchmark.envs``."""
+    grid = [
+        GridVariant("restates-the-stack-arg", "--kv-cache-dtype fp8"),
+        GridVariant("restates-the-engine-default", "--max-running-requests 512"),
+        GridVariant("restates-the-stack-env", extra_envs={"SGLANG_TEST": "1"}),
+        GridVariant("changes-an-arg", "--chunked-prefill-size 2048"),
+        GridVariant("changes-an-env", extra_envs={"SGLANG_TEST": "0"}),
+        GridVariant("removes-a-recipe-arg", remove_args=["--disable-radix-cache"]),
+    ]
+    kept, dropped = filter_baseline_noop_variants(grid, **_agentx_noop_filter_kwargs(tmp_path))
+    assert [gv.name for gv in kept] == ["changes-an-arg", "changes-an-env", "removes-a-recipe-arg"]
+    assert [name for name, _ in dropped] == [
+        "restates-the-stack-arg",
+        "restates-the-engine-default",
+        "restates-the-stack-env",
+    ]
 
 
 def test_the_engines_own_record_is_what_makes_the_filter_fire(tmp_path):
@@ -405,7 +447,6 @@ def _vllm_noop_filter_kwargs(tmp_path: Path) -> dict:
     )
     return _noop_filter_kwargs(
         tmp_path,
-        framework="vllm",
         base_yaml_path=base_yaml,
         observed_server_config={"kv_cache_dtype": "fp8", "enable_prefix_caching": True, "max_num_seqs": 64},
         observed_server_env={"VLLM_ROCM_USE_AITER": "1"},
@@ -493,6 +534,32 @@ def test_restates_the_baseline(argv, envs, base_argv, expected):
     assert restates_the_baseline(argv, envs, base_argv, {"VLLM_ROCM_USE_AITER": "1"}, "vllm") is expected
 
 
+def test_named_settings_resolve_like_the_engine():
+    """The last spelling of a boolean flag wins, negated or not; an argv it cannot account for raises."""
+    assert named_settings("--enable-prefix-caching --no-enable-prefix-caching", "vllm") == {
+        "enable_prefix_caching": ["false"]
+    }
+    assert named_settings("--no-enable-prefix-caching --enable-prefix-caching", "vllm") == {"enable_prefix_caching": []}
+    for argv in ("fp8 --kv-cache-dtype fp8", "--kv-cache-dtype 'fp8"):
+        with pytest.raises(ValueError):
+            named_settings(argv, "vllm")
+
+
+def test_launch_change_names_every_change_and_discounts_what_the_server_resolved():
+    base = Launch({"enable_x": [], "kv_cache_dtype": ["fp8"]}, {"A": "1"}, {"unset_envs": []})
+    launch = Launch({"enable_x": ["false"], "block_size": ["128"]}, {"B": "2"}, {"unset_envs": ["C"]})
+    assert launch_change(base, launch, ({}, {})) == {
+        "settings": {"enable_x": ["false"], "block_size": ["128"]},
+        "removed": ["kv_cache_dtype"],
+        "envs": {"B": "2"},
+        "unset_envs": ["A"],
+        "controls": ["unset_envs"],
+    }
+    same_server = Launch({**base.settings, "block_size": ["128"], "enable_y": ["false"]}, {"A": "1", "B": "2"}, {})
+    observed = ({"block_size": 128, "no_enable_y": True}, {"B": "2"})
+    assert launch_change(Launch(base.settings, base.envs, {}), same_server, observed) == {}
+
+
 def test_record_explore_accepted_dedup_by_fingerprint():
     state = SharedState()
     variant = {
@@ -743,6 +810,26 @@ async def test_a_stack_revalidation_runs_although_it_restates_the_stack(sub_agen
     assert out["skipped_dup"] == []
 
 
+@pytest.mark.asyncio
+async def test_rows_that_launch_the_same_server_are_benched_once(sub_agent_runner, tmp_path, monkeypatch):
+    """Restating the stack's fp8 beside the same change launches the server the first row already measured."""
+    grid = [
+        {"name": "v_seqs", "extra_args": "--max-num-seqs 256", "extra_envs": {}},
+        {"name": "v_fp8_seqs", "extra_args": "--kv-cache-dtype fp8 --max-num-seqs 256", "extra_envs": {}},
+    ]
+    out, launched = await _explore_on_an_observed_fp8_stack(
+        sub_agent_runner,
+        tmp_path,
+        monkeypatch,
+        grid,
+        {"v_seqs": 800.4, "v_fp8_seqs": 800.4},
+        inherited="--kv-cache-dtype fp8",
+    )
+
+    assert launched == ["v_seqs"]
+    assert [(sd["name"], sd["reason"]) for sd in out["skipped_dup"]] == [("v_fp8_seqs", "round_dup")]
+
+
 @pytest.mark.asyncio
 async def test_a_variant_that_was_the_stack_runs_once_a_keep_has_moved_the_stack(
     sub_agent_runner, tmp_path, monkeypatch
@@ -878,7 +965,10 @@ async def test_a_restatement_left_leading_by_the_skip_list_is_skipped_at_its_tur
 async def test_a_restatement_stays_skipped_after_a_keep_that_switches_the_stack_to_replace_mode(
     sub_agent_runner, tmp_path, monkeypatch
 ):
-    """A KEEP that removes a flag persists the stack's launched args in replace mode, which moves only that flag."""
+    """A KEEP that removes a flag persists the stack's launched args in replace mode, which moves only that flag.
+
+    ``fp8`` then launches the server ``v_drop`` just measured, so it is that round's duplicate.
+    """
     grid = [
         {
             "name": "v_drop",
@@ -900,7 +990,7 @@ async def test_a_restatement_stays_skipped_after_a_keep_that_switches_the_stack_
     assert launched == ["v_drop"]
     (keep,) = out["winners"]
     assert keep["args_mode"] == "replace"
-    assert [(sd["name"], sd["reason"]) for sd in out["skipped_dup"]] == [("v_fp8", "baseline_noop")]
+    assert [(sd["name"], sd["reason"]) for sd in out["skipped_dup"]] == [("v_fp8", "round_dup")]
 
 
 @pytest.mark.asyncio
diff --git a/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py b/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
index 6bf9793c37..4e8b7587ff 100644
--- a/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
+++ b/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
@@ -1,24 +1,27 @@
 # 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 json
 import logging
 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
 
@@ -44,20 +47,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 +91,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,6 +120,106 @@ def _opposite_setting(name: str) -> str:
     return name[3:] if name.startswith("no_") else f"no_{name}"
 
 
+@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]
+
+    @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.
+    """
+    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 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.
+
+    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.
+    """
+    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
+    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 text == str(observed)
+
+
 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:
diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py
index 842e1c0ab0..2943461808 100644
--- a/src/hyperloom/orchestrator/actions/executors/explore.py
+++ b/src/hyperloom/orchestrator/actions/executors/explore.py
@@ -95,14 +95,13 @@
     session_grid_bounds,
 )
 from hyperloom.inference_optimizer.grid_server_args import (
-    _MULTI_VALUE_FLAGS,
     compose_server_args,
     server_args_env_name,
 )
 from ._ray_serving import maybe_serving_lease
 from ._spec_decode_context_copy import CONTEXT_COPY_REFUSAL, context_copy_refusal
 
-from ._launch_restatement import config_envs, named_settings, observed_launch_after, restates_the_baseline
+from ._launch_restatement import Launch, config_envs, launch_change, observed_launch_after, read_launch
 from ._server_argv import ConfigUnreadable, config_server_argv
 from ._server_lifecycle import (
     resolve_lifecycle_params,
@@ -445,80 +444,6 @@ def observed_launch_from_state(state: Any) -> tuple[dict[str, Any], dict[str, st
     return config, env
 
 
-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
-    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 text == str(observed)
-
-
-def _launch_delta(
-    variant_argv: str,
-    variant_envs: Mapping[str, str],
-    base_argv: str,
-    base_envs: Mapping[str, str],
-    framework: str,
-) -> tuple[dict[str, list[str]], dict[str, str]]:
-    """What a variant changes about the baseline launch, in argv settings and envs alike.
-
-    Only the delta is comparable to the running server. The rest is the
-    baseline's own request, which materialization, the recipe and the engine all
-    rewrite on the way to a launch -- a requested ``--watchdog-timeout 1800``
-    resolves to ``3600.0`` -- so weighing it would reject every variant. Those
-    rewrites hit base and variant alike, so the delta cancels them.
-    """
-    base_settings = named_settings(base_argv, framework)
-    settings = {
-        name: values
-        for name, values in named_settings(variant_argv, framework).items()
-        if base_settings.get(name) != values
-    }
-    envs = {name: value for name, value in variant_envs.items() if base_envs.get(name) != value}
-    return settings, envs
-
-
-def _settings_already_hold(settings: Mapping[str, list[str]], observed_config: Mapping[str, Any]) -> bool:
-    """Whether every named setting already holds that value in the running server.
-
-    Checked setting by setting rather than as one argv string: the observed side
-    is the engine's whole resolved config, which no config's argv can equal.
-    """
-    return all(
-        name in observed_config and _setting_already_holds(values, observed_config[name])
-        for name, values in settings.items()
-    )
-
-
-def _envs_already_hold(envs: Mapping[str, str], observed_env: Mapping[str, str]) -> bool:
-    """Whether every named env already carries that value in the running server."""
-    return all(observed_env.get(name) == value for name, value in envs.items())
-
-
-def _subtracts_from_the_launch(variant: GridVariant) -> bool:
-    """Whether ``variant`` removes something, which the probe's config cannot show."""
-    return bool(getattr(variant, "remove_args", None) or getattr(variant, "unset_envs", None))
-
-
-def _touches_a_multi_value_flag(argv_text: str) -> bool:
-    """Whether ``argv_text`` names a list-valued flag, whose ordering the merged config does not preserve."""
-    return any(flag in argv_text for flag in _MULTI_VALUE_FLAGS)
-
-
 #: Names the throwaway variant that materializes the unchanged baseline to diff against.
 _NOOP_PROBE_BASE_NAME = "__baseline_noop_probe_base__"
 
@@ -529,6 +454,16 @@ def _materialized_launch(recipe: GridRecipe, gv: GridVariant, output_subdir: Pat
     return config_server_argv(out_path).text, config_envs(out_path)
 
 
+#: What reading a probe launch can raise; a launch that cannot be read is never judged unchanged.
+_UNREADABLE_LAUNCH = (OSError, ValueError, yaml.YAMLError, ConfigUnreadable)
+
+
+def _probe_launch(recipe: GridRecipe, gv: GridVariant, output_subdir: Path) -> Launch:
+    """The whole launch ``run_grid`` would start *gv* with on ``recipe``'s base."""
+    out_path = _build_variant_yaml(recipe, gv, output_subdir=output_subdir)
+    return read_launch(out_path, [*to_str_list(recipe.base.unset_envs), *to_str_list(gv.unset_envs)])
+
+
 def filter_context_copy_drafts(
     grid: list[GridVariant], *, recipe: GridRecipe
 ) -> tuple[list[GridVariant], list[tuple[str, str]]]:
@@ -570,74 +505,45 @@ def filter_context_copy_drafts(
 def filter_baseline_noop_variants(
     grid: list[GridVariant],
     *,
-    framework: str,
     observed: tuple[Mapping[str, Any], Mapping[str, str]],
     recipe: GridRecipe,
 ) -> tuple[list[GridVariant], list[tuple[str, str]]]:
     """Drop variants that would launch the server the stack is already running.
 
-    The baseline and each variant are materialized through ``_build_variant_yaml``
-    into throwaway configs, and which rule applies depends on whether the computed
-    settings and env delta of the variant against the baseline (``_launch_delta``)
-    is empty. That delta lists only what the variant sets differently, so a
-    ``"replace"`` variant that drops a baseline flag can still have an empty one:
-
-    1. Empty delta: the variant is dropped only when its ``args_mode`` is
-       ``"append"`` and its own request names only settings and envs the baseline
-       launch already holds (``restates_the_baseline``). This rule reads the
-       baseline config, not the observed launch. Every other empty-delta variant
-       -- a ``"replace"`` request, an empty request, or a change the merged config
-       hides -- is kept without consulting the observed launch.
-    2. Non-empty delta: only what the variant *changes* about the baseline is
-       judged, those settings against ``observed_server_config`` -- the engine's
-       own resolved settings -- and those envs against ``observed_server_env``.
-       The variant is dropped when both halves are already in effect.
-
-    A variant that removes something, or names a list-valued flag, is exempt:
-    neither is visible here, so both would read as a restatement of the stack.
+    The baseline and each variant are materialized through ``_build_variant_yaml`` and read back whole, so a
+    removal, a replaced argv, an overlay or an AgentX launch override shows as the change it is; ``launch_change``
+    then discounts what the observed server already resolved. A config replay always runs, and so does a variant
+    whose launch cannot be read.
 
     Returns the variants still to run, plus ``(name, reason)`` for each drop.
-    Filters nothing when the stack's launch was never observed.
     """
-    observed_config, observed_env = observed
-    if not observed_config or not grid:
-        return list(grid), []
+    if not grid:
+        return [], []
     kept: list[GridVariant] = []
     dropped: list[tuple[str, str]] = []
     with tempfile.TemporaryDirectory(prefix="explore_noop_probe_") as tmp_dir:
         tmp_root = Path(tmp_dir)
         try:
-            base_argv, base_envs = _materialized_launch(recipe, GridVariant(_NOOP_PROBE_BASE_NAME), tmp_root / "base")
-        except (OSError, ValueError, yaml.YAMLError) as exc:
+            base = _probe_launch(recipe, GridVariant(_NOOP_PROBE_BASE_NAME), tmp_root / "base")
+        except _UNREADABLE_LAUNCH as exc:
             log.warning(
-                "explore: baseline-noop probe could not materialize the baseline (%s); keeping every variant", exc
+                "explore: baseline-noop probe could not read the baseline launch (%s); keeping every variant", exc
             )
             return list(grid), []
         for idx, gv in enumerate(grid):
-            args = str(getattr(gv, "extra_server_args", "") or "")
-            if _is_config_replay_variant(gv) or _subtracts_from_the_launch(gv) or _touches_a_multi_value_flag(args):
+            if _is_config_replay_variant(gv):
                 kept.append(gv)
                 continue
             try:
-                variant_argv, variant_envs = _materialized_launch(recipe, gv, tmp_root / f"v{idx}")
-            except (OSError, ValueError, yaml.YAMLError) as exc:
-                log.warning("explore: baseline-noop probe failed for variant %s (%s); keeping it", gv.name, exc)
+                launch = _probe_launch(recipe, gv, tmp_root / f"v{idx}")
+            except _UNREADABLE_LAUNCH as exc:
+                log.warning("explore: baseline-noop probe could not read variant %s (%s); keeping it", gv.name, exc)
                 kept.append(gv)
                 continue
-            delta_settings, delta_envs = _launch_delta(variant_argv, variant_envs, base_argv, base_envs, framework)
-            if not delta_settings and not delta_envs:
-                # A change the merged config hides leaves no delta either: only a request
-                # that itself restates the baseline marks the variant as a noop.
-                restated = restates_the_baseline(gv.extra_server_args, gv.extra_envs, base_argv, base_envs, framework)
-                if restated and gv.args_mode == "append":
-                    dropped.append((gv.name, "names only settings the baseline launch already holds (baseline noop)"))
-                    continue
+            if launch_change(base, launch, observed):
                 kept.append(gv)
-                continue
-            if _settings_already_hold(delta_settings, observed_config) and _envs_already_hold(delta_envs, observed_env):
-                dropped.append((gv.name, "merged launch already active in the observed server (baseline noop)"))
-                continue
-            kept.append(gv)
+            else:
+                dropped.append((gv.name, "launches the server the stack already runs (baseline noop)"))
     return kept, dropped
 
 
@@ -968,9 +874,9 @@ def __init__(self, ctx: Any, params: dict[str, Any], output_root: Path) -> None:
         self.winners_history: list[dict[str, Any]] = []
         self.tested: dict[str, dict[str, Any]] = {}
         self.name_index: dict[str, Any] = {}
-        # Fingerprints this round launched, and the variants that reached the measured check and were answered by
-        # the stack or the ledger instead (a restatement of the stack, or a verdict the session already holds).
-        self.launched_fps: set[str] = set()
+        # Servers this round launched, and the variants that reached the measured check and were answered by the
+        # stack or the ledger instead (a restatement of the stack, or a verdict the session already holds).
+        self.launched: set[str] = set()
         self.judged_without_measuring = 0
         # Set when the loop stops because the run stopped it -- the budget ran out, or the orchestrator cancelled
         # the action -- so the round can say so instead of reporting a bare, unattributed failure: a variant that
@@ -1153,7 +1059,6 @@ def _drop_leading_noops(self) -> None:
         if self.runnable and not is_stack_revalidation(self.params):
             kept, dropped = filter_baseline_noop_variants(
                 self.runnable,
-                framework=self.framework,
                 observed=self.observed,
                 recipe=GridRecipe(
                     base_yaml_path=self.config_path,
@@ -1209,9 +1114,8 @@ def _open_schedule(self) -> None:
     # ----- one variant -----------------------------------------------------
 
     def _refused(self, gv: GridVariant, idx: int) -> bool:
-        fingerprint = _variant_fingerprint(gv)
-        if fingerprint in self.launched_fps:
-            self.skipped.append({"name": gv.name, "fingerprint": fingerprint, "reason": "round_dup"})
+        if self._launch_identity(gv) in self.launched:
+            self.skipped.append({"name": gv.name, "fingerprint": _variant_fingerprint(gv), "reason": "round_dup"})
             return True
         if restated := self._restates_stack(gv):
             _skip_baseline_noop(gv.name, restated, self.skipped)
@@ -1222,6 +1126,15 @@ def _refused(self, gv: GridVariant, idx: int) -> bool:
             return True
         return self._out_of_budget(idx) or self._refuses_context_copy(gv)
 
+    def _launch_identity(self, gv: GridVariant) -> str:
+        """The server ``gv`` would launch on the stack as it stands now; its fingerprint when that cannot be read."""
+        with tempfile.TemporaryDirectory(prefix="explore_launch_") as tmp_dir:
+            try:
+                return _probe_launch(self._recipe(), gv, Path(tmp_dir)).identity
+            except _UNREADABLE_LAUNCH as exc:
+                log.warning("explore: could not read the launch of %s (%s); deduping on its fingerprint", gv.name, exc)
+                return _variant_fingerprint(gv)
+
     def _restates_stack(self, gv: GridVariant) -> str:
         """Why ``gv`` only restates the launch the stack runs as it stands now, or ``""`` when it changes it.
 
@@ -1233,7 +1146,6 @@ def _restates_stack(self, gv: GridVariant) -> str:
             return ""
         _, restated = filter_baseline_noop_variants(
             [gv],
-            framework=self.framework,
             observed=observed_launch_after(*self.observed, self.opening_stack, self.launched_stack(), self.framework),
             recipe=self._recipe(),
         )
@@ -1340,7 +1252,7 @@ def _stopped_by_the_run(self, result: Any, gv: GridVariant, idx: int, round_labe
         return True
 
     async def _measure(self, gv: GridVariant, run_gv: GridVariant, idx: int, slot: Path) -> Any:
-        self.launched_fps.add(_variant_fingerprint(gv))
+        self.launched.add(self._launch_identity(gv))
         # The warmup and decision rounds share this slot as the lifecycle pid_dir so the decision round
         # re-attaches to the server the warmup left hot. Ray-managed GPU execution (12 T1): both reuse the
         # round-level lease, and one persistent server, so no GPU process outlives it.

From 27fdcbce79a57a86ca327c5db1f83651585b11cc Mon Sep 17 00:00:00 2001
From: lishuoshuo-amd <shuoli@amd.com>
Date: Tue, 27 Oct 2026 14:50:32 +0800
Subject: [PATCH 3/6] fix(warm-replay): judge a recipe by the launch it would
 start on the baseline config

The prelude matched the recipe's request text against the baseline
config's benchmark.envs argv. Under AgentX the baseline's server args live
in agentx.launch_overrides, so no recipe was ever found to restate the
baseline and a restating replay re-measured it for a full round.

The replay now asks the explore judgement whether the recipe, laid onto
the baseline's materialized config as a variant, starts the baseline's
own server. The observed engine settings discount a setting only when
current_best was observed running that very config. Patches, kernel work
and reference launch controls still always replay; the text matchers
restates_the_baseline and config_launch_restated_by go.
---
 scripts/code_metrics_baseline.txt             |  2 +-
 .../tests/test_explore_executor.py            | 57 ++++++++++++-------
 .../tests/test_warm_replay.py                 | 29 ++++++++++
 .../actions/executors/_launch_restatement.py  | 57 +------------------
 .../orchestrator/actions/executors/explore.py | 20 +++++++
 src/hyperloom/orchestrator/phases/prelude.py  | 10 ++--
 6 files changed, 94 insertions(+), 81 deletions(-)

diff --git a/scripts/code_metrics_baseline.txt b/scripts/code_metrics_baseline.txt
index 4154adbbf6..63bdfdee89 100644
--- a/scripts/code_metrics_baseline.txt
+++ b/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 1750
+module-lines src/hyperloom/orchestrator/actions/executors/explore.py 1769
 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
diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
index c0b60ecabc..85cffb3996 100644
--- a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
+++ b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
@@ -41,14 +41,9 @@
     _default_grid_for_framework,
     filter_baseline_noop_variants,
     observed_launch_from_state,
+    restates_the_baseline_launch,
 )
-from hyperloom.orchestrator.actions.executors._launch_restatement import (
-    Launch,
-    config_launch_restated_by,
-    launch_change,
-    named_settings,
-    restates_the_baseline,
-)
+from hyperloom.orchestrator.actions.executors._launch_restatement import Launch, launch_change, named_settings
 from hyperloom.orchestrator.state.shared_state import SharedState
 from hyperloom.common.env import is_truthy
 from hyperloom.common.perf_metric import perf_snapshot_from_mapping
@@ -493,16 +488,29 @@ def test_restated_args_with_a_kernel_overlay_are_not_a_restatement(tmp_path):
     assert dropped == []
 
 
-def test_a_null_baseline_env_is_unset_not_the_string_none(tmp_path):
-    """The launch leaves a null env unset, so a request for the string ``None`` sets something new."""
+def _vllm_baseline_state(tmp_path: Path, base_argv: str = _VLLM_BASE_ARGV, **envs: object) -> SimpleNamespace:
     config = tmp_path / "baseline.yaml"
     config.write_text(
-        yaml.safe_dump({"benchmark": {"framework": "vllm", "envs": {"FOO": None, "EXTRA_VLLM_ARGS": _VLLM_BASE_ARGV}}}),
+        yaml.safe_dump(
+            {
+                "benchmark": {
+                    "framework": "vllm",
+                    "envs": {"VLLM_ROCM_USE_AITER": "1", **envs, "EXTRA_VLLM_ARGS": base_argv},
+                }
+            }
+        ),
         encoding="utf-8",
     )
-    assert config_launch_restated_by(str(config), "", {"FOO": "None"}, "vllm") is False
-    assert config_launch_restated_by(str(config), "--kv-cache-dtype fp8", {}, "vllm") is True
-    assert config_launch_restated_by(str(tmp_path / "missing.yaml"), "--kv-cache-dtype fp8", {}, "vllm") is False
+    return SimpleNamespace(baseline_config_path=str(config), current_best_measurement={})
+
+
+def test_a_null_baseline_env_is_unset_not_the_string_none(tmp_path):
+    """The launch leaves a null env unset, so a request for the string ``None`` sets something new."""
+    state = _vllm_baseline_state(tmp_path, FOO=None)
+    assert restates_the_baseline_launch(state, "", {"FOO": "None"}) is False
+    assert restates_the_baseline_launch(state, "--kv-cache-dtype fp8", {}) is True
+    missing = SimpleNamespace(baseline_config_path=str(tmp_path / "missing.yaml"), current_best_measurement={})
+    assert restates_the_baseline_launch(missing, "--kv-cache-dtype fp8", {}) is False
 
 
 @pytest.mark.parametrize(
@@ -511,27 +519,38 @@ def test_a_null_baseline_env_is_unset_not_the_string_none(tmp_path):
         pytest.param("--kv-cache-dtype fp8", {}, _VLLM_BASE_ARGV, True, id="restated_flag"),
         pytest.param("--enable-prefix-caching", {}, _VLLM_BASE_ARGV, True, id="restated_bare_flag"),
         pytest.param("", {"VLLM_ROCM_USE_AITER": "1"}, _VLLM_BASE_ARGV, True, id="restated_env"),
-        pytest.param("", {}, _VLLM_BASE_ARGV, False, id="empty_request"),
+        pytest.param("", {}, _VLLM_BASE_ARGV, True, id="empty_request"),
         pytest.param("--kv-cache-dtype auto", {}, _VLLM_BASE_ARGV, False, id="different_value"),
         pytest.param("--block-size 128", {}, _VLLM_BASE_ARGV, False, id="setting_the_baseline_lacks"),
         pytest.param("", {"VLLM_ROCM_USE_AITER": "0"}, _VLLM_BASE_ARGV, False, id="different_env"),
         pytest.param("", {"NEW_ENV": "1"}, _VLLM_BASE_ARGV, False, id="env_the_baseline_lacks"),
         pytest.param("fp8 --kv-cache-dtype fp8", {}, _VLLM_BASE_ARGV, False, id="positional_token"),
         pytest.param("--kv-cache-dtype 'fp8", {}, _VLLM_BASE_ARGV, False, id="unparsable"),
+        pytest.param("--no-enable-prefix-caching", {}, _VLLM_BASE_ARGV, False, id="negates_a_baseline_flag"),
         pytest.param(
             "--enable-prefix-caching",
             {},
             "--enable-prefix-caching --no-enable-prefix-caching",
-            False,
-            id="negated_spelling_in_the_baseline",
+            True,
+            id="dedupe_keeps_the_negation_last",
         ),
         pytest.param("--cuda-graph-bs 1 2 4", {}, "--cuda-graph-bs 1 2 4", True, id="list_restated_in_full"),
         pytest.param("--cuda-graph-bs 1 2", {}, "--cuda-graph-bs 1 2 4", False, id="list_restated_in_part"),
     ],
 )
-def test_restates_the_baseline(argv, envs, base_argv, expected):
-    """Only a request every part of which the baseline already sets is a restatement; anything unreadable is not."""
-    assert restates_the_baseline(argv, envs, base_argv, {"VLLM_ROCM_USE_AITER": "1"}, "vllm") is expected
+def test_restates_the_baseline_launch(tmp_path, argv, envs, base_argv, expected):
+    """Only a request that leaves the materialized launch as it was restates it; anything unreadable does not."""
+    assert restates_the_baseline_launch(_vllm_baseline_state(tmp_path, base_argv), argv, envs) is expected
+
+
+def test_the_baseline_observation_counts_only_for_the_config_it_was_made_on(tmp_path):
+    """``--max-num-seqs 256`` is the observed engine default, but only on the launch that was observed."""
+    state = _vllm_baseline_state(tmp_path)
+    observed = {"launch_evidence": {"observed_server_config": {"max_num_seqs": 64, "block_size": 16}}}
+    state.current_best_measurement = observed
+    assert restates_the_baseline_launch(state, "--block-size 16", {}) is False
+    observed["launch_evidence"]["materialized_config_path"] = state.baseline_config_path
+    assert restates_the_baseline_launch(state, "--block-size 16", {}) is True
 
 
 def test_named_settings_resolve_like_the_engine():
diff --git a/src/hyperloom/inference_optimizer/tests/test_warm_replay.py b/src/hyperloom/inference_optimizer/tests/test_warm_replay.py
index 1e5632eadc..7fa46dfac6 100644
--- a/src/hyperloom/inference_optimizer/tests/test_warm_replay.py
+++ b/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."""
diff --git a/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py b/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
index 4e8b7587ff..bc16854143 100644
--- a/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
+++ b/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
@@ -11,7 +11,6 @@
 from __future__ import annotations
 
 import json
-import logging
 import shlex
 from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass
@@ -25,9 +24,7 @@
 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]:
@@ -218,55 +215,3 @@ def setting_already_holds(proposed: list[str], observed: Any) -> bool:
     if observed is None:
         return False
     return text == str(observed)
-
-
-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("-")
-
-
-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.
-    """
-    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())
-
-
-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.
-
-    False when the config is absent or unreadable: the launch to compare
-    against is then unknown.
-    """
-    if not str(config_path or "").strip():
-        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)
-        return False
-    return restates_the_baseline(argv_text, envs, base_argv, base_envs, framework)
diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py
index 2943461808..8c264d30ba 100644
--- a/src/hyperloom/orchestrator/actions/executors/explore.py
+++ b/src/hyperloom/orchestrator/actions/executors/explore.py
@@ -547,6 +547,26 @@ def filter_baseline_noop_variants(
     return kept, dropped
 
 
+def restates_the_baseline_launch(state: Any, args: str, envs: Mapping[str, Any]) -> bool:
+    """Whether ``args`` and ``envs`` laid onto the baseline's materialized config start the baseline's own server.
+
+    Judged as a variant on that config is. The observed launch discounts settings only when ``current_best`` was
+    observed running that very config.
+    """
+    config_path = str(getattr(state, "baseline_config_path", "") or "")
+    if not config_path:
+        return False
+    measurement = getattr(state, "current_best_measurement", None)
+    evidence = measurement.get("launch_evidence") if isinstance(measurement, Mapping) else None
+    observed_here = isinstance(evidence, Mapping) and evidence.get("materialized_config_path") == config_path
+    _, dropped = filter_baseline_noop_variants(
+        [GridVariant("warm_replay", str(args or ""), {str(k): str(v) for k, v in envs.items()})],
+        observed=observed_launch_from_state(state) if observed_here else ({}, {}),
+        recipe=GridRecipe(base_yaml_path=Path(config_path)),
+    )
+    return bool(dropped)
+
+
 def _skip_baseline_noop(name: str, reason: str, skipped: list[dict[str, Any]]) -> None:
     """Record a variant skipped because it launches the server the stack is already running."""
     log.info("explore: skipping baseline-noop variant %s (%s)", name, reason)
diff --git a/src/hyperloom/orchestrator/phases/prelude.py b/src/hyperloom/orchestrator/phases/prelude.py
index f678dd3ae0..84455c606f 100644
--- a/src/hyperloom/orchestrator/phases/prelude.py
+++ b/src/hyperloom/orchestrator/phases/prelude.py
@@ -47,8 +47,8 @@
     JournalEntry,
     Verdict,
 )
-from ..actions.executors._launch_restatement import config_launch_restated_by
 from ..actions.executors.baseline import revert_warm_kernel_patches
+from ..actions.executors.explore import restates_the_baseline_launch
 from ..state.analysis_failures import RECIPE_PARAM, repeated_roofline_failure, roofline_recipe
 from ..state.shared_state import inject_stack_base_params, resolve_graded_comparison
 from ..state.task_registry import Task, TaskSpec
@@ -1823,15 +1823,15 @@ def _skip_restated_replay(
     ) -> bool:
         """Skip a replay whose launch would be the baseline's own; True when it did.
 
-        Patches and kernel items change the code, which no argv comparison sees,
+        Patches and kernel items change the code, which no launch comparison sees,
         so a replay carrying any of them always runs. So does one under reference
-        launch controls: materialization re-applies them to the baseline config,
-        so the replay's launch is not that config's argv plus the request.
+        launch controls: materialization rebuilds the args from the reference, so
+        the replay's launch is not the baseline config plus the request.
         """
         state = self.shared_state
         if code_changes or state.reference_launch_controls:
             return False
-        if not config_launch_restated_by(str(state.baseline_config_path or ""), args, envs, str(state.framework or "")):
+        if not restates_the_baseline_launch(state, args, envs):
             return False
         log.info("PRELUDE: warm replay skipped: %r %s only restate the baseline launch", args, sorted(envs))
         self._skip_warm_replay(

From 609df0e56c993dfd9776742ec347ecb89742dafb Mon Sep 17 00:00:00 2001
From: lishuoshuo-amd <shuoli@amd.com>
Date: Tue, 27 Oct 2026 14:59:15 +0800
Subject: [PATCH 4/6] fix(explore): record a row answered without measuring as
 decided in the ledger

A row skipped as baseline_noop or round_dup went only to skipped_dup, so
explore_search.tested never learned of it. untested_proposal_rows reads
"not in tested" as untested, so the auto-bench offered the same head rows
again under the same idempotency key, got the finished task back, and
the real proposals behind them were never benched in that cycle.

Such a row is now written to tested as SKIPPED_DEDUP with its reason
and, for a round_dup, the fingerprint of the row that launched the same
server. A change the ledger already holds keeps its row, since a
measurement outranks a skip, and the cycle-focus negative ledger does not
count a row that rejected nothing.
---
 scripts/code_metrics_baseline.txt             |  2 +-
 .../tests/test_explore_executor.py            | 67 ++++++++++++++++++-
 .../tests/test_longrun_phase2.py              | 12 ++++
 .../orchestrator/actions/executors/explore.py | 57 +++++++++++-----
 .../orchestrator/phases/macro_cycle.py        |  4 +-
 5 files changed, 124 insertions(+), 18 deletions(-)

diff --git a/scripts/code_metrics_baseline.txt b/scripts/code_metrics_baseline.txt
index 63bdfdee89..6eca3763a1 100644
--- a/scripts/code_metrics_baseline.txt
+++ b/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 1769
+module-lines src/hyperloom/orchestrator/actions/executors/explore.py 1796
 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
diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
index 85cffb3996..edf1aedc3f 100644
--- a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
+++ b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
@@ -849,6 +849,44 @@ async def test_rows_that_launch_the_same_server_are_benched_once(sub_agent_runne
     assert [(sd["name"], sd["reason"]) for sd in out["skipped_dup"]] == [("v_fp8_seqs", "round_dup")]
 
 
+@pytest.mark.asyncio
+async def test_rows_answered_without_measuring_leave_the_proposal_queue(sub_agent_runner, tmp_path, monkeypatch):
+    """Restatements and duplicates are decided, so the auto-bench queue moves on to the proposal behind them."""
+    proposals = [
+        {"name": "v_noop_a", "extra_args": "--max-running-requests 512"},
+        {"name": "v_noop_b", "extra_args": "--kv-cache-dtype fp8"},
+        {"name": "v_seqs", "extra_args": "--max-num-seqs 256"},
+        {"name": "v_fp8_seqs", "extra_args": "--kv-cache-dtype fp8 --max-num-seqs 256"},
+        {"name": "v_lpm", "extra_args": "--schedule-policy lpm"},
+    ]
+    out, launched = await _explore_on_an_observed_fp8_stack(
+        sub_agent_runner,
+        tmp_path,
+        monkeypatch,
+        [{**row, "extra_envs": {}} for row in proposals[:4]],
+        {"v_seqs": 800.4, "v_fp8_seqs": 800.4},
+        inherited="--kv-cache-dtype fp8",
+    )
+    assert launched == ["v_seqs"]
+    tested = out["explore_search_update"]["tested"]
+    seqs_fp = next(fp for fp, row in tested.items() if row["name"] == "v_seqs")
+    skipped = {
+        row["name"]: (row["outcome"], row["reason"], row.get("duplicate_of", ""))
+        for row in tested.values()
+        if row["outcome"] == "SKIPPED_DEDUP"
+    }
+    assert skipped == {
+        "v_noop_a": ("SKIPPED_DEDUP", "baseline_noop", ""),
+        "v_noop_b": ("SKIPPED_DEDUP", "baseline_noop", ""),
+        "v_fp8_seqs": ("SKIPPED_DEDUP", "round_dup", seqs_fp),
+    }
+
+    state = SharedState()
+    state.specialist_rounds = [{"task_id": "s1", "domain": "serving_specialist", "cycle": 0, "proposal_set": proposals}]
+    state.apply_explore_search_update(out["explore_search_update"])
+    assert [row["name"] for row in state.untested_proposal_rows()] == ["v_lpm"]
+
+
 @pytest.mark.asyncio
 async def test_a_variant_that_was_the_stack_runs_once_a_keep_has_moved_the_stack(
     sub_agent_runner, tmp_path, monkeypatch
@@ -885,7 +923,8 @@ async def test_a_variant_that_is_the_stack_is_skipped_while_the_stack_has_not_mo
     assert launched == ["v_revert"]
     noop = {sd["name"] for sd in out["skipped_dup"] if sd.get("reason") == "baseline_noop"}
     assert noop == {"v_noop_first", "v_noop_after"}
-    assert {row["name"] for row in out["explore_search_update"]["tested"].values()} == {"v_revert"}
+    outcomes = {row["name"]: row["outcome"] for row in out["explore_search_update"]["tested"].values()}
+    assert outcomes == {"v_noop_first": "SKIPPED_DEDUP", "v_revert": "REVERT", "v_noop_after": "SKIPPED_DEDUP"}
 
 
 @pytest.mark.asyncio
@@ -1106,6 +1145,32 @@ async def test_a_round_of_measured_and_restated_variants_is_answered_not_failed(
     assert out["status"] == "succeeded"
 
 
+@pytest.mark.asyncio
+async def test_a_skip_never_replaces_a_measured_ledger_row(sub_agent_runner, tmp_path, monkeypatch):
+    """fp8 measured before the stack was observed running it; restating it later leaves the measurement in place."""
+    _force_cold_decision(monkeypatch)
+    sub, tr, _ = sub_agent_runner
+    state = SharedState()
+    state.baseline_tput = 800.0
+    sub.shared_state = state
+    observed: dict[str, object] = {}
+    monkeypatch.setattr(
+        "hyperloom.orchestrator.actions.executors.explore.observed_launch_from_state",
+        lambda _state: (dict(observed), {}),
+    )
+    grid = [{"name": "v_fp8", "extra_args": "--kv-cache-dtype fp8"}]
+    await _run_explore_round(sub, tr, tmp_path, state, key="round-1", grid=grid, base_args="", bench_calls=[])
+    measured = dict(state.explore_search["tested"])
+    assert [row["outcome"] for row in measured.values()] == ["REVERT"]
+
+    observed["kv_cache_dtype"] = "fp8"
+    out = await _run_explore_round(sub, tr, tmp_path, state, key="round-2", grid=grid, base_args="", bench_calls=[])
+
+    assert [(sd["name"], sd["reason"]) for sd in out["skipped_dup"]] == [("v_fp8", "baseline_noop")]
+    assert out["explore_search_update"]["tested"] == {}
+    assert state.explore_search["tested"] == measured
+
+
 @pytest.mark.asyncio
 async def test_restatements_left_leading_by_a_filter_are_dropped_before_the_serving_lease(
     sub_agent_runner, tmp_path, monkeypatch
diff --git a/src/hyperloom/inference_optimizer/tests/test_longrun_phase2.py b/src/hyperloom/inference_optimizer/tests/test_longrun_phase2.py
index 6ce582ffc4..8695d29fb5 100644
--- a/src/hyperloom/inference_optimizer/tests/test_longrun_phase2.py
+++ b/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
diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py
index 8c264d30ba..865cf42db3 100644
--- a/src/hyperloom/orchestrator/actions/executors/explore.py
+++ b/src/hyperloom/orchestrator/actions/executors/explore.py
@@ -125,6 +125,12 @@
 
 # ``skipped_dup`` reason for a variant whose launch the session already measured on the same stack.
 ALREADY_MEASURED: str = "already_measured"
+# ... for one that starts the stack's own server, or a server this round already launched.
+BASELINE_NOOP: str = "baseline_noop"
+ROUND_DUP: str = "round_dup"
+
+#: The provenance of a grid row that names none: the model proposed it directly.
+_DIRECT_PROVENANCE = "llm_direct"
 
 
 def is_stack_revalidation(params: dict | None) -> bool:
@@ -567,12 +573,6 @@ def restates_the_baseline_launch(state: Any, args: str, envs: Mapping[str, Any])
     return bool(dropped)
 
 
-def _skip_baseline_noop(name: str, reason: str, skipped: list[dict[str, Any]]) -> None:
-    """Record a variant skipped because it launches the server the stack is already running."""
-    log.info("explore: skipping baseline-noop variant %s (%s)", name, reason)
-    skipped.append({"name": name, "reason": "baseline_noop", "detail": reason})
-
-
 def filter_operator_pinned_envs(
     grid: list[GridVariant],
     baseline_envs: dict[str, Any] | None,
@@ -896,7 +896,7 @@ def __init__(self, ctx: Any, params: dict[str, Any], output_root: Path) -> None:
         self.name_index: dict[str, Any] = {}
         # Servers this round launched, and the variants that reached the measured check and were answered by the
         # stack or the ledger instead (a restatement of the stack, or a verdict the session already holds).
-        self.launched: set[str] = set()
+        self.launched: dict[str, str] = {}
         self.judged_without_measuring = 0
         # Set when the loop stops because the run stopped it -- the budget ran out, or the orchestrator cancelled
         # the action -- so the round can say so instead of reporting a bare, unattributed failure: a variant that
@@ -1094,8 +1094,8 @@ def _drop_leading_noops(self) -> None:
             )
             kept_ids = {id(gv) for gv in kept}
             leading = next((i for i, gv in enumerate(self.runnable) if id(gv) in kept_ids), before)
-            for name, reason in dropped[:leading]:
-                _skip_baseline_noop(name, reason, self.skipped)
+            for gv, (_, reason) in zip(self.runnable[:leading], dropped[:leading]):
+                self._decided_without_measuring(gv, BASELINE_NOOP, detail=reason)
             self.runnable = self.runnable[leading:]
         log.info(
             "explore grid: payload=%d → runnable=%d (leading baseline_noop=%d)",
@@ -1134,11 +1134,11 @@ def _open_schedule(self) -> None:
     # ----- one variant -----------------------------------------------------
 
     def _refused(self, gv: GridVariant, idx: int) -> bool:
-        if self._launch_identity(gv) in self.launched:
-            self.skipped.append({"name": gv.name, "fingerprint": _variant_fingerprint(gv), "reason": "round_dup"})
+        if twin := self.launched.get(self._launch_identity(gv)):
+            self._decided_without_measuring(gv, ROUND_DUP, duplicate_of=twin)
             return True
         if restated := self._restates_stack(gv):
-            _skip_baseline_noop(gv.name, restated, self.skipped)
+            self._decided_without_measuring(gv, BASELINE_NOOP, detail=restated)
             self.judged_without_measuring += 1
             return True
         if self._already_measured(gv):
@@ -1146,6 +1146,33 @@ def _refused(self, gv: GridVariant, idx: int) -> bool:
             return True
         return self._out_of_budget(idx) or self._refuses_context_copy(gv)
 
+    def _decided_without_measuring(
+        self, gv: GridVariant, reason: str, *, detail: str = "", duplicate_of: str = ""
+    ) -> None:
+        """Record ``gv`` as answered by the stack or by a twin this round launched, not measured.
+
+        Its ledger row marks the proposal decided, so the untested-proposal queue moves past it; a change the ledger
+        already holds keeps its row, since a measurement outranks a skip.
+        """
+        fingerprint = _variant_fingerprint(gv)
+        log.info("explore: skipping %s variant %s (%s)", reason, gv.name, detail or duplicate_of)
+        self.skipped.append(
+            {"name": gv.name, "fingerprint": fingerprint, "reason": reason, "detail": detail or duplicate_of}
+        )
+        known = (getattr(self.ss, "explore_search", None) or {}).get("tested") or {}
+        if not fingerprint or fingerprint in self.tested or fingerprint in known:
+            return
+        self.tested[fingerprint] = {
+            **self._variant_facts(gv),
+            "outcome": TS_SKIPPED_DEDUP,
+            "reason": reason,
+            "duplicate_of": duplicate_of,
+            "round_id": self.round_id,
+            "ts": _now_iso(),
+            "provenance": getattr(gv, "provenance", _DIRECT_PROVENANCE),
+            "workload_signature": self.ws_sig,
+        }
+
     def _launch_identity(self, gv: GridVariant) -> str:
         """The server ``gv`` would launch on the stack as it stands now; its fingerprint when that cannot be read."""
         with tempfile.TemporaryDirectory(prefix="explore_launch_") as tmp_dir:
@@ -1272,7 +1299,7 @@ def _stopped_by_the_run(self, result: Any, gv: GridVariant, idx: int, round_labe
         return True
 
     async def _measure(self, gv: GridVariant, run_gv: GridVariant, idx: int, slot: Path) -> Any:
-        self.launched.add(self._launch_identity(gv))
+        self.launched[self._launch_identity(gv)] = _variant_fingerprint(gv)
         # The warmup and decision rounds share this slot as the lifecycle pid_dir so the decision round
         # re-attaches to the server the warmup left hot. Ray-managed GPU execution (12 T1): both reuse the
         # round-level lease, and one persistent server, so no GPU process outlives it.
@@ -1344,7 +1371,7 @@ def _warmup_failed(self, gv: GridVariant, warmup: Any) -> None:
                 "base_tput": self.running_tput,
                 "round_id": self.round_id,
                 "ts": _now_iso(),
-                "provenance": getattr(gv, "provenance", "llm_direct"),
+                "provenance": getattr(gv, "provenance", _DIRECT_PROVENANCE),
                 "workload_signature": self.ws_sig,
                 "framework": self.framework,
                 "reason": "warmup_failed",
@@ -1391,7 +1418,7 @@ def _settle(self, gv: GridVariant, run_gv: GridVariant, r: Any, slot: Path) -> N
             "base_tput": self.running_tput,
             "round_id": self.round_id,
             "ts": _now_iso(),
-            "provenance": getattr(gv, "provenance", "llm_direct"),
+            "provenance": getattr(gv, "provenance", _DIRECT_PROVENANCE),
             "scope": str(getattr(gv, "scope", "") or ""),
             "workload_signature": self.ws_sig,
             "framework": self.framework,
diff --git a/src/hyperloom/orchestrator/phases/macro_cycle.py b/src/hyperloom/orchestrator/phases/macro_cycle.py
index bf12a253b2..5e7749b5f2 100644
--- a/src/hyperloom/orchestrator/phases/macro_cycle.py
+++ b/src/hyperloom/orchestrator/phases/macro_cycle.py
@@ -11,6 +11,7 @@
 
 from hyperloom.common.env import env_bool
 
+from ..actions.executors.explore import TS_SKIPPED_DEDUP
 from ..collaborator import CoordinatorCollaborator
 from . import machine_state as _phase_state
 from ..loop.maintenance import run_lease_and_db_reclaim
@@ -68,7 +69,8 @@ def _negative_ledger_domain_counts(self, *, recent_cycles: int = 3) -> dict[str,
                 rows.extend(rejected)
         counts: dict[str, int] = {}
         for row in rows:
-            if not isinstance(row, dict):
+            # A row the round answered without measuring rejects nothing.
+            if not isinstance(row, dict) or row.get("outcome") == TS_SKIPPED_DEDUP:
                 continue
             try:
                 cycle = int(row.get("cycle", cur_cycle) or 0)

From ff5b0def99afda31d79a126f641abb30f5d8f855 Mon Sep 17 00:00:00 2001
From: lishuoshuo-amd <shuoli@amd.com>
Date: Tue, 27 Oct 2026 15:01:40 +0800
Subject: [PATCH 5/6] feat(state): show orchestration the flags the running
 stack launched with

The tick summary carried current_best without its measurement, and the
orchestration sandbox cannot read the session config, so the model never
saw the operator's server flags and proposed flags the baseline already
passed as new levers. Render the deduplicated requested server args from
current_best's launch evidence, and state the explore contract as the
executor now applies it: a row starting the stack's own server is
skipped, and rows starting the same server are benched once.
---
 .../tests/test_shared_state_renderers.py         | 16 ++++++++++++++++
 .../orchestrator/prompts/prompt_builder.py       | 10 ++++------
 .../orchestrator/state/_shared_state/render.py   |  9 +++++++++
 3 files changed, 29 insertions(+), 6 deletions(-)

diff --git a/src/hyperloom/inference_optimizer/tests/test_shared_state_renderers.py b/src/hyperloom/inference_optimizer/tests/test_shared_state_renderers.py
index 5660f4507c..b97a944ffc 100644
--- a/src/hyperloom/inference_optimizer/tests/test_shared_state_renderers.py
+++ b/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)"
diff --git a/src/hyperloom/orchestrator/prompts/prompt_builder.py b/src/hyperloom/orchestrator/prompts/prompt_builder.py
index 073a020ed2..40222abfe5 100644
--- a/src/hyperloom/orchestrator/prompts/prompt_builder.py
+++ b/src/hyperloom/orchestrator/prompts/prompt_builder.py
@@ -533,10 +533,9 @@ def _format_grid_injection_hint(name: str) -> str | None:
             "base_extra_args?, base_tput?, accuracy_baseline?, "
             "keep_threshold_pct?: <raise only; floored at the session-cycle bar>}}`. "
             "Variants run serially; a KEEP is graded on its decision "
-            "round. Variant identity is content-based (args+envs+"
-            "remove_args+unset_envs+args_mode); only exact duplicates within "
-            "the same submitted grid are collapsed, so any prior fingerprint "
-            "may be re-proposed. "
+            "round. A row that starts the stack's own server (see "
+            "`current_launch_flags`) is skipped, and rows starting the same server "
+            "are benched once; any prior fingerprint may be re-proposed. "
             "Use remove_args/unset_envs to ablate harmful base flags; "
             "args_mode='replace' to drop inherited server args. "
             "Every variant requires action-time reasoning that names the "
@@ -801,8 +800,7 @@ def _idea_generation_lines() -> list[str]:
         "   be slowing the workload, emit a variant with `remove_args` and/or",
         "   `unset_envs` instead of only adding more knobs.",
         "",
-        "Variant identity is content-based (args+envs+remove_args+",
-        "unset_envs+args_mode); only exact same-grid duplicates are collapsed.",
+        "A row starting the stack's own server is skipped; rows starting the same server run once.",
         "`extra_server_args` is framework-neutral (routed to EXTRA_SGLANG_ARGS",
         "/ EXTRA_VLLM_ARGS / EXTRA_ATOM_ARGS by `--framework`).",
         "",
diff --git a/src/hyperloom/orchestrator/state/_shared_state/render.py b/src/hyperloom/orchestrator/state/_shared_state/render.py
index db3c739e7c..9982ea5714 100644
--- a/src/hyperloom/orchestrator/state/_shared_state/render.py
+++ b/src/hyperloom/orchestrator/state/_shared_state/render.py
@@ -12,6 +12,7 @@
 from hyperloom.common.env import env_bool
 from hyperloom.common.prompt_safety import flatten_for_prompt as _flatten_for_prompt
 from hyperloom.inference_optimizer.breakdown.recorder.kernel_records import GEAK_DROPPED_STATUSES
+from hyperloom.inference_optimizer.grid_server_args import dedupe_extra_server_args
 from . import gap_rows, proposal_scores, variant_line, warm_start
 from .attempt_audit import _AUDIT_ACTIONS
 from .phase_state import GAP_SEVERITY_RANK
@@ -481,6 +482,7 @@ def to_prompt_summary(self) -> str:
             f"baseline_tput={self.baseline_tput}  baseline_acc={self.baseline_accuracy}",
             f"baseline_failure_streak={self.baseline_failure_streak}",
             f"current_best={self._format_current_best()}",
+            f"current_launch_flags(already passed by the running stack)={self._format_current_launch_flags()}",
             f"optimization_stack={self._format_optimization_stack()}",
             (
                 f"cumulative_gain_validated={self.cumulative_gain_validated}% "
@@ -669,6 +671,13 @@ def _format_current_best(self) -> str:
             return "(none)"
         return str({k: v for k, v in self.current_best.items() if k not in ("optimization_stack", "measurement")})
 
+    def _format_current_launch_flags(self) -> str:
+        """The server flags ``current_best`` launched with, as its measurement recorded them."""
+        measurement = getattr(self, "current_best_measurement", None)
+        evidence = measurement.get("launch_evidence") if isinstance(measurement, dict) else None
+        flags = str(evidence.get("requested_server_args") or "") if isinstance(evidence, dict) else ""
+        return _flatten_for_prompt(dedupe_extra_server_args(flags.strip())) or "(unknown)"
+
     def _format_optimization_stack(self) -> str:
         """Render the optimization stack as ``action:variant`` parts."""
         if not self.optimization_stack:

From ae41de8ecd4c9cb52c29e6ab0c7ea56de8cbe0ae Mon Sep 17 00:00:00 2001
From: lishuoshuo-amd <shuoli@amd.com>
Date: Tue, 27 Oct 2026 15:11:52 +0800
Subject: [PATCH 6/6] fix(explore): judge a geak_revalidate row like any other
 row

The baseline-noop filter kept every row whose provenance was
geak_revalidate. The only producer of such rows enqueues them in a
stack_revalidate round, which never reaches the filter since the round
is exempt by its purpose, so the exemption was dead. It also let any
delegated row that claimed that provenance skip the judgement and
measure the stack again.
---
 scripts/code_metrics_baseline.txt                        | 2 +-
 .../inference_optimizer/tests/test_explore_executor.py   | 9 +++++++++
 src/hyperloom/orchestrator/actions/executors/explore.py  | 7 ++-----
 3 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/scripts/code_metrics_baseline.txt b/scripts/code_metrics_baseline.txt
index 6eca3763a1..035f6daf64 100644
--- a/scripts/code_metrics_baseline.txt
+++ b/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 1796
+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
diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
index edf1aedc3f..df1446e52d 100644
--- a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
+++ b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
@@ -294,6 +294,15 @@ def test_a_removal_changes_the_launch_only_when_the_stack_passes_what_it_removes
     assert dropped == [("removes-an-absent-arg", "launches the server the stack already runs (baseline noop)")]
 
 
+def test_a_row_claiming_a_revalidation_provenance_is_still_judged(tmp_path):
+    """Only a stack revalidation round is exempt; a grid row's own provenance cannot buy it a measurement."""
+    variant = _variant(_RESTATES_ARGV)
+    variant.provenance = "geak_revalidate"  # type: ignore[attr-defined]
+    kept, dropped = filter_baseline_noop_variants([variant], **_noop_filter_kwargs(tmp_path))
+    assert kept == []
+    assert dropped == [(_CANDIDATE, "launches the server the stack already runs (baseline noop)")]
+
+
 def _agentx_noop_filter_kwargs(tmp_path: Path) -> dict:
     workload = tmp_path / "agentx.yaml"
     workload.write_text(
diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py
index 865cf42db3..ac34937a60 100644
--- a/src/hyperloom/orchestrator/actions/executors/explore.py
+++ b/src/hyperloom/orchestrator/actions/executors/explore.py
@@ -518,8 +518,8 @@ def filter_baseline_noop_variants(
 
     The baseline and each variant are materialized through ``_build_variant_yaml`` and read back whole, so a
     removal, a replaced argv, an overlay or an AgentX launch override shows as the change it is; ``launch_change``
-    then discounts what the observed server already resolved. A config replay always runs, and so does a variant
-    whose launch cannot be read.
+    then discounts what the observed server already resolved. A variant whose launch cannot be read always runs;
+    a stack revalidation round never reaches this filter.
 
     Returns the variants still to run, plus ``(name, reason)`` for each drop.
     """
@@ -537,9 +537,6 @@ def filter_baseline_noop_variants(
             )
             return list(grid), []
         for idx, gv in enumerate(grid):
-            if _is_config_replay_variant(gv):
-                kept.append(gv)
-                continue
             try:
                 launch = _probe_launch(recipe, gv, tmp_root / f"v{idx}")
             except _UNREADABLE_LAUNCH as exc: