[RSI] Refuse proposals whose AITER_CONFIG file is not on disk when they are admitted; keep run_grid's check as backstop

scripts/code_metrics_baseline.txt

@@ -171,7 +171,7 @@ cognitive-complexity src/hyperloom/orchestrator/specialists/dispatch.py::Special
 cognitive-complexity src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.maybe_force_stalled_domain_specialist 39
 cognitive-complexity src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.warm_specialist_params 102
 cognitive-complexity src/hyperloom/orchestrator/specialists/patch_safety.py::resolve_patch_apply_root 40
-cognitive-complexity src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._finalize 49
+cognitive-complexity src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._finalize 41
 cognitive-complexity src/hyperloom/orchestrator/specialists/subprocess_.py::SpecialistSubprocessDispatcher._reap_loop 47
 cognitive-complexity src/hyperloom/orchestrator/specialists/subprocess_.py::SpecialistSubprocessDispatcher.run 97
 cognitive-complexity src/hyperloom/orchestrator/specialists/subprocess_.py::_codex_mcp_config 99
@@ -316,7 +316,7 @@ cyclomatic-complexity src/hyperloom/orchestrator/phases/prelude.py::PreludePhase
 cyclomatic-complexity src/hyperloom/orchestrator/phases/prelude.py::PreludePhase.maybe_enqueue_warm_replay 48
 cyclomatic-complexity src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.warm_specialist_params 41
 cyclomatic-complexity src/hyperloom/orchestrator/specialists/patch_safety.py::resolve_patch_apply_root 21
-cyclomatic-complexity src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._finalize 25
+cyclomatic-complexity src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._finalize 22
 cyclomatic-complexity src/hyperloom/orchestrator/specialists/subprocess_.py::SpecialistSubprocessDispatcher.run 39
 cyclomatic-complexity src/hyperloom/orchestrator/specialists/subprocess_.py::_codex_mcp_config 28
 cyclomatic-complexity src/hyperloom/orchestrator/trace_analysis/_bypass_trace_reader.py::_finalize 35
@@ -576,7 +576,7 @@ function-lines src/hyperloom/orchestrator/actions/executors/_geak_sweep.py::swee
 function-lines src/hyperloom/orchestrator/actions/executors/_gpu_pin.py::resolve_gpu_pin 105
 function-lines src/hyperloom/orchestrator/actions/executors/_grid_runner.py::_build_variant_yaml 135
 function-lines src/hyperloom/orchestrator/actions/executors/_grid_runner.py::_run_magpie 93
-function-lines src/hyperloom/orchestrator/actions/executors/_grid_runner.py::run_grid 1040
+function-lines src/hyperloom/orchestrator/actions/executors/_grid_runner.py::run_grid 1038
 function-lines src/hyperloom/orchestrator/actions/executors/_kernel_agent_tool.py::materialize_unified_patch_snapshot 94
 function-lines src/hyperloom/orchestrator/actions/executors/_kv_metrics.py::_read_aiperf_aggregate_json 101
 function-lines src/hyperloom/orchestrator/actions/executors/_kv_metrics.py::sample_from_families 100
@@ -764,12 +764,12 @@ function-lines src/hyperloom/orchestrator/roles/critic_agent.py::CriticAgentBack
 function-lines src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.build_specialist_round_entry 94
 function-lines src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.maybe_auto_retry_specialist 134
 function-lines src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.maybe_force_stalled_domain_specialist 114
-function-lines src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.record_specialist_result 130
+function-lines src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.record_specialist_result 128
 function-lines src/hyperloom/orchestrator/specialists/dispatch.py::SpecialistDispatchCollaborator.warm_specialist_params 179
 function-lines src/hyperloom/orchestrator/specialists/patch_safety.py::resolve_patch_apply_root 104
 function-lines src/hyperloom/orchestrator/specialists/patch_safety.py::vet_patches 107
 function-lines src/hyperloom/orchestrator/specialists/rebench.py::run_specialist_rebench 86
-function-lines src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._finalize 183
+function-lines src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._finalize 163
 function-lines src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._prepare 192
 function-lines src/hyperloom/orchestrator/specialists/runner.py::SpecialistRunner._run_via_subprocess 173
 function-lines src/hyperloom/orchestrator/specialists/subprocess_.py::SpecialistSubprocessDispatcher._build_codex_launch 97
