feat(eval): --backend=gym pilot for ns eval (gsm8k)

nemo_skills/pipeline/adapters/__init__.py

@@ -0,0 +1,14 @@
+# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Pipeline-level adapters bridging external backends to NeMo-Skills' on-disk schemas."""

nemo_skills/pipeline/adapters/gym_to_skills.py

@@ -0,0 +1,184 @@
+# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Convert Gym `ng_collect_rollouts` output into a Skills-shape `output.jsonl`.
+
+The existing `nemo_skills.pipeline.summarize_results` and the
+`nemo_skills.evaluation.metrics.MathMetrics` calculator read fields like
+`symbolic_correct`, `predicted_answer`, `expected_answer`, and `generation`
+from Skills' `output.jsonl`. Rather than teach `summarize_results` about
+Gym's `rollouts.jsonl` schema, we write a parallel Skills-shape file
+alongside each Gym rollouts file so the downstream metric path is unchanged.
+
+For the v1 pilot only `math` is supported (the `math_with_judge` resource
+server). Other `metric_type`s raise — when we expand benchmarks in Tier 2/3
+we add per-type converters here.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from pathlib import Path
+from typing import Any, Dict, Iterable, List
+
+LOG = logging.getLogger(__name__)
+
+
+# ----------------------------------------------------------------------
+# math: math_with_judge → Skills MathMetrics
+# ----------------------------------------------------------------------
+
+
+def _extract_output_text(response: Dict[str, Any]) -> str:
+    """Pull the assistant-message text out of an OpenAI Responses-API response."""
+    chunks: List[str] = []
+    for item in response.get("output", []) or []:
+        for c in item.get("content", []) or []:
+            if c.get("type") == "output_text":
+                text = c.get("text")
+                if text:
+                    chunks.append(text)
+    return "".join(chunks)
+
+
+def convert_math_rollout(rollout: Dict[str, Any]) -> Dict[str, Any]:
+    """Convert one Gym `math_with_judge` rollout dict into a Skills math prediction.
+
+    Maps the fields that `MathMetrics` reads:
+        symbolic_correct ← (library_reward == 1.0)
+        predicted_answer ← extracted_answer
+        expected_answer  ← expected_answer
+        generation       ← response.output[].content[].text (concatenated)
+
+    If a judge produced evaluations, surface them as `judgement` so Skills'
+    `is_correct_judgement` can parse them. v1 pilot benchmarks (gsm8k) don't
+    use the judge, so this is a best-effort placeholder.
+    """
+    library_reward = rollout.get("library_reward")
+    if library_reward is None:
+        # Fall back to the overall reward when the server doesn't expose the
+        # symbolic-only signal (we lose the judge/symbolic separation but at
+        # least we get a pass/fail).
+        library_reward = rollout.get("reward", 0.0)
+
+    out: Dict[str, Any] = {
+        "generation": _extract_output_text(rollout.get("response", {}) or {}),
+        "predicted_answer": rollout.get("extracted_answer"),
+        "expected_answer": rollout.get("expected_answer"),
+        "symbolic_correct": bool(library_reward == 1.0),
+    }
+
+    judge = rollout.get("judge_evaluations")
+    if judge:
+        # MathMetrics calls is_correct_judgement(prediction["judgement"]) which
+        # expects a string. The judge_evaluations entries each have their own
+        # `response.output[].content[].text`; concatenate the first verdict
+        # text for now and revisit when Tier 2 brings judge benchmarks in.
+        verdicts = []
+        for ev in judge if isinstance(judge, list) else [judge]:
+            response = ev.get("response", {}) if isinstance(ev, dict) else {}
+            text = _extract_output_text(response)
+            if text:
+                verdicts.append(text)
+        if verdicts:
+            out["judgement"] = "\n\n".join(verdicts)
+
+    return out
+
+
+# ----------------------------------------------------------------------
+# Dispatch + file IO
+# ----------------------------------------------------------------------
+
+
+_CONVERTERS = {
+    "math": convert_math_rollout,
+}
+
+
+def supported_metric_types() -> List[str]:
+    return sorted(_CONVERTERS.keys())
+
+
+def convert_rollouts(
+    rollouts: Iterable[Dict[str, Any]],
+    *,
+    metric_type: str,
+) -> Iterable[Dict[str, Any]]:
+    try:
+        converter = _CONVERTERS[metric_type]
+    except KeyError as e:
+        raise NotImplementedError(
+            f"gym_to_skills: no converter for metric_type={metric_type!r}. "
+            f"Supported: {supported_metric_types()}. "
+            f"Add a converter to nemo_skills/pipeline/adapters/gym_to_skills.py."
+        ) from e
+    for rollout in rollouts:
+        yield converter(rollout)
+
+
+def convert_file(
+    rollouts_path: str | Path,
+    output_path: str | Path,
+    *,
+    metric_type: str,
+) -> int:
+    """Read `rollouts_path`, write Skills-shape JSONL to `output_path`. Returns row count."""
+    rollouts_path = Path(rollouts_path)
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    n = 0
+    with rollouts_path.open("r", encoding="utf-8") as fin, output_path.open("w", encoding="utf-8") as fout:
+        rollouts = (json.loads(line) for line in fin if line.strip())
+        for prediction in convert_rollouts(rollouts, metric_type=metric_type):
+            fout.write(json.dumps(prediction))
+            fout.write("\n")
+            n += 1
+    LOG.info("Wrote %d converted rows from %s to %s", n, rollouts_path, output_path)
+    return n
+
+
+# ----------------------------------------------------------------------
+# CLI
+# ----------------------------------------------------------------------
+
+
+def _build_argparser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="python -m nemo_skills.pipeline.adapters.gym_to_skills",
+        description="Convert Gym rollouts.jsonl into Skills-shape output.jsonl.",
+    )
+    p.add_argument("rollouts", help="Path to the Gym rollouts JSONL file.")
+    p.add_argument("output", help="Path to write the Skills-shape JSONL file.")
+    p.add_argument(
+        "--metric_type",
+        required=True,
+        choices=supported_metric_types(),
+        help="Which Skills metric calculator the output will feed.",
+    )
+    return p
+
+
+def main(argv: List[str] | None = None) -> int:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+    args = _build_argparser().parse_args(argv)
+    convert_file(args.rollouts, args.output, metric_type=args.metric_type)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

