[FIX] make the grading anchor and the config it was measured on inseparable

src/hyperloom/inference_optimizer/tests/golden/coordinator_behavior_golden.json

@@ -117,12 +117,10 @@
         "workspace": null
       }
     ],
-    "baseline_cold_tput": 0.0,
     "baseline_config_path": "<volatile>",
     "baseline_double_run": true,
     "baseline_eager_fallback": false,
     "baseline_failure_streak": 2,
-    "baseline_hot_tput": 0.0,
     "baseline_roofline_ceiling": {},
     "baseline_runtime_sec": 0.0,
     "baseline_total_failures": 2,

src/hyperloom/inference_optimizer/tests/test_coordinator_async_methods_coverage_unit.py

@@ -63,8 +63,6 @@ async def test_promote_baseline_sets_anchor_and_current_best(coord: Coordinator)
         },
     )
     assert coord.shared_state.baseline_tput == 1000.0
-    assert coord.shared_state.baseline_cold_tput == 900.0
-    assert coord.shared_state.baseline_hot_tput == 1000.0
     assert coord.shared_state.baseline_failure_streak == 0
     assert coord.shared_state.baseline_arg_error_streak == 0
     assert coord.shared_state.current_best["action"] == "baseline"

src/hyperloom/inference_optimizer/tests/test_critic_verdict_map.py

@@ -688,6 +688,102 @@ async def create_or_return_existing(self, **kwargs: Any):  # noqa: ANN401
     assert create_calls[0]["params"]["grid"] == grid
 
 
+@pytest.mark.asyncio
+async def test_delegate_explore_seeds_the_stack_with_the_anchor(tmp_path: Path):
+    """A delegate explore carries current_best's args/envs, not just its throughput.
+
+    Seeding the anchor alone launched every variant on a bare config and graded it
+    against the established recipe, so a whole round read as a ~40% regression.
+    """
+    coord = _delegate_coord(tmp_path)
+    coord.shared_state.baseline_tput = 4663.79
+    coord.shared_state.current_best = {
+        "tput": 7725.6,
+        "extra_server_args": "--kv-cache-dtype fp8_e4m3 --max-num-seqs 64",
+        "extra_envs": {"VLLM_ROCM_USE_AITER": "1"},
+    }
+    created: list[dict[str, Any]] = []
+
+    class _TaskRegistry:
+        async def create_or_return_existing(self, **kwargs: Any):  # noqa: ANN401
+            created.append(dict(kwargs["params"]))
+            from hyperloom.orchestrator.state.task_registry import Task
+
+            return (
+                Task(
+                    task_id="t-explore-stack",
+                    kind=kwargs["kind"],
+                    state="queued",
+                    params=kwargs["params"],
+                    idempotency_key=kwargs["idempotency_key"],
+                ),
+                False,
+            )
+
+    coord.tasks = _TaskRegistry()
+    intent = Intent(
+        type=IntentType.DELEGATE,
+        payload={
+            "action_name": "explore",
+            "params": {"grid": [{"name": "v_a", "extra_args": "--flag-a"}]},
+            "idempotency_key": "explore-round-4",
+        },
+    )
+    await coord._handle_delegate("orchestration", intent)
+    assert len(created) == 1
+    params = created[0]
+    assert params["base_tput"] == 7725.6
+    assert params["base_extra_args"] == "--kv-cache-dtype fp8_e4m3 --max-num-seqs 64"
+    assert params["base_extra_envs"] == {"VLLM_ROCM_USE_AITER": "1"}
+
+
+@pytest.mark.asyncio
+async def test_delegate_sweep_seeds_the_stack_too(tmp_path: Path):
+    """A delegated sweep scans on top of current_best, as its action contract states.
+
+    ``executors/sweep.py`` assembles each (CONC, ISL, OSL) point from the
+    ``base_*`` params, so seeding only ``explore`` left a delegated sweep
+    measuring the bare baseline config.
+    """
+    coord = _delegate_coord(tmp_path)
+    coord.shared_state.baseline_tput = 4663.79
+    coord.shared_state.current_best = {
+        "tput": 7725.6,
+        "extra_server_args": "--max-num-seqs 64",
+        "extra_envs": {"VLLM_ROCM_USE_AITER": "1"},
+    }
+    created: list[dict[str, Any]] = []
+
+    class _TaskRegistry:
+        async def create_or_return_existing(self, **kwargs: Any):  # noqa: ANN401
+            created.append(dict(kwargs["params"]))
+            from hyperloom.orchestrator.state.task_registry import Task
+
+            return (
+                Task(
+                    task_id="t-sweep-stack",
+                    kind=kwargs["kind"],
+                    state="queued",
+                    params=kwargs["params"],
+                    idempotency_key=kwargs["idempotency_key"],
+                ),
+                False,
+            )
+
+    coord.tasks = _TaskRegistry()
+    intent = Intent(
+        type=IntentType.DELEGATE,
+        payload={"action_name": "sweep", "params": {}, "idempotency_key": "sweep-1"},
+    )
+    await coord._handle_delegate("orchestration", intent)
+    assert len(created) == 1
+    params = created[0]
+    assert params["base_extra_args"] == "--max-num-seqs 64"
+    assert params["base_extra_envs"] == {"VLLM_ROCM_USE_AITER": "1"}
+    # explore-only runtime knobs must not leak onto a sweep task.
+    assert "explore_overtime_kill_ratio" not in params
+
+
 # 6. Specialist prompt — proposal self-curation contract (Section 1 + 8)
 def _build_specialist_prompt_text() -> str:
     from hyperloom.orchestrator.specialists.domains import get_domain