@@ -953,7 +953,7 @@ module-lines src/hyperloom/orchestrator/phases/prelude.py 2640
 module-lines src/hyperloom/orchestrator/policy/gate.py 1588
 module-lines src/hyperloom/orchestrator/prompts/prompt_builder.py 1205
 module-lines src/hyperloom/orchestrator/prompts/specialist_prompt_builder.py 2571
-module-lines src/hyperloom/orchestrator/specialists/runner.py 1520
+module-lines src/hyperloom/orchestrator/specialists/runner.py 1505
 module-lines src/hyperloom/orchestrator/specialists/subprocess_.py 1747
 module-lines src/hyperloom/orchestrator/state/shared_state.py 2639
 module-lines src/hyperloom/orchestrator/trace_analysis/_bypass_report.py 1231

src/hyperloom/common/env_safety.py

@@ -12,6 +12,7 @@
 import os
 import re
 from collections.abc import Mapping
+from pathlib import Path
 
 log = logging.getLogger(__name__)
 
@@ -422,6 +423,46 @@ def filter_untrusted_env_mapping(
     return allowed, dropped
 
 
+# Env vars whose value names a config file the serving stack reads unconditionally: aiter's tuned-GEMM loader skips
+# its merge default and pd.read_csv's exactly the AITER_CONFIG_* path when one is set, so a missing file kills the
+# server at config load instead of falling back to the default table. Values that are not file paths (e.g. geak's
+# JSON-payload AITER_CONFIG) do not carry the trailing underscore and stay untouched.
+FILE_REFERENCING_ENV_PREFIXES: tuple[str, ...] = ("AITER_CONFIG_",)
+
+
+def _is_file(path: str) -> bool:
+    # Path.is_file() raises for an unreadable parent or an overlong name (before Python 3.14); such a path is not a
+    # file the server can read either, and one proposal must not crash what checks it.
+    try:
+        return Path(path).is_file()
+    except OSError:
+        return False
+
+
+def ungrounded_env_file_reason(envs: Mapping[str, object] | None) -> str | None:
+    """Return why ``envs`` cannot be served if one names a config file that does not exist, else ``None``.
+
+    Grounds file-like env values the way patch targets are grounded: a proposal may point a config env at an
+    artifact only a later (possibly rejected) authoring step would create, and the server would die at config load.
+    The check is existence-only; anything already on disk (a promoted, durable artifact) passes. A set value is a
+    ``:``-joined list of paths (see ``csvs_aiter_will_load``), so every non-empty element must exist, not only the
+    first.
+    """
+    for raw_key, raw_value in (envs or {}).items():
+        key = str(raw_key or "").strip().upper()
+        value = str(raw_value or "").strip()
+        if not key.startswith(FILE_REFERENCING_ENV_PREFIXES) or not value:
+            continue
+        missing = [p for p in (part.strip() for part in value.split(":")) if p and not _is_file(p)]
+        if missing:
+            return (
+                f"{key} points at {value}, whose element(s) {':'.join(missing)} do not exist; the server would read it "
+                "unconditionally and die at config load. Point it at a file a stage wrote (the gemm_tuning action "
+                "writes tuned-GEMM CSVs) or leave it unset to use the default config table."
+            )
+    return None
+
+
 OPERATOR_EXTRA_ENV_VAR = "INFERENCE_OPTIMIZER_EXTRA_ENV"
 
 
@@ -517,6 +558,7 @@ def redact_file_in_place(path: os.PathLike[str] | str, *, mode: int = 0o600) ->
     "BLOCKED_EXTERNAL_ENV_NAMES",
     "BLOCKED_UNTRUSTED_ENV_NAMES",
     "BLOCKED_VARIANT_ENV_NAMES",
+    "FILE_REFERENCING_ENV_PREFIXES",
     "GPU_MASK_ENV_NAMES",
     "OPERATOR_EXTRA_ENV_VAR",
     "SENSITIVE_ENV_NAME_MARKERS",
@@ -534,5 +576,6 @@ def redact_file_in_place(path: os.PathLike[str] | str, *, mode: int = 0o600) ->
     "redact_secret_values",
     "scrub_benchmark_process_env",
     "scrub_child_process_env",
+    "ungrounded_env_file_reason",
     "valid_env_key",
 ]

src/hyperloom/inference_optimizer/tests/test_coordinator_async_batch2_unit.py

