diff --git a/scripts/code_metrics_baseline.txt b/scripts/code_metrics_baseline.txt
index 37bcd994aa..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 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
diff --git a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
index 40cf75e320..df1446e52d 100644
--- a/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
+++ b/src/hyperloom/inference_optimizer/tests/test_explore_executor.py
@@ -41,11 +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 (
-    config_launch_restated_by,
-    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
@@ -197,10 +195,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 +282,70 @@ 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 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(
+        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 +451,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"},
@@ -452,16 +497,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(
@@ -470,27 +528,64 @@ 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():
+    """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():
@@ -684,7 +779,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 +807,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 +826,76 @@ 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_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_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
@@ -759,7 +932,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
@@ -858,7 +1032,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",
@@ -880,7 +1057,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
@@ -977,6 +1154,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/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/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 6bf9793c37..bc16854143 100644
--- a/src/hyperloom/orchestrator/actions/executors/_launch_restatement.py
+++ b/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)
diff --git a/src/hyperloom/orchestrator/actions/executors/explore.py b/src/hyperloom/orchestrator/actions/executors/explore.py
index f83ffdbd91..ac34937a60 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,
@@ -126,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:
@@ -445,101 +450,26 @@ 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)
-
-
-# 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)
     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]]]:
@@ -581,81 +511,63 @@ 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 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.
-    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_noop_filter_exempt(gv) or _subtracts_from_the_launch(gv) or _touches_a_multi_value_flag(args):
-                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
 
 
-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 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 filter_operator_pinned_envs(
@@ -979,9 +891,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: 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
@@ -1161,10 +1073,9 @@ 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,
                 observed=self.observed,
                 recipe=GridRecipe(
                     base_yaml_path=self.config_path,
@@ -1180,8 +1091,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)",
@@ -1220,12 +1131,11 @@ 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 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):
@@ -1233,15 +1143,53 @@ 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:
+            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.
 
         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,
             observed=observed_launch_after(*self.observed, self.opening_stack, self.launched_stack(), self.framework),
             recipe=self._recipe(),
         )
@@ -1348,7 +1296,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[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.
@@ -1420,7 +1368,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",
@@ -1467,7 +1415,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)
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(
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: