diff --git a/packages/nemo_evaluator_sdk/examples/gym/README.md b/packages/nemo_evaluator_sdk/examples/gym/README.md
index ca04ba40b9..f140d755b1 100644
--- a/packages/nemo_evaluator_sdk/examples/gym/README.md
+++ b/packages/nemo_evaluator_sdk/examples/gym/README.md
@@ -39,7 +39,7 @@ Working from a Gym checkout also makes its components take precedence over the p
 
 Useful flags: `--resources-server`, `--agent`, `--model-type` (`inference_provider` for OpenAI-compatible **chat** endpoints; `openai_model` uses the OpenAI **Responses API** and 500s against chat-only endpoints), `--num-repeats`, `--dataset`, `--output-dir`.
 
-For the full set of knobs the underlying `gym env start` / `gym eval run` commands accept, see the [NeMo Gym documentation](https://github.com/NVIDIA-NeMo/Gym). Anything `GymRuntimeConfig` does not expose as a field can be passed through with its `env_overrides` escape hatch (Hydra `+key=value` overrides applied to `gym env start`).
+For the full set of knobs the underlying `gym env start` / `gym eval run` commands accept, see the [NeMo Gym documentation](https://github.com/NVIDIA-NeMo/Gym). Anything `GymRuntimeConfig` does not expose as a field can be passed through with its `env_overrides` escape hatch — nested data such as `{"model": {"temperature": 0.7}}`, flattened to Hydra's override grammar and applied to `gym env start`.
 
 Each run writes its bundle to a fresh temporary directory by default. Pass `--output-dir` to choose one, but give every run its own: the runner refuses to reuse a directory that already holds Gym rollout output (Gym appends to its failures sidecar, so reusing one would mix runs) and raises rather than clearing a prior run's results.
 
diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/gym_runtime.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/gym_runtime.py
index ce623ac894..66183fdd54 100644
--- a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/gym_runtime.py
+++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/gym_runtime.py
@@ -28,10 +28,14 @@
 means a caller can run a **subset** of tasks without Gym rolling out the rest.
 
 **Execution** is the two-step Gym flow (the one that reads a dataset directly
-without triggering Gym's split-driven data-prep): ``gym env start`` brings up the
-resources-server + agent + model servers, then ``gym eval run --no-serve --input
-<materialized dataset>`` collects rollouts against them. The runtime shells out
-to the ``gym`` CLI on PATH, so this SDK never imports ``nemo_gym``. Subprocess
+without triggering Gym's split-driven data-prep), preceded by a pre-flight:
+``gym env validate`` merges the composed config and reports unset ``???`` values,
+bad paths, and dangling cross-references without starting anything; then ``gym env
+start`` brings up the resources-server + agent + model servers, and ``gym eval run
+--no-serve --input <materialized dataset>`` collects rollouts against them. Both
+commands receive the identical selection arguments, so what is validated is what
+runs. The runtime shells out to the ``gym`` CLI on PATH, so this SDK never imports
+``nemo_gym``. Subprocess
 output is streamed to log files under the run's work dir *and* mirrored to this
 module's logger at ``DEBUG``, so callers choose terminal visibility through
 ordinary ``logging`` configuration.
@@ -95,6 +99,11 @@
 #: once, and that path is what the subprocesses run — so a child whose PATH differs from ours cannot
 #: end up executing a different Gym.
 _GYM_CLI = "gym"
+#: Bound on `gym env validate`. It merges config without starting anything and returns in about a
+#: second; this only exists so a wedged invocation cannot stall the run before it begins.
+_VALIDATE_TIMEOUT_S = 120.0
+#: Where Gym's Hydra run directories are redirected, relative to the run's work dir.
+_HYDRA_SUBDIR = "gym_hydra"
 #: Gym's index fields on each rollout record. ``_ng_task_index`` is the only join back to the input
 #: rows that survives a round-trip: Gym mutates ``responses_create_params`` (even the prompt) and
 #: copies only a fixed allowlist of row keys onto the result, so no field we invent comes back. Gym
@@ -107,34 +116,177 @@
 _LOG_TAIL_LINES = 40
 
 
-#: Substrings that mark a Hydra override key as carrying a credential. Matched case-insensitively
-#: against the key half of ``+key=value``.
+#: Substrings that mark an override as carrying a credential. Matched case-insensitively against the
+#: full dotted path, so nesting cannot hide one behind an innocuous leaf name.
 _SECRET_KEY_MARKERS = ("api_key", "apikey", "token", "secret", "password", "passwd", "credential")
 #: Stand-in written in place of a redacted override value.
 _REDACTED = "<redacted>"
 
 
-def _redact_env_overrides(overrides: Sequence[str]) -> list[str]:
-    """Redact credential-looking values from Hydra overrides before they are recorded as provenance.
+#: Dict keys Hydra reads back unchanged. Its ``dictKey`` rule accepts no quoting, so a key is
+#: whatever the lexer makes of the bare text — this is deliberately narrower than what parses.
+_HYDRA_DICT_KEY = re.compile(r"[A-Za-z_][A-Za-z0-9_.-]*\Z")
+#: Bare words the lexer types rather than reading as text, so they cannot serve as string keys.
+_HYDRA_KEY_LITERALS = frozenset({"true", "false", "null", "inf", "nan"})
 
-    ``env_overrides`` is a free-form escape hatch forwarded verbatim to ``gym env start``, so nothing
-    stops a caller passing ``+model.api_key=sk-...``. ``RunnerInfo.config`` is persisted into the run
-    bundle, so a value that looks like a credential must not be written there.
+
+def _hydra_dict(value: Mapping[str, Any]) -> str:
+    """Render a mapping as a Hydra dict container, ``{key:value,...}``.
+
+    Reached for a mapping nested inside a container — ``[{"b": 1}]`` — where there is no dotted path
+    to flatten onto, so the dict has to be spelled inline. Values recurse, so the typed spellings
+    below hold at any depth.
+
+    Keys are emitted bare, because Hydra's ``dictKey`` rule has no quoted form: ``{'b':1}`` does not
+    parse at all. That leaves the key at the mercy of the lexer, which types it — ``{true:1}`` keys
+    on the boolean ``True``, ``{1.5:1}`` on a float — and rejects ``:``, ``,``, brackets, and quotes
+    outright. Anything outside the conservative shape above therefore raises here rather than
+    silently keying the config on something the caller did not write.
+    """
+    rendered = []
+    for key, item in value.items():
+        if not isinstance(key, str) or not _HYDRA_DICT_KEY.match(key) or key.casefold() in _HYDRA_KEY_LITERALS:
+            raise ValueError(
+                f"Gym config override has dict key {key!r}, which Hydra's override grammar cannot "
+                "express as a string: keys are unquoted, so only a leading letter or underscore "
+                "followed by letters, digits, '_', '.', or '-' survives the round trip. Set this "
+                "key through the override path instead of nesting it inside a list."
+            )
+        rendered.append(f"{key}:{_hydra_scalar(item)}")
+    return "{" + ",".join(rendered) + "}"
+
+
+def _hydra_scalar(value: Any) -> str:
+    """Render a leaf value the way Hydra's override grammar reads it back.
+
+    Hydra's grammar is typed, so an unquoted string is not necessarily a string: ``true`` parses as a
+    boolean, ``null`` as ``None``, ``1.5`` as a float, ``a,b`` as a *sweep*, and ``A[B`` fails to
+    parse outright. Strings are therefore always single-quoted, which round-trips every one of those
+    (verified against ``hydra.core.override_parser``). Interpolations survive quoting — the override
+    sets the literal text and OmegaConf resolves it on read — so ``${policy_base_url}`` still works.
+
+    Only ``'`` is escaped. Hydra does **not** decode ``\\\\`` inside a quoted value: escaping
+    backslashes doubles them, so they are passed through raw.
+
+    ``None`` and booleans get their own spellings, since ``str()`` would emit ``"None"``/``"True"``
+    and Hydra reads those back as text. Containers recurse for the same reason: ``str()`` on a dict
+    emits Python's repr, whose quoted keys Hydra rejects outright.
+    """
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, Mapping):
+        return _hydra_dict(value)
+    if isinstance(value, (list, tuple)):
+        return "[" + ",".join(_hydra_scalar(item) for item in value) + "]"
+    if isinstance(value, str):
+        # A trailing backslash would escape the closing quote and leave the value unterminated, and
+        # there is no spelling that avoids it — better to say so than to emit something unparseable.
+        if value.endswith("\\"):
+            raise ValueError(
+                f"Gym config override value {value!r} ends with a backslash, which Hydra's override "
+                "grammar cannot express: it escapes the closing quote."
+            )
+        return "'" + value.replace("'", "\\'") + "'"
+    return str(value)
+
+
+def _flatten_overrides(overrides: Mapping[str, Any], _prefix: str = "") -> list[str]:
+    """Flatten a nested override mapping into Hydra ``++dotted.path=value`` arguments.
+
+    Callers describe overrides as structured data — ``{"a": {"b": 1}}`` — rather than as
+    pre-serialized Hydra strings, so the config survives being sent somewhere as JSON. Hydra itself
+    only speaks the flat form, so the translation happens here, at the point of invocation.
+
+    ``++`` rather than ``+``: it sets a key whether or not it already exists, which is what an
+    override means. A bare ``+`` fails on a key the merged config already defines.
+    """
+    arguments: list[str] = []
+    for key, value in overrides.items():
+        path = f"{_prefix}{key}"
+        # An empty mapping has no leaves to descend to, so recursing would drop the override
+        # entirely. It is still a value the caller asked to set: emit it as ``++path={}``, which
+        # clears the subtree.
+        if isinstance(value, Mapping) and value:
+            arguments.extend(_flatten_overrides(value, f"{path}."))
+        else:
+            arguments.append(f"++{path}={_hydra_scalar(value)}")
+    return arguments
+
+
+def _redact_env_overrides(overrides: Mapping[str, Any], _prefix: str = "") -> dict[str, Any]:
+    """Redact credential-looking values from overrides before they are recorded as provenance.
+
+    ``env_overrides`` is a free-form escape hatch forwarded to Gym, so nothing stops a caller passing
+    ``{"model": {"api_key": "sk-..."}}``. ``RunnerInfo.config`` is persisted into the run bundle, so a
+    value that looks like a credential must not be written there.
 
     The *key* is always kept — knowing that a run overrode ``model.api_key`` is useful provenance;
-    knowing the value is a leak. Overrides that don't parse as ``key=value`` are kept verbatim: they
-    carry no value to leak.
+    knowing the value is a leak. Matching is on the full dotted path, so a marker anywhere in it
+    redacts, and nesting cannot hide a credential behind an innocuous leaf name.
+
+    Lists are walked too, since a mapping inside one — ``{"models": [{"api_key": "sk-..."}]}`` —
+    reaches Gym just as a nested mapping does. The index contributes no path segment: what marks a
+    value as a credential is the key it sits under, not where in a list it happens to fall.
     """
-    redacted: list[str] = []
-    for override in overrides:
-        key, sep, _ = override.partition("=")
-        if sep and any(marker in key.casefold() for marker in _SECRET_KEY_MARKERS):
-            redacted.append(f"{key}={_REDACTED}")
+    redacted: dict[str, Any] = {}
+    for key, value in overrides.items():
+        path = f"{_prefix}{key}"
+        if isinstance(value, Mapping):
+            redacted[key] = _redact_env_overrides(value, f"{path}.")
+        elif any(marker in path.casefold() for marker in _SECRET_KEY_MARKERS):
+            redacted[key] = _REDACTED
+        elif isinstance(value, (list, tuple)):
+            redacted[key] = [_redact_list_item(item, path) for item in value]
         else:
-            redacted.append(override)
+            redacted[key] = value
     return redacted
 
 
+def _redact_list_item(item: Any, path: str) -> Any:
+    """Redact inside one element of a list-valued override. See :func:`_redact_env_overrides`."""
+    if isinstance(item, Mapping):
+        return _redact_env_overrides(item, f"{path}.")
+    if isinstance(item, (list, tuple)):
+        return [_redact_list_item(nested, path) for nested in item]
+    return item
+
+
+def _selection_args(config: GymRuntimeConfig, work_dir: Path) -> list[str]:
+    """The environment/agent/model selection passed to Gym.
+
+    Built once and handed verbatim to both ``gym env validate`` and ``gym env start``, so what is
+    validated is exactly what runs — a pre-flight against a different config would be worse than
+    none.
+    """
+    selection = [
+        "--config",
+        config.agent_config,
+        "--model-type",
+        config.model_type,
+        "--resources-server",
+        config.resources_server,
+    ]
+    if config.bind_resources_server:
+        # Composable (Pattern-A) agents leave resources_server.name unbound ('???'); bind it to the
+        # env we're running. Assumes the agent config's top-level key equals the agent name (the
+        # simple_agent convention) *and* that the resources-server is registered under the
+        # environment's own name — not universally true, so self-contained or differently-named
+        # servers set bind_resources_server=False and bind themselves via env_overrides.
+        selection.append(
+            f"+{config.agent}.responses_api_agents.{config.agent}.resources_server.name={config.resources_server}"
+        )
+    selection.extend(_flatten_overrides(config.env_overrides))
+    # Gym is a Hydra app, so each invocation writes a timestamped run directory — by default
+    # `outputs/<date>/<time>/` under the *current* directory. Since the subprocesses inherit this
+    # process's cwd (so Gym can find env.yaml), the default would litter whatever directory the
+    # caller happened to run from. Redirect it under the run's work dir, with the rest of the run's
+    # artifacts. Applies to every Gym entry point, `gym list` included.
+    selection.append(f"hydra.run.dir={work_dir / _HYDRA_SUBDIR}")
+    return selection
+
+
 def _canonical_row_hash(row: Mapping[str, Any]) -> str:
     """Stable ``sha256`` of a Gym dataset row, excluding runtime-injected fields.
 
@@ -242,10 +394,13 @@ class GymRuntimeConfig(BaseModel):
         "(the composable/Pattern-A agent case, e.g. simple_agent whose config leaves it '???'). Set False for "
         "self-contained agents that already bind their own resources-server.",
     )
-    env_overrides: list[str] = Field(
-        default_factory=list,
-        description="Extra Hydra '+key=value' overrides for `gym env start` (escape hatch; applied after the "
-        "auto-derived resources-server binding).",
+    env_overrides: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Nested config overrides merged into Gym's config, applied after the auto-derived "
+        "resources-server binding. Structured rather than pre-serialized Hydra strings so the config "
+        "travels as JSON — `{'a': {'b': 1}}` becomes `++a.b=1` at invocation. This is the escape "
+        "hatch for what Gym does not standardize: an environment whose resources-server is registered "
+        "under a different name, or which references a model server no shipped config defines.",
     )
     num_repeats: int = Field(default=1, ge=1, description="Attempts per row; each attempt becomes one trial.")
     concurrency: int = Field(
@@ -273,18 +428,22 @@ class GymRuntimeConfig(BaseModel):
 def _gym_executable() -> str:
     """Locate the ``gym`` CLI on PATH, or fail saying what to do about it.
 
-    Gym is expected to be installed in the same environment as this SDK — there is deliberately no
-    config field pointing at a checkout or another venv, because these runner configs become
-    serialized job specs and a local filesystem path cannot cross that boundary. Resolving here
-    rather than at spawn time turns a missing Gym into one legible error instead of an ``ENOENT``
-    out of ``create_subprocess_exec`` after the run has already started.
+    Deliberately PATH-only, with no config field pointing at a checkout or a particular venv: these
+    runner configs become serialized job specs, and a local filesystem path cannot cross that
+    boundary. Resolving here rather than at spawn time turns a missing Gym into one legible error
+    instead of an ``ENOENT`` out of ``create_subprocess_exec`` after the run has already started.
+
+    Note that Gym generally cannot live in this SDK's own environment: it imports Ray at module load,
+    and nemo-platform excludes Ray by constraint over an unfixed CVE. Install Gym separately and put
+    its ``bin`` on PATH; in a job image, the image owns PATH and this resolves normally.
     """
     resolved = shutil.which(_GYM_CLI)
     if resolved is None:
         raise RuntimeError(
-            f"The {_GYM_CLI!r} CLI was not found on PATH. NeMo Gym must be installed in the same Python "
-            "environment as this SDK: `pip install nemo-gym`, plus the target environment's own "
-            "dependencies (each resources-server ships a requirements.txt)."
+            f"The {_GYM_CLI!r} CLI was not found on PATH. Install NeMo Gym in its own environment "
+            "(it needs Ray, which nemo-platform excludes over an unfixed CVE, so it cannot share this "
+            "one) and put that environment's `bin` directory on PATH. Each resources-server also "
+            "ships its own requirements.txt, installed from that server's directory."
         )
     return resolved
 
@@ -471,6 +630,54 @@ async def run_tasks(
         _require_full_coverage(tasks, covered_task_ids={trial.task_id for trial in trials}, rollouts_path=rollouts_path)
         return trials
 
+    async def _validate_config(
+        self, gym: str, selection: Sequence[str], subprocess_env: Mapping[str, str], work_dir: Path
+    ) -> None:
+        """Pre-flight the composed Gym config with ``gym env validate`` before starting anything.
+
+        Gym does not publish what configuration an environment requires — the typed
+        ``*ResourcesServerConfig`` classes cover behavioural knobs, while the model wiring lives in
+        each environment's YAML under names that vary per environment. ``gym env validate`` is the
+        only way to find out short of running: it merges configs, flags, and overrides, then reports
+        unresolved ``???`` values, bad paths, and cross-references to servers that are not defined —
+        without Ray and without starting a server.
+
+        Running it every time is worth the second it costs. A config mistake otherwise surfaces after
+        ``gym env start`` has brought up a Ray cluster and several uvicorn servers, as a readiness
+        timeout up to ``startup_timeout_s`` (240s by default) whose message says nothing about the
+        actual problem.
+        """
+        proc = await asyncio.create_subprocess_exec(
+            gym,
+            "env",
+            "validate",
+            *selection,
+            env=dict(subprocess_env),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.STDOUT,
+            # Its own process group, like the other two Gym invocations. `_terminate` signals the
+            # group via `killpg`, so without this a validate timeout would signal *our* group — the
+            # SDK process included.
+            start_new_session=True,
+        )
+        try:
+            stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=_VALIDATE_TIMEOUT_S)
+        except TimeoutError:
+            await _terminate(proc, grace_s=self._config.shutdown_grace_s)
+            raise RuntimeError(
+                f"`gym env validate` did not finish within {_VALIDATE_TIMEOUT_S}s for resources-server "
+                f"{self._config.resources_server!r}"
+            ) from None
+
+        report = stdout.decode("utf-8", errors="replace").strip()
+        (work_dir / "gym_validate.log").write_text(report, encoding="utf-8")
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"Gym rejected the composed config for resources-server {self._config.resources_server!r} "
+                f"before any server started. `gym env validate` said:\n\n{report}"
+            )
+        logger.debug("gym env validate: %s", report)
+
     async def _run_two_step(self, input_path: Path, output_path: Path, work_dir: Path) -> None:
         """Start the Gym servers, collect against them with ``--no-serve``, then tear them down."""
         cfg = self._config
@@ -483,25 +690,11 @@ async def _run_two_step(self, input_path: Path, output_path: Path, work_dir: Pat
         # hook is wrong for Gym (servers manage their own deps), so disable it for the subprocesses.
         subprocess_env = {**os.environ, "RAY_ENABLE_UV_RUN_RUNTIME_ENV": "0"}
 
-        env_cmd = [
-            gym,
-            "env",
-            "start",
-            "--config",
-            cfg.agent_config,
-            "--model-type",
-            cfg.model_type,
-            "--resources-server",
-            cfg.resources_server,
-        ]
-        if cfg.bind_resources_server:
-            # Composable (Pattern-A) agents leave resources_server.name unbound ('???'); bind it to the
-            # env we're running. Assumes the agent config's top-level key equals the agent name (the
-            # simple_agent convention). Self-contained agents set bind_resources_server=False.
-            env_cmd.append(
-                f"+{cfg.agent}.responses_api_agents.{cfg.agent}.resources_server.name={cfg.resources_server}"
-            )
-        env_cmd.extend(cfg.env_overrides)
+        selection = _selection_args(cfg, work_dir)
+
+        await self._validate_config(gym, selection, subprocess_env, work_dir)
+
+        env_cmd = [gym, "env", "start", *selection]
         # start_new_session=True puts `gym env start` in its own process group so teardown can signal
         # the *whole* Ray-cluster + uvicorn tree, not just the direct child (else they orphan).
         # stderr is merged into stdout here (unlike `eval run`, which splits them) because readiness
diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_environment_coverage.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_environment_coverage.py
new file mode 100644
index 0000000000..e1b1f40d70
--- /dev/null
+++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_environment_coverage.py
@@ -0,0 +1,345 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Coverage of the Gym runner across a representative sample of Gym environments.
+
+The other Gym tests (``test_gym_runtime.py``, ``test_gym_aggregate_scores.py``) replay recorded
+fixtures through the adapter and never invoke the CLI, so they encode whatever ``mcqa`` happens to do
+and keep passing for every other environment. This module goes the other way: it drives the real
+``gym`` CLI against several environments chosen to span Gym's dependency and wiring categories.
+
+Gym ships 105 ``resources_servers`` — 68 with two or fewer dependencies, 28 with multi-dependency CPU
+stacks, 6 heavy/GPU, and 10 requiring Docker — and ``mcqa`` is the easiest of all of them, so "the
+runner works" has only ever been established for the least demanding case.
+
+**Two layers, deliberately.** ``gym env validate`` merges configs, flags, and overrides and reports
+unset ``???`` values, bad paths, and dangling cross-references *without Ray, servers, a model
+endpoint, or credentials*. It is fast, offline, and answers most of what this sweep wants to know, so
+every environment gets a validate test. Actually collecting rollouts additionally needs a reachable
+model endpoint, so those tests skip unless one is configured.
+
+That split matters because model wiring is not standardized. ``mcqa``/``gpqa_diamond``/
+``wmt_translation`` reference a ``policy_model`` server fed by the global ``policy_base_url`` /
+``policy_api_key`` / ``policy_model_name`` keys; ``gdpval`` wants a judge server named
+``gdpval_judge_model`` that no shipped config defines; ``legal_agent_bench`` takes a raw
+``judge_base_url``; and ``wmt_translation`` also loads a local COMET model that is not an endpoint at
+all. Validate surfaces those differences as concrete errors instead of as a startup timeout.
+
+Prerequisites, in the order the skips report them:
+
+* **NeMo Gym installed in this environment** (``pip install nemo-gym``). Environments ship in the
+  wheel, so no checkout is required.
+* **The target environment's own dependencies** — each ``resources_server`` ships a
+  ``requirements.txt``; install it from that directory so its ``-e nemo-gym[dev] @ ../../`` resolves.
+* **Docker**, for environments whose agent works inside containers.
+* **A model endpoint** (``NEMO_GYM_POLICY_BASE_URL`` and friends), for the rollout tests only.
+
+Tracked by AALGO-485; the CI and GPU-runner story is AALGO-494.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import tempfile
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import pytest
+from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
+from nemo_evaluator_sdk.agent_eval.runtimes.gym_runtime import (
+    GymAgentTaskRunner,
+    GymRuntimeConfig,
+    _flatten_overrides,
+    discover_gym_tasks,
+)
+from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig
+from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrialStatus
+
+pytestmark = [pytest.mark.integration, pytest.mark.slow]
+
+#: Endpoint for the rollout tests. Gym itself reads credentials from a gitignored ``env.yaml`` in the
+#: working directory; this only gates whether we attempt a run at all.
+POLICY_BASE_URL_ENV = "NEMO_GYM_POLICY_BASE_URL"
+
+_DEFAULT_AGENT = "simple_agent"
+_DEFAULT_AGENT_CONFIG = "responses_api_agents/simple_agent/configs/simple_agent.yaml"
+
+#: Keep runs small — this establishes that an environment works at all, not how well it scores.
+_TASK_LIMIT = 2
+
+
+@dataclass(frozen=True)
+class GymEnvironmentCase:
+    """One environment in the sample, with the prerequisites it adds beyond an installed Gym."""
+
+    resources_server: str
+    category: str
+    #: Why this environment is in the sample rather than another from the same category.
+    rationale: str
+    needs_docker: bool = False
+    needs_gpu: bool = False
+    agent: str = _DEFAULT_AGENT
+    agent_config: str = _DEFAULT_AGENT_CONFIG
+    #: Mirrors ``GymRuntimeConfig.bind_resources_server``. False where the environment registers its
+    #: resources-server under a name other than its own, so the caller must bind it explicitly.
+    bind_resources_server: bool = True
+    #: Nested config overrides this environment needs beyond the runner's automatic binding. Empty
+    #: where the standard `policy_model` wiring is enough; populated as the sweep discovers what an
+    #: environment actually requires.
+    env_overrides: Mapping[str, Any] = field(default_factory=dict)
+
+    @property
+    def id(self) -> str:
+        return f"{self.resources_server}[{self.category}]"
+
+
+CASES: tuple[GymEnvironmentCase, ...] = (
+    GymEnvironmentCase(
+        resources_server="mcqa",
+        category="baseline",
+        rationale=(
+            "The environment every existing test is built on. Included so a failure elsewhere is "
+            "attributable to that environment rather than to the runner or the local setup."
+        ),
+    ),
+    GymEnvironmentCase(
+        resources_server="gpqa_diamond",
+        category="tiny",
+        rationale=(
+            "Control. Same multiple-choice shape as mcqa with comparably small dependencies, so a "
+            "failure here means we encoded mcqa's specifics rather than the category's."
+        ),
+    ),
+    GymEnvironmentCase(
+        resources_server="gdpval",
+        category="heavy-cpu-judge",
+        rationale=(
+            "Seven dependencies including binary wheels (PyMuPDF, pdfminer.six, python-docx), no "
+            "GPU, and it ships a prompt config so it exercises the data-prep path mcqa skips. Also "
+            "the clearest case of model wiring that is not standardized — see the overrides below."
+        ),
+        # Two Gym conventions this environment does not follow, both of which the caller has to know
+        # about and neither of which is discoverable without reading its YAML or failing:
+        #
+        # 1. Its resources-server is registered as `gdpval_resources_server`, not `gdpval`. The
+        #    runner's `bind_resources_server` binds the *environment* name (the simple_agent
+        #    convention), so it must be disabled and bound by hand here.
+        # 2. It references a judge model server named `gdpval_judge_model` that no shipped config
+        #    defines — `responses_api_models/` provides model *types*, not this block — so the caller
+        #    must construct it.
+        #
+        # We deliberately do not paper over either in the runner: they are Gym's inconsistencies, and
+        # hiding them behind guesswork would make the runner wrong for environments that follow the
+        # convention. Recorded here so the cost to a caller is visible.
+        bind_resources_server=False,
+        env_overrides={
+            "simple_agent": {
+                "responses_api_agents": {"simple_agent": {"resources_server": {"name": "gdpval_resources_server"}}}
+            },
+            "gdpval_judge_model": {
+                "responses_api_models": {
+                    "inference_provider": {
+                        "entrypoint": "app.py",
+                        # Interpolations, not literals: the judge shares the policy endpoint rather
+                        # than needing a second one configured.
+                        "base_url": "${policy_base_url}",
+                        "api_key": "${policy_api_key}",
+                        "model": "${policy_model_name}",
+                    }
+                }
+            },
+        },
+    ),
+    GymEnvironmentCase(
+        resources_server="legal_agent_bench",
+        category="docker",
+        rationale=(
+            "Harbor executes the task-local verifier inside Docker; needs a running daemon, ~10 GB "
+            "of space, and a multi-minute first image build. Takes a raw `judge_base_url` rather "
+            "than a model-server reference. Also runnable through the SDK's Harbor and Fabric "
+            "examples, so it is the one environment comparable across three runners."
+        ),
+        needs_docker=True,
+    ),
+    GymEnvironmentCase(
+        resources_server="wmt_translation",
+        category="gpu",
+        rationale=(
+            "torch, torchvision, and unbabel-comet, plus a local COMET model (Unbabel/XCOMET-XXL) "
+            "that is not an endpoint at all. Expected to fail without a GPU — the point of running "
+            "it anyway is to check the failure is legible."
+        ),
+        needs_gpu=True,
+    ),
+)
+
+
+def _gym_cli() -> str:
+    """The `gym` CLI, or skip. Everything here needs it; nothing here needs a checkout."""
+    resolved = shutil.which("gym")
+    if resolved is None:
+        pytest.skip("NeMo Gym is not installed in this environment (`pip install nemo-gym`)")
+    return resolved
+
+
+def _bounded_probe(argv: list[str]) -> bool:
+    """True when the command exists and exits zero. Bounded: a wedged daemon must not stall the run."""
+    if shutil.which(argv[0]) is None:
+        return False
+    try:
+        return subprocess.run(argv, capture_output=True, timeout=10).returncode == 0
+    except subprocess.TimeoutExpired:
+        return False
+
+
+def _environment_dir(gym: str, case: GymEnvironmentCase) -> Path:
+    """Locate an environment's directory by asking the CLI, or skip.
+
+    Deliberately not ``importlib.resources``: nemo-platform bans Ray by constraint
+    (``"ray; sys_platform == 'never'"`` in the root pyproject, for an unfixed CVE) while Gym imports
+    Ray at module load, so Gym cannot be installed in this SDK's environment. It lives in its own,
+    reachable only through the ``gym`` binary on PATH — which means ``resources_servers.<name>`` is
+    not importable here even when the environment is perfectly available.
+
+    ``gym list resources-servers <name>`` reports the environment's config path and exits non-zero
+    for one it cannot resolve, so it answers both questions across the venv boundary.
+    """
+    # `gym list` is a Hydra entry point too, so it also wants somewhere to put its run directory —
+    # without this it drops an `outputs/<date>/<time>/` into whatever directory pytest ran from.
+    with tempfile.TemporaryDirectory(prefix="gym-list-hydra-") as hydra_dir:
+        proc = subprocess.run(
+            [gym, "list", "resources-servers", case.resources_server, f"hydra.run.dir={hydra_dir}"],
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+
+    if proc.returncode != 0:
+        pytest.skip(
+            f"Gym cannot resolve resources-server {case.resources_server!r}; install its own "
+            "requirements.txt from the environment's directory (its `-e nemo-gym[dev] @ ../../` is "
+            "relative to that directory, so it must be installed from there)"
+        )
+    config_line = next((ln for ln in proc.stdout.splitlines() if ln.startswith("config:")), None)
+    if config_line is None:
+        pytest.skip(f"`gym list resources-servers {case.resources_server}` reported no config path to locate it by")
+    # <env>/configs/<env>.yaml -> <env>
+    return Path(config_line.removeprefix("config:").strip()).parent.parent
+
+
+def _require_environment(gym: str, case: GymEnvironmentCase) -> Path:
+    """Skip with one specific reason unless this case's prerequisites are present.
+
+    The reasons are the deliverable: ``pytest -ra`` over this module is the coverage map, so each one
+    names the single missing thing rather than a generic "prerequisites not met".
+    """
+    environment_dir = _environment_dir(gym, case)
+
+    if case.needs_docker and not _bounded_probe(["docker", "info"]):
+        pytest.skip(f"{case.resources_server} runs its verifier in Docker; no working daemon found")
+
+    if case.needs_gpu and not _bounded_probe(["nvidia-smi"]):
+        pytest.skip(f"{case.resources_server} scores with GPU models; no working nvidia-smi found")
+
+    return environment_dir
+
+
+def _selection(case: GymEnvironmentCase, hydra_dir: Path) -> list[str]:
+    """The selection arguments the runner passes to both `gym env validate` and `gym env start`.
+
+    ``hydra.run.dir`` mirrors what the runner does: Gym is a Hydra app and writes a timestamped run
+    directory per invocation, defaulting to ``outputs/`` under the current directory. Left alone, a
+    sweep drops one of those into the repo for every environment it checks.
+    """
+    argv = [
+        "--config",
+        case.agent_config,
+        "--model-type",
+        "inference_provider",
+        "--resources-server",
+        case.resources_server,
+    ]
+    if case.bind_resources_server:
+        argv.append(f"+{case.agent}.responses_api_agents.{case.agent}.resources_server.name={case.resources_server}")
+    argv.extend(_flatten_overrides(case.env_overrides))
+    argv.append(f"hydra.run.dir={hydra_dir}")
+    return argv
+
+
+@pytest.mark.parametrize("case", CASES, ids=[case.id for case in CASES])
+def test_gym_environment_config_validates(case: GymEnvironmentCase, tmp_path: Path) -> None:
+    """`gym env validate` accepts the config the runner would run.
+
+    No Ray, no servers, no model endpoint, no credentials — so this runs anywhere Gym is installed
+    and is where most of the sweep's signal comes from. A failure here is a concrete, readable
+    report: an unset ``???``, a bad path, or a cross-reference to a server that is not defined.
+    """
+    gym = _gym_cli()
+    # Only the environment itself is required. Docker and GPU are *execution* prerequisites, and
+    # gating on them here would throw away the coverage this test exists for — `wmt_translation`'s
+    # config would go unvalidated on every machine without an NVIDIA card, which is all of them
+    # until AALGO-494 lands the GPU runner.
+    _environment_dir(gym, case)
+
+    proc = subprocess.run(
+        [gym, "env", "validate", *_selection(case, tmp_path / "hydra")],
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    assert proc.returncode == 0, (
+        f"{case.resources_server}: `gym env validate` rejected the runner's config.\n{proc.stdout}\n{proc.stderr}"
+    )
+
+
+@pytest.mark.timeout(1800)
+@pytest.mark.parametrize("case", CASES, ids=[case.id for case in CASES])
+@pytest.mark.asyncio
+async def test_gym_environment_runs_end_to_end(case: GymEnvironmentCase, tmp_path: Path) -> None:
+    """Collect real rollouts and confirm the runner turns them into scored trials.
+
+    Deliberately shallow: this establishes that a category works, not how well it scores. Reward
+    values are Gym's own and are not asserted on — that would be testing Gym, not the runner.
+    """
+    gym = _gym_cli()
+    environment_dir = _require_environment(gym, case)
+    if not os.environ.get(POLICY_BASE_URL_ENV):
+        pytest.skip(f"{POLICY_BASE_URL_ENV} is not set; rollout collection needs a reachable model endpoint")
+
+    dataset = environment_dir / "data" / "example.jsonl"
+    if not dataset.exists():
+        pytest.skip(f"{case.resources_server} ships no data/example.jsonl at {dataset}")
+
+    tasks = discover_gym_tasks(dataset)
+    assert tasks, f"{case.resources_server}: discover_gym_tasks found no rows in {dataset}"
+    tasks = tasks[:_TASK_LIMIT]
+
+    runner = GymAgentTaskRunner(
+        config=GymRuntimeConfig(
+            agent=case.agent,
+            agent_config=case.agent_config,
+            resources_server=case.resources_server,
+            num_repeats=1,
+            bind_resources_server=case.bind_resources_server,
+            env_overrides=dict(case.env_overrides),
+        )
+    )
+
+    result = await AgentEvaluator().run(
+        tasks=tasks,
+        target=runner,
+        config=AgentEvalRunConfig(work_dir=tmp_path / "gym_run", parallelism=1),
+    )
+
+    assert result.summary.task_count == len(tasks)
+    assert len(result.trials) == len(tasks)
+
+    failed = [trial for trial in result.trials if trial.status is not AgentEvalTrialStatus.COMPLETED]
+    assert not failed, f"{case.resources_server}: trials did not complete: {[(t.task_id, t.status) for t in failed]}"
+
+    assert runner.runner_info().name == "gym"
+    assert result.scores, f"{case.resources_server}: no metric scores were produced"
diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_runtime.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_runtime.py
index 96798ae089..6093443061 100644
--- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_runtime.py
+++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_gym_runtime.py
@@ -14,6 +14,7 @@
 import asyncio
 import json
 import logging
+import shutil
 from collections import Counter, deque
 from pathlib import Path
 
@@ -22,17 +23,23 @@
     _LOG_TAIL_LINES,
     NG_ROLLOUT_INDEX,
     NG_TASK_INDEX,
+    GymAgentTaskRunner,
     GymRewardMetric,
+    GymRuntimeConfig,
     _aggregate_metrics_path_for,
     _canonical_row_hash,
     _content_text,
     _drain_pumps,
     _ensure_fresh_output,
+    _flatten_overrides,
+    _gym_executable,
+    _hydra_scalar,
     _materialize_dataset,
     _pump_stream,
     _read_run_aggregations,
     _render_instruction,
     _require_full_coverage,
+    _selection_args,
     _source_datasets,
     _trials_from_rollouts,
     discover_gym_tasks,
@@ -578,3 +585,162 @@ def test_aggregate_metrics_sidecar_with_invalid_utf8_is_skipped_not_raised(tmp_p
     _aggregate_metrics_path_for(rollouts_path).write_bytes(b'[{"agent_ref": {"name": "a"}, "x": \xff}]')
 
     assert _read_run_aggregations(rollouts_path) is None
+
+
+# ---------------------------------------------------------------------------
+# Config pre-flight: override serialization, selection args, `gym env validate`
+# ---------------------------------------------------------------------------
+
+
+def test_hydra_scalars_use_hydra_spellings_not_python_ones() -> None:
+    # `str(None)` is "None" and `str(True)` is "True" — both of which Hydra reads back as *strings*,
+    # silently setting the literal text instead of a null or a boolean.
+    assert _hydra_scalar(None) == "null"
+    assert _hydra_scalar(True) == "true"
+    assert _hydra_scalar(False) == "false"
+    assert _hydra_scalar([1, None]) == "[1,null]"
+    assert _hydra_scalar(0.7) == "0.7"
+
+
+def test_hydra_strings_are_quoted_so_the_grammar_cannot_retype_them() -> None:
+    # Hydra's grammar is typed: unquoted, "true" parses as a bool, "null" as None, "1.5" as a float,
+    # and "a,b" as a *sweep* — so a string override would silently become something else.
+    assert _hydra_scalar("true") == "'true'"
+    assert _hydra_scalar("null") == "'null'"
+    assert _hydra_scalar("1.5") == "'1.5'"
+    assert _hydra_scalar("a,b") == "'a,b'"
+    assert _hydra_scalar("A[B") == "'A[B'"  # unquoted this does not parse at all
+    assert _hydra_scalar("") == "''"
+    # Only the quote is escaped: Hydra does not decode `\\` inside a quoted value, so escaping
+    # backslashes would double them.
+    assert _hydra_scalar("he'llo") == "'he\\'llo'"
+    assert _hydra_scalar("back\\slash") == "'back\\slash'"
+    # Interpolation survives quoting — the override sets the text, OmegaConf resolves it on read.
+    assert _hydra_scalar("${policy_base_url}") == "'${policy_base_url}'"
+
+
+def test_hydra_rejects_a_value_it_cannot_express() -> None:
+    # A trailing backslash escapes the closing quote and leaves the value unterminated; there is no
+    # spelling that avoids it, so say so rather than emit something unparseable.
+    with pytest.raises(ValueError, match="ends with a backslash"):
+        _hydra_scalar("ends\\")
+
+
+def test_hydra_dicts_nested_in_a_list_use_the_grammars_bare_keys() -> None:
+    # A mapping reached through a list has no dotted path to flatten onto, so it has to be spelled
+    # inline. `str()` would emit Python's repr — `{'b': 1}` — whose quoted keys Hydra's `dictKey`
+    # rule has no form for and rejects outright.
+    assert _hydra_scalar({"b": 1}) == "{b:1}"
+    assert _hydra_scalar([{"b": 1}, {"c": "true"}]) == "[{b:1},{c:'true'}]"
+    # The typed spellings hold at depth: values recurse through the same function.
+    assert _hydra_scalar({"b": None, "c": True, "d": "a,b"}) == "{b:null,c:true,d:'a,b'}"
+    assert _hydra_scalar({"b": {"c": [1, "x"]}}) == "{b:{c:[1,'x']}}"
+    assert _hydra_scalar({}) == "{}"
+
+
+@pytest.mark.parametrize("key", ["true", "NULL", "1.5", "inf", "b:c", "b,c", "b}c", "b'c", "", 1])
+def test_hydra_rejects_dict_keys_it_would_retype_or_fail_to_parse(key: object) -> None:
+    # Keys are emitted bare because the grammar has no quoted form, which leaves them at the mercy
+    # of the lexer: `{true:1}` keys on the boolean True and `{b:c:1}` does not parse. Neither is what
+    # the caller wrote, and the second is not even loud, so refuse both.
+    with pytest.raises(ValueError, match="dict key"):
+        _hydra_scalar([{key: 1}])
+
+
+def test_flatten_overrides_produces_forcing_dotted_paths() -> None:
+    flattened = _flatten_overrides({"a": {"b": {"c": 1}}, "d": "x"})
+    # `++` not `+`: a bare `+` fails on a key the merged config already defines, which is precisely
+    # the case an override exists for.
+    assert flattened == ["++a.b.c=1", "++d='x'"]
+    assert _flatten_overrides({}) == []
+
+
+def test_flatten_overrides_keeps_an_empty_mapping_instead_of_dropping_it() -> None:
+    # An empty mapping has no leaves to descend to, so recursing emits nothing and the override
+    # vanishes — the caller asked to clear `a` and the run would silently keep the config's value.
+    assert _flatten_overrides({"a": {}}) == ["++a={}"]
+
+
+def test_flatten_overrides_serializes_a_list_of_dicts() -> None:
+    assert _flatten_overrides({"a": {"b": [{"c": 1}]}}) == ["++a.b=[{c:1}]"]
+
+
+def _config(**kwargs: object) -> GymRuntimeConfig:
+    return GymRuntimeConfig(agent="simple_agent", agent_config="cfg.yaml", resources_server="mcqa", **kwargs)  # type: ignore[arg-type]
+
+
+def test_selection_binds_the_resources_server_by_default(tmp_path: Path) -> None:
+    selection = _selection_args(_config(), tmp_path)
+    assert "--resources-server" in selection and "mcqa" in selection
+    assert "+simple_agent.responses_api_agents.simple_agent.resources_server.name=mcqa" in selection
+
+
+def test_selection_omits_the_binding_when_the_caller_binds_it_themselves(tmp_path: Path) -> None:
+    # gdpval registers its server as `gdpval_resources_server`, so the automatic binding is wrong for
+    # it and the caller supplies their own. Emitting both would leave the config ambiguous.
+    selection = _selection_args(
+        _config(
+            bind_resources_server=False,
+            env_overrides={
+                "simple_agent": {"responses_api_agents": {"simple_agent": {"resources_server": {"name": "other"}}}}
+            },
+        ),
+        tmp_path,
+    )
+    assert not [arg for arg in selection if arg.startswith("+simple_agent.")]
+    assert "++simple_agent.responses_api_agents.simple_agent.resources_server.name='other'" in selection
+
+
+def test_selection_redirects_hydra_output_under_the_run_work_dir(tmp_path: Path) -> None:
+    # Gym writes `outputs/<date>/<time>/` relative to cwd, and the subprocesses inherit ours so Gym
+    # can find env.yaml — so without this every run litters the caller's directory.
+    selection = _selection_args(_config(), tmp_path)
+    assert f"hydra.run.dir={tmp_path / 'gym_hydra'}" in selection
+
+
+def _stub_gym(tmp_path: Path, *, exit_code: int, message: str) -> str:
+    """A stand-in for the `gym` CLI that records its argv and exits how the test wants."""
+    script = tmp_path / "stub-gym"
+    script.write_text(
+        "#!/usr/bin/env python3\n"
+        "import pathlib, sys\n"
+        f"pathlib.Path({str(tmp_path / 'argv.txt')!r}).write_text('\\n'.join(sys.argv[1:]))\n"
+        f"print({message!r})\n"
+        f"sys.exit({exit_code})\n",
+        encoding="utf-8",
+    )
+    script.chmod(0o755)
+    return str(script)
+
+
+@pytest.mark.asyncio
+async def test_validate_config_passes_the_selection_and_logs_the_report(tmp_path: Path) -> None:
+    runner = GymAgentTaskRunner(config=_config())
+    gym = _stub_gym(tmp_path, exit_code=0, message="Config is valid.")
+
+    await runner._validate_config(gym, ["--resources-server", "mcqa"], {}, tmp_path)
+
+    assert (tmp_path / "argv.txt").read_text().splitlines() == ["env", "validate", "--resources-server", "mcqa"]
+    # Kept next to the run's other logs so a passing pre-flight is still auditable afterwards.
+    assert (tmp_path / "gym_validate.log").read_text().strip() == "Config is valid."
+
+
+@pytest.mark.asyncio
+async def test_validate_config_raises_with_gyms_own_report(tmp_path: Path) -> None:
+    # The whole point of the pre-flight: surface Gym's diagnosis before a Ray cluster and several
+    # uvicorn servers start, rather than as a readiness timeout that says nothing about the cause.
+    complaint = "Error: references resources_servers/'gdpval', which is not defined"
+    runner = GymAgentTaskRunner(config=_config())
+    gym = _stub_gym(tmp_path, exit_code=1, message=complaint)
+
+    with pytest.raises(RuntimeError) as excinfo:
+        await runner._validate_config(gym, [], {}, tmp_path)
+
+    assert complaint in str(excinfo.value)
+    assert "mcqa" in str(excinfo.value)
+
+
+def test_gym_executable_reports_how_to_install_when_absent(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(shutil, "which", lambda _: None)
+    with pytest.raises(RuntimeError, match="own environment"):
+        _gym_executable()
diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_run_metadata.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_run_metadata.py
index 326ede598e..496d85587d 100644
--- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_run_metadata.py
+++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_run_metadata.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import cast
 
 from nemo_evaluator_sdk.agent_eval.evaluator import _describe_target
@@ -195,24 +196,35 @@ def test_gym_redacts_credential_looking_env_overrides() -> None:
             agent="a",
             agent_config="c",
             resources_server="r",
-            env_overrides=[
-                "+model.api_key=sk-should-not-be-recorded",
-                "+env.HF_TOKEN=hf_should-not-be-recorded",
-                "+agent.temperature=0.7",
-                "--flagged-without-a-value",
-            ],
+            env_overrides={
+                "model": {"api_key": "sk-should-not-be-recorded", "temperature": 0.7},
+                "env": {"HF_TOKEN": "hf_should-not-be-recorded"},
+                "agent": {"nested": {"secret": "deep-should-not-be-recorded"}},
+                "models": [
+                    {"name": "m", "api_key": "sk-in-a-list-should-not-be-recorded"},
+                ],
+                "api_keys": ["sk-whole-list-should-not-be-recorded"],
+            },
         )
     )
 
     recorded = runner.runner_info().config["env_overrides"]
 
-    assert recorded == [
-        "+model.api_key=<redacted>",
-        "+env.HF_TOKEN=<redacted>",
-        "+agent.temperature=0.7",  # not credential-shaped, kept verbatim for reproducibility
-        "--flagged-without-a-value",  # no value to leak
-    ]
-    assert not any("sk-" in entry or "hf_" in entry for entry in recorded)
+    assert recorded == {
+        "model": {
+            "api_key": "<redacted>",
+            "temperature": 0.7,  # not credential-shaped, kept verbatim for reproducibility
+        },
+        "env": {"HF_TOKEN": "<redacted>"},
+        # Matching is on the full dotted path, so a credential stays redacted at any depth.
+        "agent": {"nested": {"secret": "<redacted>"}},
+        # A mapping inside a list reaches Gym just as a nested one does, so it is walked too. The
+        # index contributes no path segment: the key marks the credential, not the position.
+        "models": [{"name": "m", "api_key": "<redacted>"}],
+        # A credential-shaped key wins over descending into it — the whole list goes.
+        "api_keys": "<redacted>",
+    }
+    assert "should-not-be-recorded" not in json.dumps(recorded)
 
 
 def test_model_provenance_records_the_endpoint_and_invocation_params() -> None:
diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml
index 68d07883f4..62cd0f1f86 100644
--- a/plugins/nemo-evaluator/openapi/openapi.yaml
+++ b/plugins/nemo-evaluator/openapi/openapi.yaml
@@ -3778,12 +3778,13 @@ components:
             Set False for self-contained agents that already bind their own resources-server.
           default: true
         env_overrides:
-          items:
-            type: string
-          type: array
+          additionalProperties: true
+          type: object
           title: Env Overrides
-          description: Extra Hydra '+key=value' overrides for `gym env start` (applied
-            after the auto-derived resources-server binding).
+          description: "Extra config overrides for `gym env start`, as nested data\
+            \ \u2014 {'model': {'temperature': 0.7}} rather than pre-serialized Hydra\
+            \ strings \u2014 so a spec survives being sent as JSON. Flattened to Hydra's\
+            \ grammar at invocation, after the auto-derived resources-server binding."
         num_repeats:
           type: integer
           minimum: 1.0
diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py
index a9671e816a..77afdcc88c 100644
--- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py
+++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py
@@ -173,10 +173,11 @@ class GymRunnerTarget(BaseModel):
         description="Auto-bind the agent's `resources_server.name` via a Hydra override. Set False for "
         "self-contained agents that already bind their own resources-server.",
     )
-    env_overrides: list[str] = Field(
-        default_factory=list,
-        description="Extra Hydra '+key=value' overrides for `gym env start` (applied after the auto-derived "
-        "resources-server binding).",
+    env_overrides: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Extra config overrides for `gym env start`, as nested data — {'model': {'temperature': 0.7}} "
+        "rather than pre-serialized Hydra strings — so a spec survives being sent as JSON. Flattened to "
+        "Hydra's grammar at invocation, after the auto-derived resources-server binding.",
     )
     num_repeats: int = Field(default=1, ge=1, description="Attempts per row; each attempt becomes one trial.")
     concurrency: int = Field(
diff --git a/plugins/nemo-evaluator/tests/test_agent_evaluate.py b/plugins/nemo-evaluator/tests/test_agent_evaluate.py
index 3ca9c7f2ee..248aef70ff 100644
--- a/plugins/nemo-evaluator/tests/test_agent_evaluate.py
+++ b/plugins/nemo-evaluator/tests/test_agent_evaluate.py
@@ -314,6 +314,7 @@ def test_resolve_target_builds_gym_runtime_from_runner_target(tmp_path: Path) ->
         num_repeats=2,
         concurrency=4,
         reward_key="score",
+        env_overrides={"model": {"temperature": 0.7}},
     )
     target, prompt_template, params = AgentEvalJob._resolve_target(gym_target, ctx)
     assert isinstance(target, GymAgentTaskRunner)
@@ -321,6 +322,9 @@ def test_resolve_target_builds_gym_runtime_from_runner_target(tmp_path: Path) ->
     assert target._config.resources_server == "mcqa"
     assert target._config.num_repeats == 2
     assert target._config.reward_key == "score"
+    # Overrides are nested data on both sides of the seam — the spec model and the runtime config
+    # must agree on the shape, or the spec validates and the runtime rejects it.
+    assert target._config.env_overrides == {"model": {"temperature": 0.7}}
     # A runner shapes its own request, so it contributes no prompt template or inference params.
     assert prompt_template is None
     assert params is None
diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/gym_runtime.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/gym_runtime.py
index 912d6e8c30..2bbf175d18 100644
--- a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/gym_runtime.py
+++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/gym_runtime.py
@@ -28,10 +28,14 @@
 means a caller can run a **subset** of tasks without Gym rolling out the rest.
 
 **Execution** is the two-step Gym flow (the one that reads a dataset directly
-without triggering Gym's split-driven data-prep): ``gym env start`` brings up the
-resources-server + agent + model servers, then ``gym eval run --no-serve --input
-<materialized dataset>`` collects rollouts against them. The runtime shells out
-to the ``gym`` CLI on PATH, so this SDK never imports ``nemo_gym``. Subprocess
+without triggering Gym's split-driven data-prep), preceded by a pre-flight:
+``gym env validate`` merges the composed config and reports unset ``???`` values,
+bad paths, and dangling cross-references without starting anything; then ``gym env
+start`` brings up the resources-server + agent + model servers, and ``gym eval run
+--no-serve --input <materialized dataset>`` collects rollouts against them. Both
+commands receive the identical selection arguments, so what is validated is what
+runs. The runtime shells out to the ``gym`` CLI on PATH, so this SDK never imports
+``nemo_gym``. Subprocess
 output is streamed to log files under the run's work dir *and* mirrored to this
 module's logger at ``DEBUG``, so callers choose terminal visibility through
 ordinary ``logging`` configuration.
@@ -95,6 +99,11 @@
 #: once, and that path is what the subprocesses run — so a child whose PATH differs from ours cannot
 #: end up executing a different Gym.
 _GYM_CLI = "gym"
+#: Bound on `gym env validate`. It merges config without starting anything and returns in about a
+#: second; this only exists so a wedged invocation cannot stall the run before it begins.
+_VALIDATE_TIMEOUT_S = 120.0
+#: Where Gym's Hydra run directories are redirected, relative to the run's work dir.
+_HYDRA_SUBDIR = "gym_hydra"
 #: Gym's index fields on each rollout record. ``_ng_task_index`` is the only join back to the input
 #: rows that survives a round-trip: Gym mutates ``responses_create_params`` (even the prompt) and
 #: copies only a fixed allowlist of row keys onto the result, so no field we invent comes back. Gym
@@ -107,34 +116,177 @@
 _LOG_TAIL_LINES = 40
 
 
-#: Substrings that mark a Hydra override key as carrying a credential. Matched case-insensitively
-#: against the key half of ``+key=value``.
+#: Substrings that mark an override as carrying a credential. Matched case-insensitively against the
+#: full dotted path, so nesting cannot hide one behind an innocuous leaf name.
 _SECRET_KEY_MARKERS = ("api_key", "apikey", "token", "secret", "password", "passwd", "credential")
 #: Stand-in written in place of a redacted override value.
 _REDACTED = "<redacted>"
 
 
-def _redact_env_overrides(overrides: Sequence[str]) -> list[str]:
-    """Redact credential-looking values from Hydra overrides before they are recorded as provenance.
+#: Dict keys Hydra reads back unchanged. Its ``dictKey`` rule accepts no quoting, so a key is
+#: whatever the lexer makes of the bare text — this is deliberately narrower than what parses.
+_HYDRA_DICT_KEY = re.compile(r"[A-Za-z_][A-Za-z0-9_.-]*\Z")
+#: Bare words the lexer types rather than reading as text, so they cannot serve as string keys.
+_HYDRA_KEY_LITERALS = frozenset({"true", "false", "null", "inf", "nan"})
 
-    ``env_overrides`` is a free-form escape hatch forwarded verbatim to ``gym env start``, so nothing
-    stops a caller passing ``+model.api_key=sk-...``. ``RunnerInfo.config`` is persisted into the run
-    bundle, so a value that looks like a credential must not be written there.
+
+def _hydra_dict(value: Mapping[str, Any]) -> str:
+    """Render a mapping as a Hydra dict container, ``{key:value,...}``.
+
+    Reached for a mapping nested inside a container — ``[{"b": 1}]`` — where there is no dotted path
+    to flatten onto, so the dict has to be spelled inline. Values recurse, so the typed spellings
+    below hold at any depth.
+
+    Keys are emitted bare, because Hydra's ``dictKey`` rule has no quoted form: ``{'b':1}`` does not
+    parse at all. That leaves the key at the mercy of the lexer, which types it — ``{true:1}`` keys
+    on the boolean ``True``, ``{1.5:1}`` on a float — and rejects ``:``, ``,``, brackets, and quotes
+    outright. Anything outside the conservative shape above therefore raises here rather than
+    silently keying the config on something the caller did not write.
+    """
+    rendered = []
+    for key, item in value.items():
+        if not isinstance(key, str) or not _HYDRA_DICT_KEY.match(key) or key.casefold() in _HYDRA_KEY_LITERALS:
+            raise ValueError(
+                f"Gym config override has dict key {key!r}, which Hydra's override grammar cannot "
+                "express as a string: keys are unquoted, so only a leading letter or underscore "
+                "followed by letters, digits, '_', '.', or '-' survives the round trip. Set this "
+                "key through the override path instead of nesting it inside a list."
+            )
+        rendered.append(f"{key}:{_hydra_scalar(item)}")
+    return "{" + ",".join(rendered) + "}"
+
+
+def _hydra_scalar(value: Any) -> str:
+    """Render a leaf value the way Hydra's override grammar reads it back.
+
+    Hydra's grammar is typed, so an unquoted string is not necessarily a string: ``true`` parses as a
+    boolean, ``null`` as ``None``, ``1.5`` as a float, ``a,b`` as a *sweep*, and ``A[B`` fails to
+    parse outright. Strings are therefore always single-quoted, which round-trips every one of those
+    (verified against ``hydra.core.override_parser``). Interpolations survive quoting — the override
+    sets the literal text and OmegaConf resolves it on read — so ``${policy_base_url}`` still works.
+
+    Only ``'`` is escaped. Hydra does **not** decode ``\\\\`` inside a quoted value: escaping
+    backslashes doubles them, so they are passed through raw.
+
+    ``None`` and booleans get their own spellings, since ``str()`` would emit ``"None"``/``"True"``
+    and Hydra reads those back as text. Containers recurse for the same reason: ``str()`` on a dict
+    emits Python's repr, whose quoted keys Hydra rejects outright.
+    """
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, Mapping):
+        return _hydra_dict(value)
+    if isinstance(value, (list, tuple)):
+        return "[" + ",".join(_hydra_scalar(item) for item in value) + "]"
+    if isinstance(value, str):
+        # A trailing backslash would escape the closing quote and leave the value unterminated, and
+        # there is no spelling that avoids it — better to say so than to emit something unparseable.
+        if value.endswith("\\"):
+            raise ValueError(
+                f"Gym config override value {value!r} ends with a backslash, which Hydra's override "
+                "grammar cannot express: it escapes the closing quote."
+            )
+        return "'" + value.replace("'", "\\'") + "'"
+    return str(value)
+
+
+def _flatten_overrides(overrides: Mapping[str, Any], _prefix: str = "") -> list[str]:
+    """Flatten a nested override mapping into Hydra ``++dotted.path=value`` arguments.
+
+    Callers describe overrides as structured data — ``{"a": {"b": 1}}`` — rather than as
+    pre-serialized Hydra strings, so the config survives being sent somewhere as JSON. Hydra itself
+    only speaks the flat form, so the translation happens here, at the point of invocation.
+
+    ``++`` rather than ``+``: it sets a key whether or not it already exists, which is what an
+    override means. A bare ``+`` fails on a key the merged config already defines.
+    """
+    arguments: list[str] = []
+    for key, value in overrides.items():
+        path = f"{_prefix}{key}"
+        # An empty mapping has no leaves to descend to, so recursing would drop the override
+        # entirely. It is still a value the caller asked to set: emit it as ``++path={}``, which
+        # clears the subtree.
+        if isinstance(value, Mapping) and value:
+            arguments.extend(_flatten_overrides(value, f"{path}."))
+        else:
+            arguments.append(f"++{path}={_hydra_scalar(value)}")
+    return arguments
+
+
+def _redact_env_overrides(overrides: Mapping[str, Any], _prefix: str = "") -> dict[str, Any]:
+    """Redact credential-looking values from overrides before they are recorded as provenance.
+
+    ``env_overrides`` is a free-form escape hatch forwarded to Gym, so nothing stops a caller passing
+    ``{"model": {"api_key": "sk-..."}}``. ``RunnerInfo.config`` is persisted into the run bundle, so a
+    value that looks like a credential must not be written there.
 
     The *key* is always kept — knowing that a run overrode ``model.api_key`` is useful provenance;
-    knowing the value is a leak. Overrides that don't parse as ``key=value`` are kept verbatim: they
-    carry no value to leak.
+    knowing the value is a leak. Matching is on the full dotted path, so a marker anywhere in it
+    redacts, and nesting cannot hide a credential behind an innocuous leaf name.
+
+    Lists are walked too, since a mapping inside one — ``{"models": [{"api_key": "sk-..."}]}`` —
+    reaches Gym just as a nested mapping does. The index contributes no path segment: what marks a
+    value as a credential is the key it sits under, not where in a list it happens to fall.
     """
-    redacted: list[str] = []
-    for override in overrides:
-        key, sep, _ = override.partition("=")
-        if sep and any(marker in key.casefold() for marker in _SECRET_KEY_MARKERS):
-            redacted.append(f"{key}={_REDACTED}")
+    redacted: dict[str, Any] = {}
+    for key, value in overrides.items():
+        path = f"{_prefix}{key}"
+        if isinstance(value, Mapping):
+            redacted[key] = _redact_env_overrides(value, f"{path}.")
+        elif any(marker in path.casefold() for marker in _SECRET_KEY_MARKERS):
+            redacted[key] = _REDACTED
+        elif isinstance(value, (list, tuple)):
+            redacted[key] = [_redact_list_item(item, path) for item in value]
         else:
-            redacted.append(override)
+            redacted[key] = value
     return redacted
 
 
+def _redact_list_item(item: Any, path: str) -> Any:
+    """Redact inside one element of a list-valued override. See :func:`_redact_env_overrides`."""
+    if isinstance(item, Mapping):
+        return _redact_env_overrides(item, f"{path}.")
+    if isinstance(item, (list, tuple)):
+        return [_redact_list_item(nested, path) for nested in item]
+    return item
+
+
+def _selection_args(config: GymRuntimeConfig, work_dir: Path) -> list[str]:
+    """The environment/agent/model selection passed to Gym.
+
+    Built once and handed verbatim to both ``gym env validate`` and ``gym env start``, so what is
+    validated is exactly what runs — a pre-flight against a different config would be worse than
+    none.
+    """
+    selection = [
+        "--config",
+        config.agent_config,
+        "--model-type",
+        config.model_type,
+        "--resources-server",
+        config.resources_server,
+    ]
+    if config.bind_resources_server:
+        # Composable (Pattern-A) agents leave resources_server.name unbound ('???'); bind it to the
+        # env we're running. Assumes the agent config's top-level key equals the agent name (the
+        # simple_agent convention) *and* that the resources-server is registered under the
+        # environment's own name — not universally true, so self-contained or differently-named
+        # servers set bind_resources_server=False and bind themselves via env_overrides.
+        selection.append(
+            f"+{config.agent}.responses_api_agents.{config.agent}.resources_server.name={config.resources_server}"
+        )
+    selection.extend(_flatten_overrides(config.env_overrides))
+    # Gym is a Hydra app, so each invocation writes a timestamped run directory — by default
+    # `outputs/<date>/<time>/` under the *current* directory. Since the subprocesses inherit this
+    # process's cwd (so Gym can find env.yaml), the default would litter whatever directory the
+    # caller happened to run from. Redirect it under the run's work dir, with the rest of the run's
+    # artifacts. Applies to every Gym entry point, `gym list` included.
+    selection.append(f"hydra.run.dir={work_dir / _HYDRA_SUBDIR}")
+    return selection
+
+
 def _canonical_row_hash(row: Mapping[str, Any]) -> str:
     """Stable ``sha256`` of a Gym dataset row, excluding runtime-injected fields.
 
@@ -242,10 +394,13 @@ class GymRuntimeConfig(BaseModel):
         "(the composable/Pattern-A agent case, e.g. simple_agent whose config leaves it '???'). Set False for "
         "self-contained agents that already bind their own resources-server.",
     )
-    env_overrides: list[str] = Field(
-        default_factory=list,
-        description="Extra Hydra '+key=value' overrides for `gym env start` (escape hatch; applied after the "
-        "auto-derived resources-server binding).",
+    env_overrides: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Nested config overrides merged into Gym's config, applied after the auto-derived "
+        "resources-server binding. Structured rather than pre-serialized Hydra strings so the config "
+        "travels as JSON — `{'a': {'b': 1}}` becomes `++a.b=1` at invocation. This is the escape "
+        "hatch for what Gym does not standardize: an environment whose resources-server is registered "
+        "under a different name, or which references a model server no shipped config defines.",
     )
     num_repeats: int = Field(default=1, ge=1, description="Attempts per row; each attempt becomes one trial.")
     concurrency: int = Field(
@@ -273,18 +428,22 @@ class GymRuntimeConfig(BaseModel):
 def _gym_executable() -> str:
     """Locate the ``gym`` CLI on PATH, or fail saying what to do about it.
 
-    Gym is expected to be installed in the same environment as this SDK — there is deliberately no
-    config field pointing at a checkout or another venv, because these runner configs become
-    serialized job specs and a local filesystem path cannot cross that boundary. Resolving here
-    rather than at spawn time turns a missing Gym into one legible error instead of an ``ENOENT``
-    out of ``create_subprocess_exec`` after the run has already started.
+    Deliberately PATH-only, with no config field pointing at a checkout or a particular venv: these
+    runner configs become serialized job specs, and a local filesystem path cannot cross that
+    boundary. Resolving here rather than at spawn time turns a missing Gym into one legible error
+    instead of an ``ENOENT`` out of ``create_subprocess_exec`` after the run has already started.
+
+    Note that Gym generally cannot live in this SDK's own environment: it imports Ray at module load,
+    and nemo-platform excludes Ray by constraint over an unfixed CVE. Install Gym separately and put
+    its ``bin`` on PATH; in a job image, the image owns PATH and this resolves normally.
     """
     resolved = shutil.which(_GYM_CLI)
     if resolved is None:
         raise RuntimeError(
-            f"The {_GYM_CLI!r} CLI was not found on PATH. NeMo Gym must be installed in the same Python "
-            "environment as this SDK: `pip install nemo-gym`, plus the target environment's own "
-            "dependencies (each resources-server ships a requirements.txt)."
+            f"The {_GYM_CLI!r} CLI was not found on PATH. Install NeMo Gym in its own environment "
+            "(it needs Ray, which nemo-platform excludes over an unfixed CVE, so it cannot share this "
+            "one) and put that environment's `bin` directory on PATH. Each resources-server also "
+            "ships its own requirements.txt, installed from that server's directory."
         )
     return resolved
 
@@ -471,6 +630,54 @@ async def run_tasks(
         _require_full_coverage(tasks, covered_task_ids={trial.task_id for trial in trials}, rollouts_path=rollouts_path)
         return trials
 
+    async def _validate_config(
+        self, gym: str, selection: Sequence[str], subprocess_env: Mapping[str, str], work_dir: Path
+    ) -> None:
+        """Pre-flight the composed Gym config with ``gym env validate`` before starting anything.
+
+        Gym does not publish what configuration an environment requires — the typed
+        ``*ResourcesServerConfig`` classes cover behavioural knobs, while the model wiring lives in
+        each environment's YAML under names that vary per environment. ``gym env validate`` is the
+        only way to find out short of running: it merges configs, flags, and overrides, then reports
+        unresolved ``???`` values, bad paths, and cross-references to servers that are not defined —
+        without Ray and without starting a server.
+
+        Running it every time is worth the second it costs. A config mistake otherwise surfaces after
+        ``gym env start`` has brought up a Ray cluster and several uvicorn servers, as a readiness
+        timeout up to ``startup_timeout_s`` (240s by default) whose message says nothing about the
+        actual problem.
+        """
+        proc = await asyncio.create_subprocess_exec(
+            gym,
+            "env",
+            "validate",
+            *selection,
+            env=dict(subprocess_env),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.STDOUT,
+            # Its own process group, like the other two Gym invocations. `_terminate` signals the
+            # group via `killpg`, so without this a validate timeout would signal *our* group — the
+            # SDK process included.
+            start_new_session=True,
+        )
+        try:
+            stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=_VALIDATE_TIMEOUT_S)
+        except TimeoutError:
+            await _terminate(proc, grace_s=self._config.shutdown_grace_s)
+            raise RuntimeError(
+                f"`gym env validate` did not finish within {_VALIDATE_TIMEOUT_S}s for resources-server "
+                f"{self._config.resources_server!r}"
+            ) from None
+
+        report = stdout.decode("utf-8", errors="replace").strip()
+        (work_dir / "gym_validate.log").write_text(report, encoding="utf-8")
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"Gym rejected the composed config for resources-server {self._config.resources_server!r} "
+                f"before any server started. `gym env validate` said:\n\n{report}"
+            )
+        logger.debug("gym env validate: %s", report)
+
     async def _run_two_step(self, input_path: Path, output_path: Path, work_dir: Path) -> None:
         """Start the Gym servers, collect against them with ``--no-serve``, then tear them down."""
         cfg = self._config
@@ -483,25 +690,11 @@ async def _run_two_step(self, input_path: Path, output_path: Path, work_dir: Pat
         # hook is wrong for Gym (servers manage their own deps), so disable it for the subprocesses.
         subprocess_env = {**os.environ, "RAY_ENABLE_UV_RUN_RUNTIME_ENV": "0"}
 
-        env_cmd = [
-            gym,
-            "env",
-            "start",
-            "--config",
-            cfg.agent_config,
-            "--model-type",
-            cfg.model_type,
-            "--resources-server",
-            cfg.resources_server,
-        ]
-        if cfg.bind_resources_server:
-            # Composable (Pattern-A) agents leave resources_server.name unbound ('???'); bind it to the
-            # env we're running. Assumes the agent config's top-level key equals the agent name (the
-            # simple_agent convention). Self-contained agents set bind_resources_server=False.
-            env_cmd.append(
-                f"+{cfg.agent}.responses_api_agents.{cfg.agent}.resources_server.name={cfg.resources_server}"
-            )
-        env_cmd.extend(cfg.env_overrides)
+        selection = _selection_args(cfg, work_dir)
+
+        await self._validate_config(gym, selection, subprocess_env, work_dir)
+
+        env_cmd = [gym, "env", "start", *selection]
         # start_new_session=True puts `gym env start` in its own process group so teardown can signal
         # the *whole* Ray-cluster + uvicorn tree, not just the direct child (else they orphan).
         # stderr is merged into stdout here (unlike `eval run`, which splits them) because readiness