@@ -2106,6 +2106,25 @@ async def test_record_specialist_result_logs_ungrounded_patches(coord: Coordinat
     assert "vllm/nope.py" in failures[0]["error_excerpt"]
 
 
+@pytest.mark.asyncio
+async def test_record_specialist_result_logs_proposals_whose_env_files_are_missing(coord: Coordinator) -> None:
+    """A proposal dropped at admission for an env file no stage wrote reaches the failure log the same way."""
+    task = _ptask("rec-spec-env", "specialist")
+    await coord.specialist_dispatch.record_specialist_result(
+        task=task,
+        done_payload={
+            "domain": "candidate_discovery_specialist",
+            "gap_canonical_id": "g1",
+            "proposal_set": [],
+            "proposals_env_ungrounded": ["proposal 'tuned-gemm': AITER_CONFIG_GEMM_BF16 points at /tmp/t.csv"],
+        },
+        source="specialist:rec-spec-env",
+    )
+    failures = [f for f in coord.shared_state.last_action_failures if f["task_id"] == "rec-spec-env"]
+    assert [f["error_class"] for f in failures] == ["proposal_env_file_missing"]
+    assert "AITER_CONFIG_GEMM_BF16" in failures[0]["error_excerpt"]
+
+
 @pytest.mark.asyncio
 async def test_record_specialist_result_no_dead_research_evidence_log(
     coord: Coordinator,

src/hyperloom/inference_optimizer/tests/test_grid_runner_helpers_coverage_unit.py

@@ -61,6 +61,74 @@ def test_compatibility_filter_no_model_path_assumes_compatible() -> None:
     assert [v.name for v in kept] == ["moe"] and dropped == []
 
 
+# -- ungrounded_env_file_reason (env-value file grounding) ----------------
+def test_ungrounded_env_file_reason_drops_missing_config_path(tmp_path) -> None:
+    # A proposal may point a config env at an artifact no stage authored; the server would read it
+    # unconditionally and die at config load, so the variant must be dropped with a reason.
+    v = _variant("gemm", envs={"AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE": str(tmp_path / "nope.csv")})
+    reason = gr.ungrounded_env_file_reason(v.extra_envs)
+    assert reason and "AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE" in reason
+    assert str(tmp_path / "nope.csv") in reason
+
+
+def test_ungrounded_env_file_reason_passes_authored_file(tmp_path) -> None:
+    csv = tmp_path / "tuned.csv"
+    csv.write_text("gfx,M,N,K,splitK\ngfx950,64,5120,5120,2\n", encoding="utf-8")
+    v = _variant("gemm", envs={"AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE": str(csv)})
+    assert gr.ungrounded_env_file_reason(v.extra_envs) is None
+
+
+def test_ungrounded_env_file_reason_accepts_colon_joined_existing_files(tmp_path) -> None:
+    # A set AITER_CONFIG_* is aiter's SET form: a `:`-joined list, each element loaded (csvs_aiter_will_load).
+    # Both files on disk is a valid variant and must not be dropped.
+    a = tmp_path / "t.csv"
+    a.write_text("gfx,M,N,K\ngfx950,1,2,3\n", encoding="utf-8")
+    b = tmp_path / "t2.csv"
+    b.write_text("gfx,M,N,K\ngfx950,4,5,6\n", encoding="utf-8")
+    v = _variant("gemm", envs={"AITER_CONFIG_GEMM_BF16": f"{a}:{b}"})
+    assert gr.ungrounded_env_file_reason(v.extra_envs) is None
+
+
+def test_ungrounded_env_file_reason_drops_colon_joined_missing_element(tmp_path) -> None:
+    a = tmp_path / "t.csv"
+    a.write_text("gfx,M,N,K\ngfx950,1,2,3\n", encoding="utf-8")
+    missing = str(tmp_path / "never_authored.csv")
+    v = _variant("gemm", envs={"AITER_CONFIG_GEMM_BF16": f"{a}:{missing}"})
+    reason = gr.ungrounded_env_file_reason(v.extra_envs)
+    assert reason and missing in reason and str(a) not in reason.split("element(s)")[1].split(" do not exist")[0]
+
+
+def test_ungrounded_env_file_reason_treats_an_unreadable_path_as_missing(monkeypatch) -> None:
+    # An unreadable parent or an overlong name makes Path.is_file() raise; the variant is dropped, the grid runs on.
+    def raise_eacces(self):
+        raise PermissionError(13, "Permission denied")
+
+    monkeypatch.setattr(gr.Path, "is_file", raise_eacces)
+    v = _variant("gemm", envs={"AITER_CONFIG_GEMM_BF16": "/root/private/tuned.csv"})
+    reason = gr.ungrounded_env_file_reason(v.extra_envs)
+    assert reason and "/root/private/tuned.csv" in reason
+
+
+def test_run_grid_fast_fail_drops_a_variant_whose_config_file_is_missing(tmp_path, monkeypatch) -> None:
+    # The backstop of the admission check: run_grid's fast-fail drops it before boot, whatever the framework.
+    monkeypatch.setenv("FRAMEWORK", "sglang")
+    v = _variant("gemm", envs={"AITER_CONFIG_GEMM_BF16": str(tmp_path / "never_written.csv")})
+    assert "AITER_CONFIG_GEMM_BF16" in (gr.unsupported_capability_reason(v) or "")
+
+
+def test_ungrounded_env_file_reason_noop_without_config_envs(tmp_path) -> None:
+    # Non-file envs (even under aiter-ish prefixes like the JSON-payload AITER_CONFIG) must stay untouched.
+    v = _variant(
+        "plain",
+        envs={
+            "AITER_CONFIG": '{"path": "/tmp/never_scanned.json"}',
+            "SGLANG_USE_AITER": "1",
+            "AITER_CONFIG_GEMM_A8W8": "",
+        },
+    )
+    assert gr.ungrounded_env_file_reason(v.extra_envs) is None
+
+
 # -- unsupported_capability_reason (env-flag build probe) -----------------
 def _clear_cap_cache() -> None:
     gr._CAP_PROBE_CACHE.clear()

src/hyperloom/inference_optimizer/tests/test_specialist_patch_safety_unit.py

@@ -199,6 +199,50 @@ def test_patch_safety_report_notes_empty():
     assert ps.PatchSafetyReport().notes() == []
 
 
+# ---- proposals whose env files are not on disk -------------------------------
+def test_a_proposal_pointing_an_env_at_an_unwritten_file_is_dropped_at_admission(tmp_path):
+    """Qwen3-14B-FP8 session 20261014T024938Z: two-step proposals ("author the CSV, then point the env at it") whose
+    first step nothing ran; benched as config arms, their servers died at config load."""
+    written = tmp_path / "tuned.csv"
+    written.write_text("gfx,M,N,K\n", encoding="utf-8")
+    missing = str(tmp_path / "aiter_configs" / "never_written.csv")
+    payload = {
+        "proposal_set": [
+            {"name": "tuned-gemm", "extra_envs": {"AITER_CONFIG_GEMM_A8W8_BLOCKSCALE_BPRESHUFFLE": missing}},
+            {"name": "written", "extra_envs": {"AITER_CONFIG_GEMM_BF16": str(written)}},
+            {"name": "flag-only", "extra_args": "--max-num-seqs 64", "extra_envs": {"VLLM_USE_V1": "1"}},
+        ]
+    }
+    dropped = ps.drop_env_ungrounded_proposals(payload)
+    assert [p["name"] for p in payload["proposal_set"]] == ["written", "flag-only"]
+    assert len(dropped) == 1 and dropped[0].startswith("proposal 'tuned-gemm': ")
+    assert missing in dropped[0] and "gemm_tuning" in dropped[0]
+    assert ps.drop_env_ungrounded_proposals({"proposal_set": "not a list"}) == []
+
+
+def test_admit_payload_books_the_gate_and_reports_the_env_drops(tmp_path):
+    missing = str(tmp_path / "never_written.csv")
+    payload = {
+        "summary": "s",
+        "expected_gain_pct": 12,
+        "proposal_set": [{"name": "tuned-gemm", "extra_envs": {"AITER_CONFIG_GEMM_BF16": missing}}],
+    }
+    vetted = (
+        ["a.patch"],
+        [{"path": "b.patch", "verdict": ps.GROUND_MISSING_TARGET, "detail": "missing_target: b"}],
+        {"a.patch": ps.GROUND_APPLIES},
+        False,
+    )
+    report = ps.admit_payload(payload, vetted, {"a.patch": "/src/vllm", "b.patch": "/src/vllm"})
+    assert payload["proposal_set"] == [] and "expected_gain_pct" not in payload
+    assert payload["patches_written"] == ["a.patch"] and payload["patch_roots"] == {"a.patch": "/src/vllm"}
+    assert payload["patches_ungrounded"] == ["missing_target: b"]
+    assert len(payload["proposals_env_ungrounded"]) == 1 and "patches_span_multiple_roots" not in payload
+    assert any("expected_gain_pct" in f for f in report.forbidden_fields)
+    assert report.env_ungrounded == payload["proposals_env_ungrounded"]
+    assert any(n.startswith("proposal_env_file_missing:") for n in report.notes())
+
+
 # ---- scan_numeric_claims ---------------------------------------------------
 def test_scan_numeric_claims():
     payload = {

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

@@ -27,6 +27,7 @@
     is_python_package_root,
     redact_secret_values,
     scrub_benchmark_process_env,
+    ungrounded_env_file_reason,
 )
 
 from ...phases import machine_state as _phase_state
@@ -236,11 +237,12 @@ def _probe_vllm_aiter_shared_expert_unsupported() -> str | None:
 
 
 def unsupported_capability_reason(variant: "GridVariant") -> str | None:
-    """Return a drop reason if a variant sets an env flag the installed framework build cannot honour, else ``None``."""
-    fw = (os.environ.get("FRAMEWORK", "") or "sglang").strip().lower()
-    if fw != "vllm":
-        return None
+    """Return why a variant cannot boot (a config file not on disk, an env flag the build lacks), else ``None``."""
     envs = {str(k): str(v) for k, v in (getattr(variant, "extra_envs", None) or {}).items()}
+    if missing_file := ungrounded_env_file_reason(envs):
+        return missing_file
+    if (os.environ.get("FRAMEWORK", "") or "sglang").strip().lower() != "vllm":
+        return None
     val = envs.get("VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS")
     if val is None or not is_truthy(val, default=True):
         return None
@@ -944,8 +946,6 @@ def _skip_rest_for_budget(idx: int, *, spent_on: str, rounds_left: int = variant
             await _unit_started(i, "variant")
             slot = output_root / f"variant_{i:02d}_{_safe(variant.name)}"
             server_log = slot / "server.log"
-            # Capability fast-fail: drop a variant whose env flag the build cannot honour before booting a doomed server,
-            # still recording the failure so the LLM learns not to re-pick it.
             cap_reason = unsupported_capability_reason(variant)
             if cap_reason:
                 log.warning(

src/hyperloom/orchestrator/specialists/dispatch.py

@@ -35,6 +35,12 @@
 __all__ = ["SpecialistDispatchCollaborator"]
 
 _SOURCE_PATCH_FAMILY = "source_patch"
+#: ``(specialist_done key, failure class)`` for what the patch-safety gate dropped at admission: each reaches the
+#: durable failure log, which every SEED turn renders.
+_ADMISSION_DROPS: tuple[tuple[str, str], ...] = (
+    ("patches_ungrounded", "patch_targets_ungrounded"),
+    ("proposals_env_ungrounded", "proposal_env_file_missing"),
+)
 
 #: ``(SharedState field, param key, conversion)`` for the values a specialist is warmed with. The converted value
 #: decides: an empty string, an empty container and a zero are not exported, because an absent key is how the
@@ -880,16 +886,14 @@ async def record_specialist_result(
         )
         # Specialist notes reach the prompt only through this task's one inbox
         # line; ``last_action_failures`` is rendered every SEED turn.
-        ungrounded = done_payload.get("patches_ungrounded")
-        if isinstance(ungrounded, list) and ungrounded:
-            self.shared_state.record_action_failure(
-                action="specialist",
-                task_id=task.task_id,
-                result={
-                    "error_class": "patch_targets_ungrounded",
-                    "error": "; ".join(str(d) for d in ungrounded[:4]),
-                },
-            )
+        for dropped_key, error_class in _ADMISSION_DROPS:
+            dropped = done_payload.get(dropped_key)
+            if isinstance(dropped, list) and dropped:
+                self.shared_state.record_action_failure(
+                    action="specialist",
+                    task_id=task.task_id,
+                    result={"error_class": error_class, "error": "; ".join(str(d) for d in dropped[:4])},
+                )
         # Advisory multi-model scoring of the proposal_set; informational only, gates nothing.
         if self._proposal_scorer is not None and proposals:
             scores = await self._proposal_scorer.score(