src/hyperloom/inference_optimizer/tests/test_explore_executor.py

@@ -53,6 +53,19 @@ def _isolate_leak_root(tmp_path_factory, monkeypatch):
     monkeypatch.setenv("INFERENCE_OPTIMIZER_LEAK_ROOTS", str(sandbox))
 
 
+def _force_cold_decision(monkeypatch) -> None:
+    """Make server_lifecycle reuse ineligible, one of the two warm-decision preconditions."""
+    monkeypatch.setattr(
+        "hyperloom.orchestrator.actions.executors.explore.resolve_lifecycle_params",
+        lambda _config_path: {
+            "eligible": False,
+            "framework": "sglang",
+            "port": 30000,
+            "reason": "test: server_lifecycle reuse disabled",
+        },
+    )
+
+
 def _write_baseline_yaml(path: Path) -> None:
     cfg = {
         "benchmark": {
@@ -814,10 +827,129 @@ def _fake_run(cmd, *args, **kwargs):
     assert tested["outcome"] == expected_outcome
 
 
+@pytest.mark.asyncio
+async def test_explore_executor_takes_live_base_args_with_the_live_anchor(
+    sub_agent_runner,
+    tmp_path,
+):
+    """Superseding a stale ``base_tput`` also re-reads the args it was measured on."""
+    sub, tr, _ = sub_agent_runner
+    state = SharedState()
+    state.baseline_tput = 800.0
+    state.current_best = {
+        "action": "explore",
+        "tput": 1000.0,
+        "extra_server_args": "--live-layer 1",
+        "extra_envs": {"LIVE_ENV": "1"},
+    }
+    sub.shared_state = state
+
+    base = tmp_path / "base.yaml"
+    _write_baseline_yaml(base)
+
+    def _fake_run(cmd, *args, **kwargs):
+        out_idx = cmd.index("--output-dir")
+        _fake_workspace(Path(cmd[out_idx + 1]), tput=1100.0)  # +10% vs the live 1000
+        return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="ok", stderr="")
+
+    task = await tr.create(
+        kind="explore",
+        params={
+            "config_path": str(base),
+            "output_dir": str(tmp_path / "explore-live-base"),
+            # Snapshotted together at dispatch, before the newer layer landed.
+            "base_tput": 800.0,
+            "base_extra_args": "--stale-layer 1",
+            "enable_stack_rebench": False,
+            "grid": [
+                {
+                    "name": "on_live_stack",
+                    "extra_args": "--variant 2",
+                    "extra_envs": {},
+                    "provenance": "llm_direct",
+                }
+            ],
+            "variant_timeout_sec": 10,
+        },
+        idempotency_key="ex-live-base-args",
+    )
+    sub.register_executor("explore", ExploreExecutor(session_dir=tmp_path, enable_stack_rebench=False))
+    with patch(
+        "hyperloom.orchestrator.actions.executors._grid_runner.run_with_session_kill",
+        side_effect=_fake_run,
+    ):
+        res = await sub.run_task(task)
+
+    out = res.result
+    fp = canonical_fingerprint("--variant 2", {})
+    assert out["explore_search_update"]["tested"][fp]["base_tput"] == 1000.0
+    winner = out["winners"][0]
+    assert "--live-layer 1" in winner["extra_server_args"]
+    assert "--variant 2" in winner["extra_server_args"]
+    assert "--stale-layer" not in winner["extra_server_args"]
+    assert winner["extra_envs"]["LIVE_ENV"] == "1"
+
+
+@pytest.mark.asyncio
+async def test_explore_stack_rebench_floor_follows_the_in_batch_anchor(
+    sub_agent_runner,
+    tmp_path,
+):
+    """A 2nd KEEP that regresses against the 1st is evicted, not KEPT with a negative gain."""
+    sub, tr, _ = sub_agent_runner
+    base = tmp_path / "base.yaml"
+    _write_baseline_yaml(base)
+
+    def _fake_run(cmd, *args, **kwargs):
+        out_idx = cmd.index("--output-dir")
+        slot = Path(cmd[out_idx + 1])
+        # Match on path segments: ``tmp_path`` is named after the test, so a
+        # substring check would fire on every round.
+        parts = set(slot.parts)
+        if "v01_second" in parts:
+            # Round 1 clears the advanced bar (1260 vs 1200); the confirmation
+            # round drops below it while still beating the round-start 1000.
+            tput = 1100.0 if "stack_rebench" in parts else 1260.0
+        else:
+            tput = 1200.0
+        _fake_workspace(slot, tput=tput)
+        return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="ok", stderr="")
+
+    task = await tr.create(
+        kind="explore",
+        params={
+            "config_path": str(base),
+            "output_dir": str(tmp_path / "explore-floor"),
+            "base_tput": 1000.0,
+            "grid": [
+                {"name": "first", "extra_args": "--first", "extra_envs": {}, "provenance": "llm_direct"},
+                {"name": "second", "extra_args": "--second", "extra_envs": {}, "provenance": "llm_direct"},
+            ],
+            "variant_timeout_sec": 10,
+        },
+        idempotency_key="ex-rebench-floor",
+    )
+    sub.register_executor("explore", ExploreExecutor(session_dir=tmp_path))
+    with patch(
+        "hyperloom.orchestrator.actions.executors._grid_runner.run_with_session_kill",
+        side_effect=_fake_run,
+    ):
+        res = await sub.run_task(task)
+
+    out = res.result
+    tested = out["explore_search_update"]["tested"]
+    assert tested[canonical_fingerprint("--first", {})]["outcome"] == "KEEP"
+    assert tested[canonical_fingerprint("--second", {})]["outcome"] == "KEEP_UNSTABLE"
+    assert {w["name"] for w in out["winners"]} == {"first"}
+    assert all(w["gain_pct"] > 0 for w in out["winners"])
+    # The evicted variant must not drag the stack anchor down with it.
+    assert out["running_base_tput"] == 1200.0
+
+
 @pytest.mark.asyncio
 async def test_explore_executor_dedups_against_ledger(sub_agent_runner, tmp_path, monkeypatch):
     """A variant whose fingerprint already lives in explore_search.tested lands in ``skipped_dup``, not re-benched."""