nemo_skills/pipeline/eval.py

@@ -31,7 +31,18 @@
     EvalGenerationUnit,
     prepare_eval_commands,
 )
-from nemo_skills.pipeline.utils.scripts import EvalClientScript, SandboxScript, ServerScript
+from nemo_skills.pipeline.utils.gym import (
+    GymBenchmarkConfig,
+    get_gym_config,
+    is_registered,
+    registered_benchmarks,
+)
+from nemo_skills.pipeline.utils.scripts import (
+    EvalClientScript,
+    GymEvalClientScript,
+    SandboxScript,
+    ServerScript,
+)
 from nemo_skills.utils import (
     get_logger_name,
     setup_logging,
@@ -46,6 +57,11 @@ class SingleNodeMode(str, enum.Enum):
     parallel = "parallel"
 
 
+class EvalBackend(str, enum.Enum):
+    skills = "skills"
+    gym = "gym"
+
+
 def _resolve_child_sbatch_kwargs(sbatch_kwargs, child_sbatch_kwargs):
     if child_sbatch_kwargs is None:
         return sbatch_kwargs
@@ -272,6 +288,12 @@ def eval(
         "If running in parallel, ++max_concurrent_requests parameter is respected per "
         "benchmark, but not globally across benchmarks.",
     ),
+    backend: EvalBackend = typer.Option(
+        EvalBackend.skills,
+        help="Which generation backend to use. 'skills' is the legacy nemo_skills.inference.generate "
+        "path; 'gym' runs through NeMo-Gym (ng_run + ng_collect_rollouts) and is gated on the "
+        "benchmark having a Gym counterpart. See convert-eval-to-gym/TIERED_PLAN.md.",
+    ),
     run_after: List[str] = typer.Option(
         None, help="Can specify a list of expnames that need to be completed before this one starts"
     ),
@@ -385,6 +407,25 @@ def convert_server_type_to_string(st):
     except AttributeError:
         pass
 
+    try:
+        backend = backend.value
+    except AttributeError:
+        pass
+
+    if backend == EvalBackend.gym.value:
+        # Validate every requested benchmark has a Gym wiring. Strip the optional
+        # `:N` (num-samples) suffix before lookup. Fail early with a clear pointer
+        # to the Skills fallback for benchmarks not yet ported.
+        requested_names = [b.split(":", 1)[0] for b in benchmarks.split(",")]
+        missing = [name for name in requested_names if not is_registered(name)]
+        if missing:
+            raise ValueError(
+                f"--backend=gym does not yet support benchmark(s): {missing}. "
+                f"Registered benchmarks: {registered_benchmarks()}. "
+                f"Use --backend=skills for unported benchmarks, or extend "
+                f"nemo_skills/pipeline/utils/gym/registry.py."
+            )
+
     if log_samples:
         wandb_parameters = {
             "name": wandb_name or expname,
@@ -575,17 +616,50 @@ def convert_server_type_to_string(st):
                 else:
                     unit_dicts.append(dict(u))
 
-            client_script = EvalClientScript(
-                units=unit_dicts,
-                single_node_mode=single_node_mode,
-                with_sandbox=sandbox_enabled,
-                servers=server_scripts,
-                server_addresses_prehosted=server_addresses_list,
-                model_names=models_list,
-                server_types=server_types_list,
-                sandbox=sandbox_script,
-                installation_command=installation_command,
-            )
+            if backend == EvalBackend.gym.value:
+                # Gym runs one shared `ng_run` mesh per SLURM job, so all units
+                # in a job must share the same Gym wiring. Enforce one benchmark
+                # per job rather than per-unit mesh-switching.
+                if len(job_benchmarks) > 1:
+                    raise ValueError(
+                        f"--backend=gym requires one benchmark per SLURM job; this job "
+                        f"would mix {sorted(job_benchmarks)}. Re-run with --num_jobs=-1 "
+                        f"(or large enough that each benchmark gets its own job)."
+                    )
+                (gym_benchmark,) = job_benchmarks
+                gym_cfg: GymBenchmarkConfig = get_gym_config(gym_benchmark)
+                # Effective metric_type for the gym→skills converter: prefer
+                # the CLI override, else the benchmark's METRICS_TYPE module
+                # constant. Same precedence summarize_results uses.
+                gym_metric_type = metric_type or benchmarks_dict[gym_benchmark].metrics_type
+                client_script = GymEvalClientScript(
+                    units=unit_dicts,
+                    config_paths=list(gym_cfg.config_paths),
+                    agent_name=gym_cfg.agent_name,
+                    gym_input_jsonl_fpath=gym_cfg.input_jsonl_fpath,
+                    gym_prompt_config=gym_cfg.prompt_config,
+                    metric_type=gym_metric_type,
+                    single_node_mode=single_node_mode,
+                    with_sandbox=sandbox_enabled,
+                    servers=server_scripts,
+                    server_addresses_prehosted=server_addresses_list,
+                    model_names=models_list,
+                    server_types=server_types_list,
+                    sandbox=sandbox_script,
+                    installation_command=installation_command,
+                )
+            else:
+                client_script = EvalClientScript(
+                    units=unit_dicts,
+                    single_node_mode=single_node_mode,
+                    with_sandbox=sandbox_enabled,
+                    servers=server_scripts,
+                    server_addresses_prehosted=server_addresses_list,
+                    model_names=models_list,
+                    server_types=server_types_list,
+                    sandbox=sandbox_script,
+                    installation_command=installation_command,
+                )
 
             # Build groups: group0 = (optional server0) + (optional sandbox) + client
             groups = []
@@ -618,10 +692,23 @@ def convert_server_type_to_string(st):
                     )
                 )
 
+            # Pick the right container for the client task. The Skills client
+            # uses `nemo-skills`; the Gym client needs `ng_run`/ng_collect_rollouts`
+            # which live in the gym container (falls back to nemo-rl per
+            # nemo_gym_rollouts' convention when `nemo-gym` is absent).
+            if backend == EvalBackend.gym.value:
+                client_container = (
+                    main_container
+                    or cluster_config["containers"].get("nemo-gym")
+                    or cluster_config["containers"]["nemo-rl"]
+                )
+            else:
+                client_container = main_container or cluster_config["containers"]["nemo-skills"]
+
             group0_components.append(
                 Command(
                     script=client_script,
-                    container=main_container or cluster_config["containers"]["nemo-skills"],
+                    container=client_container,
                     name=f"{task_name}",
                 )
             )

nemo_skills/pipeline/utils/gym/__init__.py

@@ -0,0 +1,34 @@
+# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Pipeline utilities for the Gym eval/generate backend."""
+
+from nemo_skills.pipeline.utils.gym.registry import (
+    GymBenchmarkConfig,
+    get_gym_config,
+    is_registered,
+    registered_benchmarks,
+)
+from nemo_skills.pipeline.utils.gym.translator import (
+    UnsupportedSkillsOverrideError,
+    translate_skills_overrides_to_gym,
+)
+
+__all__ = [
+    "GymBenchmarkConfig",
+    "UnsupportedSkillsOverrideError",
+    "get_gym_config",
+    "is_registered",
+    "registered_benchmarks",
+    "translate_skills_overrides_to_gym",
+]