-    monkeypatch.setenv("INFERENCE_OPTIMIZER_EXPLORE_WARM_DECISION", "0")
+    _force_cold_decision(monkeypatch)
     sub, tr, _ = sub_agent_runner
     base = tmp_path / "base.yaml"
     _write_baseline_yaml(base)
@@ -891,7 +1023,6 @@ async def test_explore_executor_defaults_to_warm_decision_matching_hot_baseline(
     monkeypatch,
 ):
     """Default EXPLORE measures hot decisions, matching default hot baseline."""
-    monkeypatch.delenv("INFERENCE_OPTIMIZER_EXPLORE_WARM_DECISION", raising=False)
     sub, tr, _ = sub_agent_runner
     base = tmp_path / "base.yaml"
     _write_baseline_yaml(base)
@@ -959,7 +1090,6 @@ async def test_explore_decision_round_skips_eval_warmup_keeps_it(
     """The overtime deadline is anchored on a throughput-only baseline, so the
     rounds it gates must measure throughput only. The warmup round is ungated and
     remains the accuracy source."""
-    monkeypatch.delenv("INFERENCE_OPTIMIZER_EXPLORE_WARM_DECISION", raising=False)
     sub, tr, _ = sub_agent_runner
     base = tmp_path / "base.yaml"
     _write_baseline_yaml(base)
@@ -1016,9 +1146,9 @@ async def test_explore_cold_decision_keeps_eval(
     tmp_path,
     monkeypatch,
 ):
-    """With warm-decision disabled there is no warmup eval to fall back on, so
-    the decision round must still run its own accuracy gate."""
-    monkeypatch.setenv("INFERENCE_OPTIMIZER_EXPLORE_WARM_DECISION", "0")
+    """Without server_lifecycle reuse there is no warmup round whose eval the
+    decision round could fall back on, so it must run its own accuracy gate."""
+    _force_cold_decision(monkeypatch)
     sub, tr, _ = sub_agent_runner
     base = tmp_path / "base.yaml"
     _write_baseline_yaml(base)
@@ -1065,14 +1195,62 @@ def _fake_run(cmd, *args, **kwargs):
     assert decision and all(ev not in _RUN_EVAL_FALSE for ev in decision)
 
 
+@pytest.mark.asyncio
+async def test_explore_decision_stays_cold_when_the_session_skips_the_double_run(
+    sub_agent_runner,
+    tmp_path,
+):
+    """A cold ``baseline_tput`` must be graded cold even when lifecycle reuse is available.
+
+    The baseline gates its cold+hot double run on ``baseline_double_run`` as well
+    as lifecycle eligibility, so warm-decision has to honour both or a hot
+    candidate is scored against a cold anchor.
+    """
+    sub, tr, _ = sub_agent_runner
+    state = SharedState()
+    state.baseline_tput = 800.0
+    state.baseline_double_run = False
+    sub.shared_state = state
+
+    base = tmp_path / "base.yaml"
+    _write_baseline_yaml(base)
+    seen: list[str] = []
+
+    def _fake_run(cmd, *args, **kwargs):
+        out_idx = cmd.index("--output-dir")
+        slot = Path(cmd[out_idx + 1])
+        seen.append(str(slot))
+        _fake_workspace(slot, tput=920.0)
+        return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="ok", stderr="")
+
+    task = await tr.create(
+        kind="explore",
+        params={
+            "config_path": str(base),
+            "output_dir": str(tmp_path / "explore-singleround"),
+            "base_tput": 800.0,
+            "grid": [{"name": "v", "extra_args": "--flag", "extra_envs": {}, "provenance": "llm_direct"}],
+            "variant_timeout_sec": 30,
+        },
+        idempotency_key="ex-no-double-run",
+    )
+    sub.register_executor("explore", ExploreExecutor(session_dir=tmp_path))
+    with patch(
+        "hyperloom.orchestrator.actions.executors._grid_runner.run_with_session_kill",
+        side_effect=_fake_run,
+    ):
+        await sub.run_task(task)
+
+    assert not [slot for slot in seen if "warmup_round" in slot]
+
+
 @pytest.mark.asyncio
 async def test_explore_executor_warm_decision_warmup_failure_marks_failed(
     sub_agent_runner,
     tmp_path,
     monkeypatch,
 ):
     """A failed warmup round records the variant FAILED(reason=warmup_failed), no decision run."""
-    monkeypatch.setenv("INFERENCE_OPTIMIZER_EXPLORE_WARM_DECISION", "1")
     sub, tr, _ = sub_agent_runner
     base = tmp_path / "base.yaml"
     _write_baseline_yaml(base)
@@ -1221,7 +1399,7 @@ async def test_explore_executor_killed_overtime_no_tput_no_keep(
     monkeypatch,
 ):
     """A fired soft deadline records KILLED_OVERTIME (no tput, no KEEP/REVERT, stack unchanged)."""
-    monkeypatch.setenv("INFERENCE_OPTIMIZER_EXPLORE_WARM_DECISION", "0")
+    _force_cold_decision(monkeypatch)
     sub, tr, _ = sub_agent_runner
     base = tmp_path / "base.yaml"
     _write_baseline_yaml(base)