diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/README.md b/plugins/nemo-evaluator/scripts/gym-custom-environment/README.md new file mode 100644 index 0000000000..cb58574e22 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/README.md @@ -0,0 +1,299 @@ + + + +# Custom Gym environment workflow + +This directory contains a repeatable workflow for validating custom `wheels-v1` Gym environments +with NeMo Evaluator and OpenSandbox. The workflow prepares or accepts an environment, uploads it, +runs an evaluation, verifies the result, records evidence, and removes temporary Platform +resources. With no input flags it runs the bundled ASCII Tree example; developers can instead +provide any complete compatible environment and Gym dataset. + +## What makes the environment custom? + +The published `nmp-gym-host` image provides Gym and a standard `simple_agent`. It does not contain +the evaluation-specific code used here. + +The default example builds and uploads a `wheels-v1` environment FileSet containing: + +- a Python wheel that implements an ASCII Tree reward function; and +- a Gym resources server that extracts the model response and invokes that function. + +Evaluator installs the wheel and launches the resources server inside a fresh OpenSandbox sandbox. +The workflow confirms that the uploaded resources server produced the persisted reward, which +demonstrates that Evaluator delivered and executed the selected custom environment. + +The example data and scoring semantics come from `primeintellect/ascii-tree==0.1.5`. Prime Intellect +is an implementation detail of this fixture, not the workflow's interface. Its converter normally +produces an `adapter-wheels-v1` package for the unsupported `verifiers` runtime. This utility adapts +that output to Evaluator's supported `wheels-v1` format. + +## Project layout + +The workflow is organized into the following components: + +- `run.py` is the command-line entry point. +- `workflow.py` coordinates generic preparation, upload, inference, evaluation, verification, and + cleanup. +- `prerequisites.py` performs read-only workstation, cluster, configuration, and image checks. +- `prepare.py` validates and stages complete environment packages and datasets. +- `ascii_tree_example.py` converts the default fixture, builds its scorer wheel, and adapts its + dataset. +- `submit.py` constructs the Evaluator job, waits for it, and validates downloaded results. +- `config.py` derives settings, temporary resource names, and evidence paths. +- `commands.py`, `artifacts.py`, and `console.py` contain shared helpers. +- `environment/` and `scorer/` contain the bundled example package and wheel source. +- `helm/` contains the values example for enabling the required sandboxed Gym configuration. + +## Setup + +### Prerequisites + +The workflow expects an existing, correctly configured environment. Its prerequisite checks are +read-only; it does not install or upgrade Kubernetes components. + +Required infrastructure and access: + +- a non-production Kubernetes cluster with NeMo Platform installed; +- OpenSandbox installed and reachable from the Platform namespace; +- sandboxed Gym enabled in the Platform ConfigMap; +- an empty or absent `evaluator.sandbox_runtime_image`; +- the OpenSandbox API-key Secret in the Platform namespace; +- a registry pull Secret for the configured Platform images; +- a `ReadWriteMany` Platform files PVC; +- matching published `nmp-api`, `nmp-cpu-tasks`, and `nmp-gym-host` images at an immutable tag; +- a local checkout compatible with the deployed images; +- `uv`, Docker Buildx, `kubectl`, and Helm; +- workstation access to NGC and, for the default example, Prime Intellect Hub, GitHub, and Hugging + Face; and +- an NVIDIA API key authorized for the selected Inference Hub model. + +The `nmp-temp1` deployment in `nemo-dev-blue` already satisfies these prerequisites. +Before running the workflow there, follow the **Configure sandboxed Gym** steps to apply the +settings required by this test. + +The defaults are: + +- Helm release `nemo-platform` in namespace `nmp-temp1`; +- OpenSandbox service `opensandbox-server-crun` in namespace `opensandbox-system`; +- Secrets `opensandbox-server-crun-api-key` and `nvcrimagepullsecret`; +- PVC `nemo-platform-core-storage`; +- workspace `default` +- model entity `default/nvidia-meta-llama-3-3-70b-instruct`. + +### Configure sandboxed Gym + +Apply the example values to an existing Platform release before running the workflow. The +`--reuse-values` flag preserves the release's current image coordinates and unrelated settings. + +```bash +export KUBECONFIG="${HOME}/teleport-kubeconfig.yaml" +export NMP_GYM_CUSTOM_NAMESPACE="${NMP_GYM_CUSTOM_NAMESPACE:-nmp-temp1}" +export NMP_GYM_CUSTOM_RELEASE="${NMP_GYM_CUSTOM_RELEASE:-nemo-platform}" +export GYM_VALUES=/tmp/gym-custom-environment-values.yaml + +cp \ + plugins/nemo-evaluator/scripts/gym-custom-environment/helm/opensandbox-values.example.yaml \ + "$GYM_VALUES" + +# Set the namespace used by the internal Platform API URL. +export INTERNAL_PLATFORM_API_URL="http://nemo-platform-api.${NMP_GYM_CUSTOM_NAMESPACE}.svc.cluster.local:8080" +yq -i ' + .platformConfig.evaluator.sandbox_policy_base_urls = [strenv(INTERNAL_PLATFORM_API_URL)] +' "$GYM_VALUES" + +# Render against the live cluster first. Review this output locally; existing +# ConfigMap values can include sensitive configuration. +helm upgrade "$NMP_GYM_CUSTOM_RELEASE" k8s/helm \ + -n "$NMP_GYM_CUSTOM_NAMESPACE" \ + --reuse-values \ + -f "$GYM_VALUES" \ + --dry-run=server \ + --hide-secret + +helm upgrade "$NMP_GYM_CUSTOM_RELEASE" k8s/helm \ + -n "$NMP_GYM_CUSTOM_NAMESPACE" \ + --reuse-values \ + -f "$GYM_VALUES" \ + --timeout 20m + +# The Platform config is mounted with subPath, so restart consumers after the +# ConfigMap changes. +kubectl rollout restart deployment \ + -l "app.kubernetes.io/instance=${NMP_GYM_CUSTOM_RELEASE}" \ + -n "$NMP_GYM_CUSTOM_NAMESPACE" + +kubectl rollout status deployment \ + -l "app.kubernetes.io/instance=${NMP_GYM_CUSTOM_RELEASE}" \ + -n "$NMP_GYM_CUSTOM_NAMESPACE" \ + --timeout 20m +``` + +Confirm that both Platform and Evaluator sandboxing are enabled and that +`sandbox_runtime_image` is empty: + +```bash +kubectl get configmap nemo-platform-config \ + -n "$NMP_GYM_CUSTOM_NAMESPACE" \ + -o 'jsonpath={.data.config\.yaml}' | + yq '.platform, .evaluator' +``` + +Optionally, override only values that differ: + +```bash +export NMP_GYM_CUSTOM_NAMESPACE="" +export NMP_GYM_CUSTOM_RELEASE="" +export NMP_GYM_CUSTOM_WORKSPACE="" +export NMP_GYM_CUSTOM_JOB_PVC="" +export NMP_GYM_CUSTOM_REGISTRY_SECRET="" +export NMP_GYM_CUSTOM_OPENSANDBOX_NAMESPACE="" +export NMP_GYM_CUSTOM_OPENSANDBOX_SERVICE="" +export NMP_GYM_CUSTOM_OPENSANDBOX_API_SECRET="" +export NMP_GYM_CUSTOM_PLATFORM_API_SERVICE="" +export NMP_GYM_CUSTOM_MODEL_ENTITY_ID="/" +export NMP_GYM_CUSTOM_LOCAL_PORT=18080 +``` + +If Docker is not authenticated to NGC, log in without putting the key in shell history: + +```bash +test -n "${NGC_API_KEY:-}" || { + printf 'NGC API key: ' >&2 + IFS= read -rs NGC_API_KEY + printf '\n' >&2 + export NGC_API_KEY +} + +printf '%s' "$NGC_API_KEY" | + docker login nvcr.io --username '$oauthtoken' --password-stdin +``` + +### Connect to the Platform API + +The workflow accesses the Platform API through a local Kubernetes port-forward. It checks +`127.0.0.1:18080` first and reuses an existing healthy connection. If none exists, the workflow +starts a temporary port-forward automatically and closes it when the run finishes. + +To manage the connection yourself, run the following in a separate terminal and leave it running +for the duration of the workflow: + +```bash +export KUBECONFIG="${HOME}/teleport-kubeconfig.yaml" +export NMP_GYM_CUSTOM_NAMESPACE="${NMP_GYM_CUSTOM_NAMESPACE:-nmp-temp1}" +export NMP_GYM_CUSTOM_PLATFORM_API_SERVICE="${NMP_GYM_CUSTOM_PLATFORM_API_SERVICE:-nemo-platform-api}" +export NMP_GYM_CUSTOM_LOCAL_PORT="${NMP_GYM_CUSTOM_LOCAL_PORT:-18080}" + +kubectl port-forward \ + -n "$NMP_GYM_CUSTOM_NAMESPACE" \ + "service/$NMP_GYM_CUSTOM_PLATFORM_API_SERVICE" \ + "$NMP_GYM_CUSTOM_LOCAL_PORT:8080" +``` + +## Run + +### Bundled example + +From the repository root, run the workflow without input flags to use the bundled ASCII Tree +example: + +```bash +export KUBECONFIG="${HOME}/teleport-kubeconfig.yaml" + +uv run --isolated --frozen \ + --package nemo-evaluator-plugin \ + --with-editable packages/filesets \ + --with-editable packages/models \ + python \ + plugins/nemo-evaluator/scripts/gym-custom-environment/run.py +``` + +The script securely prompts for the NVIDIA API key. For non-interactive use, export +`INFERENCE_NVIDIA_API_KEY` before running it. + +The command runs the workflow in an isolated, package-scoped uv environment and does not modify +the repository `.venv`. Within that environment, the pinned Prime Intellect converter runs with +only its `conversion` extra. The workflow retains and caches only the two-row JSONL fixture, +discarding the converter's adapter package and dataset snapshot. The custom scorer wheel is also +cached by a digest of its source. Subsequent runs reuse both artifacts until their inputs change. + +### A different custom environment + +Supply a complete environment directory and its separate Gym JSONL dataset: + +```bash +uv run --isolated --frozen \ + --package nemo-evaluator-plugin \ + --with-editable packages/filesets \ + --with-editable packages/models \ + python \ + plugins/nemo-evaluator/scripts/gym-custom-environment/run.py \ + --environment-dir path/to/environment \ + --dataset path/to/dataset.jsonl +``` + +The environment directory must: + +- contain a valid `nemo-environment.yaml` with `format: wheels-v1`; +- contain every manifest-listed config and a non-empty, flat `wheels/` directory; +- declare at least one resources server; and +- remain separate from the JSONL dataset. + +Every dataset row must satisfy Gym's dataset contract, including a +`responses_create_params` mapping. The workflow copies both inputs into its run directory and +does not modify the originals. If the package declares multiple resources servers, select one: + +```bash +--resources-server +``` + +The evaluation uses the Gym host's standard `simple_agent`; custom packages provide the +resources server and its wheel-delivered dependencies. + +The terminal shows seven numbered stages: + +1. check workstation and cluster prerequisites; +2. prepare the custom environment; +3. create temporary Platform resources; +4. smoke-test the selected model; +5. run the custom Gym evaluation; +6. verify evaluation evidence; and +7. clean up temporary resources. + +A successful run ends with: + +```text +SUCCESS: Custom Gym environment evaluation passed +``` + +The model does not need a perfect reward. The workflow passes when the FileSet-provided resources +server returns a finite reward, persisted trial and metric values agree, image and OpenSandbox +lifecycle checks pass, and there are no failed or missing results. + +## Evidence + +Each invocation creates a unique `/tmp/nmp-gym-custom-environment-*` directory. Pass +`--run-dir ` to choose a different location. Important evidence includes: + +- `evidence/run-summary.json`; +- `evidence/platform-config.json`; +- `evidence/job-status.json` and `evidence/job-logs.json`; +- `evidence/verification.json`; +- `evidence/agent-eval-results/summary.json`, `scores.jsonl`, and `trials.jsonl`; +- `evidence/nmp-api-image.txt`; +- `evidence/nmp-cpu-tasks-image.txt`; and +- `evidence/nmp-gym-host-image.txt`. + +Evidence contains prompts and model responses but never the NVIDIA API key. + +## Troubleshooting + +- **An image cannot be inspected:** authenticate Docker to the configured registry and confirm all + three images exist at the configured tag. +- **A stale Gym host override is reported:** clear `evaluator.sandbox_runtime_image` and restart + API/controller deployments. +- **The PVC check fails:** configure `evaluator.sandbox_job_storage_pvc_claim` with a + `ReadWriteMany` claim shared by task and sandbox pods. +- **The local API port is occupied:** set `NMP_GYM_CUSTOM_LOCAL_PORT` to another port. +- **The model is unavailable:** select an entity served by the temporary Inference Hub provider. +- **The job fails:** inspect `evidence/job-status.json` and `evidence/job-logs.json`. diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/artifacts.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/artifacts.py new file mode 100644 index 0000000000..09a56e64f7 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/artifacts.py @@ -0,0 +1,70 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Internal JSON evidence helpers used by preparation, submission, and verification. + +These functions keep artifact encoding and validation consistent across the +implementation modules. Evidence is written beneath the run directory selected +by ``run.py``; this module is not itself executable. +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, is_dataclass +from datetime import date, datetime +from enum import Enum +from pathlib import Path +from typing import Any + + +def _json_default(value: Any) -> Any: + """Convert common SDK, dataclass, enum, date, and path values to JSON data.""" + model_dump = getattr(value, "model_dump", None) + if callable(model_dump): + return model_dump(mode="json") + if is_dataclass(value) and not isinstance(value, type): + return asdict(value) + if isinstance(value, Enum): + return value.value + if isinstance(value, date | datetime): + return value.isoformat() + if isinstance(value, Path): + return str(value) + raise TypeError(f"cannot serialize {type(value).__name__} as JSON") + + +def write_json(path: Path, value: Any) -> None: + """Serialize Python and SDK values, creating parent directories as needed.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + value, + indent=2, + ensure_ascii=False, + default=_json_default, + ) + + "\n", + encoding="utf-8", + ) + + +def read_json(path: Path) -> dict[str, Any]: + """Read an evidence file and require a JSON object at its root.""" + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ValueError(f"{path} must contain a JSON object") + return value + + +def read_jsonl(path: Path) -> list[dict[str, Any]]: + """Read non-empty JSON Lines records and require each record to be an object.""" + rows: list[dict[str, Any]] = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): + if not line.strip(): + continue + value = json.loads(line) + if not isinstance(value, dict): + raise ValueError(f"{path}:{line_number} must contain a JSON object") + rows.append(value) + return rows diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/ascii_tree_example.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/ascii_tree_example.py new file mode 100644 index 0000000000..781f93bbbb --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/ascii_tree_example.py @@ -0,0 +1,267 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Prepare the bundled ASCII Tree example consumed by the generic workflow. + +This module contains every Prime Intellect and ASCII Tree implementation detail: +it converts the pinned source fixture, adapts its rows to Gym's image-provided +``simple_agent``, builds and caches the custom scorer wheel, and materializes a +complete ``wheels-v1`` package. The shared preparation, submission, and +verification modules treat its outputs like any caller-provided environment. +""" + +from __future__ import annotations + +import hashlib +import json +import shutil +import tempfile +from pathlib import Path +from typing import Any + +from artifacts import read_jsonl +from commands import CommandRunner + +SCRIPT_DIRECTORY = Path(__file__).resolve().parent +ENVIRONMENT_TEMPLATE = SCRIPT_DIRECTORY / "environment" +WHEEL_SOURCE = SCRIPT_DIRECTORY / "scorer" +WHEEL_CACHE = Path(tempfile.gettempdir()) / "nmp-gym-custom-environment-wheel-cache" +FIXTURE_CACHE = Path(tempfile.gettempdir()) / "nmp-gym-custom-environment-fixture-cache" +SCORER_DISTRIBUTION = "nmp_ascii_tree_evaluator" +PRIME_FIXTURE_HUB_ID = "primeintellect/ascii-tree" +PRIME_FIXTURE_HUB_VERSION = "0.1.5" +PRIME_FIXTURE_SIZE = 2 +PRIME_FIXTURE_SEED = 0 +FIXTURE_CACHE_FORMAT_VERSION = 1 + + +def _fixture_cache_path() -> Path: + """Return the content-addressed cache path for the pinned source fixture.""" + cache_inputs = { + "format_version": FIXTURE_CACHE_FORMAT_VERSION, + "hub_id": PRIME_FIXTURE_HUB_ID, + "hub_version": PRIME_FIXTURE_HUB_VERSION, + "size": PRIME_FIXTURE_SIZE, + "seed": PRIME_FIXTURE_SEED, + } + cache_key = hashlib.sha256(json.dumps(cache_inputs, sort_keys=True).encode()).hexdigest() + return FIXTURE_CACHE / cache_key / "training.jsonl" + + +def _valid_fixture_cache(cache_path: Path) -> bool: + """Return whether a cached conversion contains the expected usable rows.""" + if not cache_path.is_file(): + return False + try: + rows = read_jsonl(cache_path) + return len(rows) == PRIME_FIXTURE_SIZE and all( + isinstance(row.get("responses_create_params"), dict) + and isinstance(row.get("answer"), str) + and bool(row["answer"].strip()) + for row in rows + ) + except (OSError, ValueError, json.JSONDecodeError): + return False + + +def _copy_cached_fixture(cache_path: Path, converter_dataset: Path) -> None: + """Copy the reusable JSONL fixture into this run's working directory.""" + if converter_dataset.exists(): + shutil.rmtree(converter_dataset) + converter_dataset.mkdir(parents=True) + shutil.copy2(cache_path, converter_dataset / "training.jsonl") + + +def _convert_fixture( + runner: CommandRunner, + *, + converter_environment: Path, + converter_dataset: Path, +) -> bool: + """Create or reuse the pinned two-row Prime Intellect fixture. + + The converter runs in an isolated uv environment. Only its JSONL output is + cached; generated adapters, snapshots, and package caches are discarded. + + Returns ``True`` when an existing valid conversion was reused. + """ + cache_path = _fixture_cache_path() + if _valid_fixture_cache(cache_path): + _copy_cached_fixture(cache_path, converter_dataset) + return True + + if cache_path.exists(): + cache_path.unlink() + cache_path.parent.mkdir(parents=True, exist_ok=True) + + try: + runner.run( + [ + "uv", + "run", + "--isolated", + "--frozen", + "--package", + "nmp-rl", + "--extra", + "conversion", + "pi-to-gym-conversion", + "--hub-id", + PRIME_FIXTURE_HUB_ID, + "--hub-version", + PRIME_FIXTURE_HUB_VERSION, + "--out-dir", + str(converter_environment), + "--dataset-dir", + str(converter_dataset), + "--dataset-size", + str(PRIME_FIXTURE_SIZE), + "--dataset-seed", + str(PRIME_FIXTURE_SEED), + ] + ) + converted_dataset = converter_dataset / "training.jsonl" + if not _valid_fixture_cache(converted_dataset): + raise RuntimeError(f"converter did not produce {PRIME_FIXTURE_SIZE} valid rows at {converted_dataset}") + + # Replace atomically so an interrupted conversion cannot expose a partial cache. + with tempfile.NamedTemporaryFile(dir=cache_path.parent, delete=False) as temporary_file: + temporary_cache_path = Path(temporary_file.name) + try: + shutil.copy2(converted_dataset, temporary_cache_path) + temporary_cache_path.replace(cache_path) + finally: + temporary_cache_path.unlink(missing_ok=True) + finally: + if converter_environment.exists(): + shutil.rmtree(converter_environment) + if converter_dataset.exists(): + shutil.rmtree(converter_dataset) + + _copy_cached_fixture(cache_path, converter_dataset) + return False + + +def _adapt_row(row: dict[str, Any], *, position: int) -> dict[str, Any]: + """Retarget one source row from Prime Intellect's agent to ``simple_agent``.""" + response_parameters = row.get("responses_create_params") + if not isinstance(response_parameters, dict): + raise ValueError(f"row {position} has no responses_create_params mapping") + prompt = response_parameters.get("input") + if not isinstance(prompt, list) or not prompt: + raise ValueError(f"row {position} has no prompt in responses_create_params.input") + answer = row.get("answer") + if not isinstance(answer, str) or not answer.strip(): + raise ValueError(f"row {position} has no non-empty answer") + + adapted_row = dict(row) + adapted_row["agent_ref"] = { + "type": "responses_api_agents", + "name": "simple_agent", + } + return adapted_row + + +def _adapt_dataset(converter_dataset: Path, dataset_output: Path) -> int: + """Write source rows in the standard Gym JSONL shape and return their count.""" + source_rows = read_jsonl(converter_dataset) + if not source_rows: + raise ValueError(f"{converter_dataset} contains no dataset rows") + adapted_rows = [_adapt_row(row, position=position) for position, row in enumerate(source_rows, start=1)] + dataset_output.parent.mkdir(parents=True, exist_ok=True) + dataset_output.write_text( + "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in adapted_rows), + encoding="utf-8", + ) + return len(adapted_rows) + + +def _wheel_source_digest() -> str: + """Hash scorer source paths and contents for deterministic cache invalidation.""" + digest = hashlib.sha256() + source_files = sorted( + path for path in WHEEL_SOURCE.rglob("*") if path.is_file() and "__pycache__" not in path.parts + ) + for source_file in source_files: + relative_path = source_file.relative_to(WHEEL_SOURCE) + digest.update(relative_path.as_posix().encode()) + digest.update(b"\0") + digest.update(source_file.read_bytes()) + digest.update(b"\0") + return digest.hexdigest() + + +def _cached_scorer_wheel(runner: CommandRunner) -> tuple[Path, bool]: + """Return a source-matched scorer wheel, building it only when absent.""" + cache_directory = WHEEL_CACHE / _wheel_source_digest() + cached_wheels = sorted(cache_directory.glob(f"{SCORER_DISTRIBUTION}-*.whl")) + if len(cached_wheels) == 1: + return cached_wheels[0], True + if cached_wheels: + shutil.rmtree(cache_directory) + + cache_directory.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="nmp-ascii-tree-wheel-") as temporary_directory: + distribution_directory = Path(temporary_directory) + runner.run( + [ + "uv", + "build", + "--wheel", + "--clear", + "--out-dir", + str(distribution_directory), + str(WHEEL_SOURCE), + ] + ) + built_wheels = sorted(distribution_directory.glob(f"{SCORER_DISTRIBUTION}-*.whl")) + if len(built_wheels) != 1: + raise RuntimeError(f"expected one {SCORER_DISTRIBUTION} wheel, found: {built_wheels}") + cached_wheel = cache_directory / built_wheels[0].name + shutil.copy2(built_wheels[0], cached_wheel) + return cached_wheel, False + + +def _materialize_environment(runner: CommandRunner, environment_output: Path) -> tuple[Path, bool]: + """Copy the example package and place its cached scorer wheel in the wheelhouse.""" + if environment_output.exists(): + shutil.rmtree(environment_output) + shutil.copytree(ENVIRONMENT_TEMPLATE, environment_output) + + scorer_wheel, cache_hit = _cached_scorer_wheel(runner) + wheels_directory = environment_output / "wheels" + wheels_directory.mkdir(parents=True, exist_ok=True) + for existing_path in wheels_directory.iterdir(): + if existing_path.is_dir(): + shutil.rmtree(existing_path) + else: + existing_path.unlink() + wheel_destination = wheels_directory / scorer_wheel.name + shutil.copy2(scorer_wheel, wheel_destination) + return wheel_destination, cache_hit + + +def prepare_example( + runner: CommandRunner, + *, + converter_environment: Path, + converter_dataset: Path, + environment_output: Path, + dataset_output: Path, +) -> dict[str, Any]: + """Build the default complete environment and dataset for generic validation.""" + fixture_cache_hit = _convert_fixture( + runner, + converter_environment=converter_environment, + converter_dataset=converter_dataset, + ) + scorer_wheel, wheel_cache_hit = _materialize_environment(runner, environment_output) + row_count = _adapt_dataset(converter_dataset / "training.jsonl", dataset_output) + return { + "input_mode": "default-ascii-tree", + "input_label": "bundled ASCII Tree example", + "rows": row_count, + "wheels": [scorer_wheel.name], + "fixture_cache_hit": fixture_cache_hit, + "wheel_cache_hit": wheel_cache_hit, + } diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/commands.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/commands.py new file mode 100644 index 0000000000..3b7c860564 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/commands.py @@ -0,0 +1,103 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Internal subprocess wrapper shared by prerequisite and workflow modules. + +The wrapper executes argument vectors without a shell, merges only explicit +environment overrides, converts command failures into readable exceptions, and +optionally records stdout as evidence. Developers do not invoke this module. +""" + +from __future__ import annotations + +import json +import os +import subprocess +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any, TextIO + + +class CommandError(RuntimeError): + """Raised when an external command exits unsuccessfully.""" + + +class CommandRunner: + """Run argument-vector commands without shell interpolation.""" + + def __init__(self, *, working_directory: Path) -> None: + """Configure the working directory shared by child processes.""" + self.working_directory = working_directory + + def run( + self, + arguments: Sequence[str], + *, + environment: Mapping[str, str] | None = None, + input_text: str | None = None, + output_path: Path | None = None, + ) -> str: + """Run a command, optionally supply stdin, and return captured stdout.""" + command_environment = os.environ.copy() + if environment: + command_environment.update(environment) + try: + result = subprocess.run( + list(arguments), + cwd=self.working_directory, + env=command_environment, + input=input_text, + text=True, + capture_output=True, + check=True, + ) + except subprocess.CalledProcessError as error: + detail = error.stderr.strip() or error.stdout.strip() or f"exit code {error.returncode}" + command = " ".join(arguments) + raise CommandError(f"{command} failed: {detail}") from error + + if output_path is not None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(result.stdout, encoding="utf-8") + return result.stdout + + def run_json( + self, + arguments: Sequence[str], + *, + environment: Mapping[str, str] | None = None, + input_text: str | None = None, + output_path: Path | None = None, + ) -> dict[str, Any]: + """Run a command and require its stdout to contain a JSON object.""" + output = self.run( + arguments, + environment=environment, + input_text=input_text, + output_path=output_path, + ) + value = json.loads(output) + if not isinstance(value, dict): + command = " ".join(arguments) + raise CommandError(f"{command} did not return a JSON object") + return value + + def start( + self, + arguments: Sequence[str], + *, + environment: Mapping[str, str] | None = None, + stdout: TextIO | int | None = None, + ) -> subprocess.Popen[str]: + """Start a long-running command and return its process handle.""" + command_environment = os.environ.copy() + if environment: + command_environment.update(environment) + return subprocess.Popen( + list(arguments), + cwd=self.working_directory, + env=command_environment, + text=True, + stdout=stdout, + stderr=subprocess.STDOUT, + ) diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/config.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/config.py new file mode 100644 index 0000000000..61ae04c392 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/config.py @@ -0,0 +1,156 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Configuration shared by the custom Gym environment implementation modules. + +``run.py`` creates one ``Settings`` object from documented environment +variables and passes it into ``workflow.py``. The dataclasses here centralize +cluster names, temporary Platform resource names, and evidence paths so those +values are derived consistently throughout a run. +""" + +from __future__ import annotations + +import os +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path + +ENVIRONMENT_PREFIX = "NMP_GYM_CUSTOM_" + + +@dataclass(frozen=True, slots=True) +class RunPaths: + """Filesystem locations generated during one workflow run.""" + + root: Path + evidence: Path + converter_environment: Path + converter_dataset: Path + environment: Path + dataset: Path + + @classmethod + def create(cls, root: Path) -> RunPaths: + """Derive every generated path from one run directory.""" + return cls( + root=root, + evidence=root / "evidence", + converter_environment=root / "converter-environment", + converter_dataset=root / "converter-dataset", + environment=root / "environment", + dataset=root / "dataset.jsonl", + ) + + +@dataclass(frozen=True, slots=True) +class Settings: + """Workflow inputs, cluster prerequisites, and run-scoped resource names.""" + + repo_root: Path + asset_root: Path + paths: RunPaths + environment_source: Path | None + dataset_source: Path | None + requested_resources_server: str | None + namespace: str + release: str + workspace: str + job_pvc: str + registry_secret: str + opensandbox_namespace: str + opensandbox_service: str + opensandbox_api_secret: str + platform_api_service: str + local_port: int + model_entity_id: str + fileset: str + inference_secret: str + inference_provider: str + + @classmethod + def from_environment( + cls, + *, + run_dir: Path | None = None, + environment_dir: Path | None = None, + dataset: Path | None = None, + resources_server: str | None = None, + environment: dict[str, str] | None = None, + ) -> Settings: + """Load inputs and optional overrides, rejecting incomplete custom input.""" + if (environment_dir is None) != (dataset is None): + raise ValueError("--environment-dir and --dataset must be supplied together") + if resources_server is not None and not resources_server.strip(): + raise ValueError("--resources-server cannot be empty") + + env = dict(os.environ if environment is None else environment) + asset_root = Path(__file__).resolve().parent + repo_root = asset_root.parents[3] + generated_run_id = f"{datetime.now(UTC):%Y%m%d%H%M%S}-{uuid.uuid4().hex[:6]}" + run_id = env.get(f"{ENVIRONMENT_PREFIX}RUN_ID", generated_run_id) + default_run_dir = Path(f"/tmp/nmp-gym-custom-environment-{run_id}") + workspace = env.get(f"{ENVIRONMENT_PREFIX}WORKSPACE", "default") + + return cls( + repo_root=repo_root, + asset_root=asset_root, + paths=RunPaths.create((run_dir or default_run_dir).resolve()), + environment_source=environment_dir.resolve() if environment_dir is not None else None, + dataset_source=dataset.resolve() if dataset is not None else None, + requested_resources_server=resources_server, + namespace=env.get(f"{ENVIRONMENT_PREFIX}NAMESPACE", "nmp-temp1"), + release=env.get(f"{ENVIRONMENT_PREFIX}RELEASE", "nemo-platform"), + workspace=workspace, + job_pvc=env.get(f"{ENVIRONMENT_PREFIX}JOB_PVC", "nemo-platform-core-storage"), + registry_secret=env.get(f"{ENVIRONMENT_PREFIX}REGISTRY_SECRET", "nvcrimagepullsecret"), + opensandbox_namespace=env.get( + f"{ENVIRONMENT_PREFIX}OPENSANDBOX_NAMESPACE", + "opensandbox-system", + ), + opensandbox_service=env.get( + f"{ENVIRONMENT_PREFIX}OPENSANDBOX_SERVICE", + "opensandbox-server-crun", + ), + opensandbox_api_secret=env.get( + f"{ENVIRONMENT_PREFIX}OPENSANDBOX_API_SECRET", + "opensandbox-server-crun-api-key", + ), + platform_api_service=env.get( + f"{ENVIRONMENT_PREFIX}PLATFORM_API_SERVICE", + "nemo-platform-api", + ), + local_port=int(env.get(f"{ENVIRONMENT_PREFIX}LOCAL_PORT", "18080")), + model_entity_id=env.get( + f"{ENVIRONMENT_PREFIX}MODEL_ENTITY_ID", + f"{workspace}/nvidia-meta-llama-3-3-70b-instruct", + ), + fileset=env.get( + f"{ENVIRONMENT_PREFIX}FILESET", + f"gym-custom-environment-{run_id}", + ), + inference_secret=env.get( + f"{ENVIRONMENT_PREFIX}SECRET", + f"gym-custom-inference-key-{run_id}", + ), + inference_provider=env.get( + f"{ENVIRONMENT_PREFIX}PROVIDER", + f"gym-custom-inference-hub-{run_id}", + ), + ) + + @property + def base_url(self) -> str: + """Return the workstation URL for the temporary API port-forward.""" + return f"http://127.0.0.1:{self.local_port}" + + @property + def internal_api(self) -> str: + """Return the Platform API URL reachable from sandbox pods.""" + return f"http://{self.platform_api_service}.{self.namespace}.svc.cluster.local:8080" + + @property + def uses_default_example(self) -> bool: + """Return whether the workflow should prepare its bundled example inputs.""" + return self.environment_source is None diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/console.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/console.py new file mode 100644 index 0000000000..32abfec8a0 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/console.py @@ -0,0 +1,39 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Internal terminal formatting used by the workflow. + +``workflow.py`` uses ``Console`` for numbered steps, aligned details, success +messages, and warnings. Centralizing output here prevents implementation +modules from inventing inconsistent machine-like status strings. +""" + +from __future__ import annotations + +import sys +from typing import TextIO + + +class Console: + """Render concise, human-readable workflow progress.""" + + def __init__(self, *, output: TextIO = sys.stdout, error_output: TextIO = sys.stderr) -> None: + """Create a console that writes normal and warning messages separately.""" + self.output = output + self.error_output = error_output + + def step(self, number: int, total: int, message: str) -> None: + """Print the heading for a workflow step.""" + print(f"\n[{number}/{total}] {message}", file=self.output, flush=True) + + def detail(self, label: str, value: object) -> None: + """Print an indented label and value below the current step.""" + print(f" {label}: {value}", file=self.output, flush=True) + + def success(self, message: str) -> None: + """Print the final success message.""" + print(f"\nSUCCESS: {message}", file=self.output, flush=True) + + def warning(self, message: str) -> None: + """Print a warning without hiding the workflow's primary failure.""" + print(f"\nWARNING: {message}", file=self.error_output, flush=True) diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/nemo-environment.yaml b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/nemo-environment.yaml new file mode 100644 index 0000000000..a44b0635cd --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/nemo-environment.yaml @@ -0,0 +1,11 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +format: wheels-v1 +config_paths: + - resources_servers/ascii_tree/configs/ascii_tree.yaml +metadata: + name: custom-ascii-tree-environment + description: Custom ASCII Tree scorer delivered through an Evaluator FileSet + hub_id: primeintellect/ascii-tree + vf_env_id: ascii-tree diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/app.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/app.py new file mode 100644 index 0000000000..4d549434b6 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/app.py @@ -0,0 +1,95 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""FileSet-delivered Gym resources server for the ASCII Tree example. + +Gym launches this module inside OpenSandbox using ``ascii_tree.yaml``. It +extracts assistant text from each model response, calls the scorer installed +from the adjacent wheel requirement, and returns the reward and parsed tree. +The server is environment code; developers invoke the parent ``run.py`` rather +than running this file locally. +""" + +from __future__ import annotations + +from typing import Any + +from nemo_gym.base_resources_server import ( # ty: ignore[unresolved-import] - supplied by the Gym runtime + BaseResourcesServerConfig, + BaseVerifyRequest, + BaseVerifyResponse, + SimpleResourcesServer, +) +from nmp_ascii_tree_evaluator.scoring import ( # ty: ignore[unresolved-import] - installed from the FileSet wheel + ascii_tree_reward, + extract_ascii_formatted, +) + + +class AsciiTreeResourcesServerConfig(BaseResourcesServerConfig): + """Configuration for the FileSet-provided ASCII Tree resources server.""" + + +class AsciiTreeVerifyRequest(BaseVerifyRequest): + """Expected ASCII tree and model response supplied by Gym.""" + + answer: str + question: str = "" + task: str = "ascii_tree_formatting" + example_id: int | str | None = None + info: dict[str, Any] | None = None + task_idx: int | None = None + vf_env_id: str | None = None + + +class AsciiTreeVerifyResponse(BaseVerifyResponse): + """Custom reward and parsed model output returned to Gym.""" + + answer: str + observed_text: str + parsed_ascii: str | None + + +def assistant_text(response: Any) -> str: + """Collect output text from a Responses API object or equivalent mapping.""" + outputs = response.get("output", []) if isinstance(response, dict) else getattr(response, "output", []) + text_parts: list[str] = [] + for output_item in outputs: + item_type = output_item.get("type") if isinstance(output_item, dict) else getattr(output_item, "type", None) + if item_type != "message": + continue + content_items = ( + output_item.get("content", []) if isinstance(output_item, dict) else getattr(output_item, "content", []) + ) + for content_item in content_items: + content_type = ( + content_item.get("type") if isinstance(content_item, dict) else getattr(content_item, "type", None) + ) + if content_type != "output_text": + continue + text = content_item.get("text", "") if isinstance(content_item, dict) else getattr(content_item, "text", "") + text_parts.append(str(text)) + return "".join(text_parts) + + +class AsciiTreeResourcesServer(SimpleResourcesServer): + """Score model responses with FileSet-provided ASCII Tree code.""" + + config: AsciiTreeResourcesServerConfig + + async def verify( + self, + body: AsciiTreeVerifyRequest, + ) -> AsciiTreeVerifyResponse: + """Extract the assistant response, calculate its reward, and return evidence.""" + observed_text = assistant_text(body.response) + return AsciiTreeVerifyResponse( + **body.model_dump(), + reward=ascii_tree_reward(observed_text, body.answer), + observed_text=observed_text, + parsed_ascii=extract_ascii_formatted(observed_text), + ) + + +if __name__ == "__main__": + AsciiTreeResourcesServer.run_webserver() diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/configs/ascii_tree.yaml b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/configs/ascii_tree.yaml new file mode 100644 index 0000000000..a4d804a886 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/configs/ascii_tree.yaml @@ -0,0 +1,11 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +ascii_tree: + resources_servers: + ascii_tree: + entrypoint: app.py + domain: instruction_following + verified: false + description: Score structured ASCII trees with Prime Intellect ascii-tree 0.1.5 semantics + value: Compare generated ASCII tree lines with the expected tree diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/requirements.txt b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/requirements.txt new file mode 100644 index 0000000000..b58071c237 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/environment/resources_servers/ascii_tree/requirements.txt @@ -0,0 +1,2 @@ +# Installed by Gym from the wheel bundled in this environment FileSet. +nmp-ascii-tree-evaluator==0.1.0 diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/helm/opensandbox-values.example.yaml b/plugins/nemo-evaluator/scripts/gym-custom-environment/helm/opensandbox-values.example.yaml new file mode 100644 index 0000000000..ac337913f7 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/helm/opensandbox-values.example.yaml @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Example values for enabling sandboxed Gym through an existing OpenSandbox +# deployment. Apply this file with `helm upgrade --reuse-values` so unrelated +# settings and deployed image coordinates remain unchanged. +# +# If the Platform namespace is not nmp-temp1, update sandbox_policy_base_urls +# before applying this file. + +sandboxClusterCapable: true + +opensandbox: + domain: opensandbox-server-crun.opensandbox-system.svc.cluster.local + protocol: http + apiKeySecret: opensandbox-server-crun-api-key + apiKeySecretKey: api-key + +platformConfig: + evaluator: + sandboxed_gym_default: true + sandbox_cluster_capable: true + sandbox_host_provider: opensandbox + # Allow enough time for OpenSandbox to pull and start the 6+ GiB Gym host + # image. The SDK's 30-second default might be too short on an uncached node. + sandbox_host_provider_options: + connection: + request_timeout_s: 120 + + # Keep this unset so Evaluator derives nmp-gym-host from the Platform + # registry and immutable release tag. + sandbox_runtime_image: null + + sandbox_job_storage_pvc_claim: nemo-platform-core-storage + sandbox_resources: + cpu: "1" + memory: 2Gi + sandbox_policy_base_urls: + - http://nemo-platform-api.nmp-temp1.svc.cluster.local:8080 + sandbox_egress_allow: + - pypi.org:443 + - files.pythonhosted.org:443 diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/prepare.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/prepare.py new file mode 100644 index 0000000000..485dcb98f4 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/prepare.py @@ -0,0 +1,165 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Validate and stage complete custom Gym environment inputs. + +The shared workflow calls this module after either a developer supplies a +complete ``wheels-v1`` directory and Gym JSONL dataset or the default example +adapter produces those same inputs. This module copies caller-owned inputs into +the run directory, validates the package and dataset without importing customer +code, discovers selectable resources servers, and extracts package metadata +used by generic runtime verification. +""" + +from __future__ import annotations + +import shutil +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +from nemo_evaluator_sdk.agent_eval.runtimes.gym import discover_gym_tasks +from sandboxed_gym.environment_package import ( + WheelsV1Package, + inspect_environment_components, + load_environment_package, +) + + +@dataclass(frozen=True, slots=True) +class PreparedEnvironment: + """Validated environment and dataset metadata consumed by later stages.""" + + root: Path + dataset: Path + name: str + resources_server: str + resources_servers: tuple[str, ...] + wheel_names: tuple[str, ...] + task_count: int + input_mode: str + + def evidence(self) -> dict[str, Any]: + """Return JSON-serializable preparation evidence.""" + return asdict(self) + + +def _require_safe_copy(source: Path, destination: Path, *, label: str) -> None: + """Reject copy layouts that could overwrite or recursively copy an input.""" + resolved_source = source.resolve() + resolved_destination = destination.resolve() + if resolved_source == resolved_destination: + raise ValueError(f"{label} source and run destination must differ: {resolved_source}") + if resolved_destination.is_relative_to(resolved_source): + raise ValueError(f"{label} run destination cannot be inside its source: {resolved_destination}") + if resolved_source.is_relative_to(resolved_destination): + raise ValueError(f"{label} source cannot be inside its run destination: {resolved_source}") + + +def _copy_custom_inputs( + environment_source: Path, + dataset_source: Path, + *, + environment_output: Path, + dataset_output: Path, +) -> None: + """Copy caller-owned package and dataset into the isolated run directory.""" + if not environment_source.is_dir(): + raise ValueError(f"environment directory does not exist: {environment_source}") + if not dataset_source.is_file(): + raise ValueError(f"dataset does not exist or is not a file: {dataset_source}") + _require_safe_copy(environment_source, environment_output, label="environment") + _require_safe_copy(dataset_source, dataset_output, label="dataset") + + if environment_output.exists(): + shutil.rmtree(environment_output) + environment_output.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(environment_source, environment_output) + dataset_output.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(dataset_source, dataset_output) + + +def _select_resources_server( + available_servers: frozenset[str], + requested_server: str | None, +) -> str: + """Select a declared resources server or explain why selection is ambiguous.""" + if not available_servers: + raise ValueError("environment does not declare a resources server") + if requested_server is not None: + if requested_server not in available_servers: + raise ValueError( + f"resources server {requested_server!r} is not declared; " + f"available servers: {', '.join(sorted(available_servers))}" + ) + return requested_server + if len(available_servers) != 1: + raise ValueError( + "environment declares multiple resources servers; select one with " + f"--resources-server: {', '.join(sorted(available_servers))}" + ) + return next(iter(available_servers)) + + +def inspect_prepared_inputs( + environment: Path, + dataset: Path, + *, + requested_resources_server: str | None, + input_mode: str, +) -> PreparedEnvironment: + """Validate staged inputs and derive all component metadata dynamically.""" + environment_package = load_environment_package(environment) + if not isinstance(environment_package, WheelsV1Package): + actual_format = type(environment_package).__name__ + raise ValueError(f"expected a wheels-v1 package, got {actual_format}") + + components = inspect_environment_components(environment_package) + resources_server = _select_resources_server( + components.resources_servers, + requested_resources_server, + ) + tasks = discover_gym_tasks(dataset) + if not tasks: + raise ValueError(f"Gym discovered no tasks in {dataset}") + + return PreparedEnvironment( + root=environment_package.root, + dataset=dataset.resolve(), + name=environment_package.manifest.metadata.name, + resources_server=resources_server, + resources_servers=tuple(sorted(components.resources_servers)), + wheel_names=tuple(wheel.name for wheel in environment_package.wheel_files), + task_count=len(tasks), + input_mode=input_mode, + ) + + +def prepare_custom_inputs( + environment_source: Path, + dataset_source: Path, + *, + environment_output: Path, + dataset_output: Path, + requested_resources_server: str | None, +) -> PreparedEnvironment: + """Copy and validate a developer-provided environment package and dataset.""" + # Validate the caller-owned tree before copying so symlinks cannot be + # dereferenced into apparently valid files inside the run directory. + source_package = load_environment_package(environment_source) + if not isinstance(source_package, WheelsV1Package): + actual_format = type(source_package).__name__ + raise ValueError(f"expected a wheels-v1 package, got {actual_format}") + + _copy_custom_inputs( + environment_source, + dataset_source, + environment_output=environment_output, + dataset_output=dataset_output, + ) + return inspect_prepared_inputs( + environment_output, + dataset_output, + requested_resources_server=requested_resources_server, + input_mode="custom", + ) diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/prerequisites.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/prerequisites.py new file mode 100644 index 0000000000..0dc8b5b74f --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/prerequisites.py @@ -0,0 +1,296 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Read-only prerequisite checks used by ``workflow.py``. + +``PrerequisiteChecker`` validates local tools, Kubernetes resources, active +Platform configuration, deployed image references, and registry visibility +before the workflow creates anything. Keeping those checks here makes their +read-only boundary explicit and keeps the lifecycle orchestration readable. +This module is an implementation detail; developers invoke ``run.py``. +""" + +from __future__ import annotations + +import json +import shutil +from dataclasses import dataclass +from typing import Any + +import yaml +from artifacts import write_json +from commands import CommandRunner +from config import Settings +from console import Console + + +class PrerequisiteError(RuntimeError): + """Raised when the existing workstation or cluster is not ready.""" + + +def require_prerequisite(condition: bool, message: str) -> None: + """Raise a focused prerequisite error when a required condition is false.""" + if not condition: + raise PrerequisiteError(message) + + +@dataclass(frozen=True, slots=True) +class DeploymentImages: + """Release-matched images required by the evaluation workflow.""" + + registry: str + tag: str + + @property + def api(self) -> str: + """Return the API/controller image reference.""" + return f"{self.registry}/nmp-api:{self.tag}" + + @property + def cpu_tasks(self) -> str: + """Return the Evaluator task image reference.""" + return f"{self.registry}/nmp-cpu-tasks:{self.tag}" + + @property + def gym_host(self) -> str: + """Return the OpenSandbox Gym runtime image reference.""" + return f"{self.registry}/nmp-gym-host:{self.tag}" + + +class PrerequisiteChecker: + """Validate the preconfigured cluster and published images without changing them.""" + + def __init__( + self, + settings: Settings, + runner: CommandRunner, + console: Console, + ) -> None: + """Store workflow collaborators used by every prerequisite check.""" + self.settings = settings + self.runner = runner + self.console = console + + def _kubectl(self, *arguments: str) -> str: + """Run a read-only kubectl command and return stdout.""" + return self.runner.run(["kubectl", *arguments]) + + def _load_platform_config(self) -> dict[str, Any]: + """Load the active Platform YAML configuration from its ConfigMap.""" + rendered_config = self._kubectl( + "get", + "configmap", + "nemo-platform-config", + "-n", + self.settings.namespace, + "-o", + r"jsonpath={.data.config\.yaml}", + ) + config = yaml.safe_load(rendered_config) + require_prerequisite( + isinstance(config, dict), + "Platform ConfigMap does not contain a YAML mapping", + ) + return config + + def _check_workstation_tools(self) -> None: + """Require each command-line tool used later in the workflow.""" + for command in ("docker", "helm", "kubectl", "uv"): + require_prerequisite( + shutil.which(command) is not None, + f"required command not found: {command}", + ) + + def _check_cluster_resources(self) -> dict[str, Any]: + """Require the Platform, OpenSandbox, Secrets, and shared storage.""" + self._kubectl("get", "namespace", self.settings.namespace) + self.runner.run( + [ + "helm", + "status", + self.settings.release, + "-n", + self.settings.namespace, + "-o", + "json", + ], + output_path=self.settings.paths.evidence / "helm-status.json", + ) + self._kubectl( + "get", + "service", + self.settings.opensandbox_service, + "-n", + self.settings.opensandbox_namespace, + ) + self._kubectl( + "get", + "secret", + self.settings.opensandbox_api_secret, + "-n", + self.settings.namespace, + ) + self._kubectl( + "get", + "secret", + self.settings.registry_secret, + "-n", + self.settings.namespace, + ) + + persistent_volume_claim = json.loads( + self._kubectl( + "get", + "pvc", + self.settings.job_pvc, + "-n", + self.settings.namespace, + "-o", + "json", + ) + ) + access_modes = persistent_volume_claim.get("spec", {}).get("accessModes", []) + require_prerequisite( + "ReadWriteMany" in access_modes, + f"PVC {self.settings.job_pvc} must support ReadWriteMany", + ) + write_json( + self.settings.paths.evidence / "job-pvc.json", + persistent_volume_claim, + ) + return persistent_volume_claim + + def _resolve_images(self, config: dict[str, Any]) -> DeploymentImages: + """Resolve and validate the immutable image release from Platform config.""" + platform_config = config.get("platform", {}) + image_registry = platform_config.get("image_registry") + image_tag = platform_config.get("image_tag") + require_prerequisite( + isinstance(image_registry, str) and bool(image_registry), + "platform.image_registry is missing", + ) + require_prerequisite( + isinstance(image_tag, str) and bool(image_tag), + "platform.image_tag is missing", + ) + require_prerequisite( + image_tag != "latest", + "platform.image_tag must be immutable; latest is not supported", + ) + return DeploymentImages(registry=image_registry, tag=image_tag) + + def _check_sandbox_config( + self, + config: dict[str, Any], + images: DeploymentImages, + ) -> None: + """Require Platform and Evaluator settings needed by sandboxed Gym.""" + platform_config = config.get("platform", {}) + evaluator_config = config.get("evaluator", {}) + require_prerequisite( + platform_config.get("sandbox_cluster_capable") is True, + "Platform sandboxing is disabled", + ) + require_prerequisite( + evaluator_config.get("sandbox_cluster_capable") is True, + "Evaluator sandboxing is disabled", + ) + require_prerequisite( + evaluator_config.get("sandboxed_gym_default") is True, + "sandboxed Gym is not the default", + ) + require_prerequisite( + evaluator_config.get("sandbox_host_provider") == "opensandbox", + "Evaluator is not configured to use OpenSandbox", + ) + # An explicit runtime image can silently run stale code instead of the release-matched image. + require_prerequisite( + not evaluator_config.get("sandbox_runtime_image"), + "evaluator.sandbox_runtime_image must be absent or empty", + ) + require_prerequisite( + evaluator_config.get("sandbox_job_storage_pvc_claim") == self.settings.job_pvc, + "Evaluator uses the wrong sandbox storage PVC", + ) + require_prerequisite( + self.settings.internal_api in evaluator_config.get("sandbox_policy_base_urls", []), + "the internal Platform API is absent from sandbox_policy_base_urls", + ) + + selected_config = { + "platform": { + "image_registry": images.registry, + "image_tag": images.tag, + "sandbox_cluster_capable": platform_config.get("sandbox_cluster_capable"), + }, + "evaluator": { + key: evaluator_config.get(key) + for key in ( + "sandbox_cluster_capable", + "sandboxed_gym_default", + "sandbox_host_provider", + "sandbox_runtime_image", + "sandbox_job_storage_pvc_claim", + "sandbox_policy_base_urls", + ) + }, + } + write_json(self.settings.paths.evidence / "platform-config.json", selected_config) + + def _check_deployment_images(self, images: DeploymentImages) -> None: + """Require API and controller deployments to use the configured API image.""" + deployments = json.loads( + self._kubectl( + "get", + "deployments", + "-n", + self.settings.namespace, + "-l", + f"app.kubernetes.io/instance={self.settings.release}", + "-o", + "json", + ) + ) + deployed_api_images = [ + container["image"] + for deployment in deployments.get("items", []) + for container in deployment["spec"]["template"]["spec"]["containers"] + if "/nmp-api:" in container["image"] + ] + require_prerequisite( + bool(deployed_api_images), + "no nmp-api API/controller deployment was found", + ) + require_prerequisite( + all(image == images.api for image in deployed_api_images), + "an API/controller deployment uses an image that differs from Platform config", + ) + write_json(self.settings.paths.evidence / "deployments.json", deployments) + + def _check_published_images(self, images: DeploymentImages) -> None: + """Require all release-matched image manifests to be readable locally.""" + for image_name, image_reference in ( + ("nmp-api", images.api), + ("nmp-cpu-tasks", images.cpu_tasks), + ("nmp-gym-host", images.gym_host), + ): + self.runner.run( + ["docker", "buildx", "imagetools", "inspect", image_reference], + output_path=self.settings.paths.evidence / f"{image_name}-image.txt", + ) + + def check(self) -> DeploymentImages: + """Run every read-only prerequisite check and return resolved images.""" + self.settings.paths.evidence.mkdir(parents=True, exist_ok=True) + self._check_workstation_tools() + self._check_cluster_resources() + platform_config = self._load_platform_config() + images = self._resolve_images(platform_config) + self._check_sandbox_config(platform_config, images) + self._check_deployment_images(images) + self._check_published_images(images) + + self.console.detail("Image registry", images.registry) + self.console.detail("Image tag", images.tag) + self.console.detail("Platform namespace", self.settings.namespace) + return images diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/run.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/run.py new file mode 100644 index 0000000000..8ef0052de2 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/run.py @@ -0,0 +1,77 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Developer entry point for the custom Gym environment workflow. + +Run this file directly from the repository root. It parses the small public CLI, +loads optional environment overrides, obtains the NVIDIA API key, and delegates +the complete operation to ``CustomGymEnvironmentWorkflow``. The other Python +files in this directory are implementation modules and are not run directly. +""" + +from __future__ import annotations + +import argparse +import getpass +import os +from pathlib import Path + +from config import Settings + + +def _parser() -> argparse.ArgumentParser: + """Build the command-line parser for the single developer entry point.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--run-dir", + type=Path, + help="Artifact directory; defaults to a unique /tmp/nmp-gym-custom-environment-* path", + ) + parser.add_argument( + "--environment-dir", + type=Path, + help="Complete wheels-v1 environment directory; requires --dataset", + ) + parser.add_argument( + "--dataset", + type=Path, + help="Gym JSONL dataset for --environment-dir", + ) + parser.add_argument( + "--resources-server", + help="Resources server to run; inferred when the environment declares exactly one", + ) + return parser + + +def _inference_api_key() -> str: + """Read the inference API key from the environment or a hidden prompt.""" + configured_key = os.environ.get("INFERENCE_NVIDIA_API_KEY", "") + if configured_key: + return configured_key + return getpass.getpass("NVIDIA Inference API key: ") + + +def main() -> int: + """Parse configuration and execute the complete workflow once.""" + parser = _parser() + arguments = parser.parse_args() + if (arguments.environment_dir is None) != (arguments.dataset is None): + parser.error("--environment-dir and --dataset must be supplied together") + + # Delay plugin-heavy imports so ``--help`` stays fast and side-effect free. + from workflow import CustomGymEnvironmentWorkflow + + settings = Settings.from_environment( + run_dir=arguments.run_dir, + environment_dir=arguments.environment_dir, + dataset=arguments.dataset, + resources_server=arguments.resources_server, + ) + workflow = CustomGymEnvironmentWorkflow(settings) + workflow.run(inference_api_key=_inference_api_key()) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/scorer/pyproject.toml b/plugins/nemo-evaluator/scripts/gym-custom-environment/scorer/pyproject.toml new file mode 100644 index 0000000000..a02021ab2a --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/scorer/pyproject.toml @@ -0,0 +1,19 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Build metadata used by prepare.py to create the wheel delivered in the environment FileSet. +# This project is not published independently. + +[project] +name = "nmp-ascii-tree-evaluator" +version = "0.1.0" +description = "Custom ASCII Tree scorer for thenmp-ascii-tree-evaluator Gym environment workflow" +requires-python = ">=3.11" +license = "Apache-2.0" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/nmp_ascii_tree_evaluator"] diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/scorer/src/nmp_ascii_tree_evaluator/scoring.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/scorer/src/nmp_ascii_tree_evaluator/scoring.py new file mode 100644 index 0000000000..c2a9a8258f --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/scorer/src/nmp_ascii_tree_evaluator/scoring.py @@ -0,0 +1,90 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""FileSet-delivered reward functions for the ASCII Tree example. + +``prepare.py`` packages this module into a wheel. Gym installs that wheel in the +OpenSandbox runtime because the resources server declares it in +``requirements.txt``. The resources server then calls ``ascii_tree_reward()`` +for each model response. The formulas preserve Prime Intellect ASCII Tree 0.1.5 +semantics while remaining independent of its unsupported ``verifiers`` runtime. +""" + +from __future__ import annotations + +import difflib +import re + +_ASCII_FORMATTED = re.compile( + r"\s*(.*?)\s*", + flags=re.DOTALL, +) + + +def extract_ascii_formatted(text: str) -> str | None: + """Return the first ``ascii_formatted`` payload, or ``None`` when absent.""" + match = _ASCII_FORMATTED.search(text) + if match is None: + return None + parsed_tree = match.group(1).strip() + return parsed_tree or None + + +def _format_multiplier(lines: list[str]) -> float: + """Apply indentation and tree-branch formatting penalties.""" + multiplier = 1.0 + if not all(line.startswith(" ") or line.rstrip() == lines[0] for line in lines[1:]): + multiplier *= 0.5 + if not any("--" in line for line in lines[1:]): + multiplier *= 0.5 + return multiplier + + +def similarity_reward(completion: str, answer: str) -> float: + """Score whole-sequence line similarity with ASCII formatting penalties.""" + parsed_tree = extract_ascii_formatted(completion) + if parsed_tree is None or not answer.strip(): + return 0.0 + + try: + completion_lines = parsed_tree.split("\n") + expected_lines = answer.strip().split("\n") + similarity = difflib.SequenceMatcher( + None, + completion_lines, + expected_lines, + ).ratio() + return similarity * _format_multiplier(completion_lines) + except Exception: + # A malformed model response should receive zero reward, not crash the Gym run. + return 0.0 + + +def continuous_reward(completion: str, answer: str) -> float: + """Score the longest contiguous line match with formatting penalties.""" + parsed_tree = extract_ascii_formatted(completion) + if parsed_tree is None or not answer.strip(): + return 0.0 + + try: + completion_lines = parsed_tree.split("\n") + expected_lines = answer.strip().split("\n") + matcher = difflib.SequenceMatcher(None, completion_lines, expected_lines) + longest_block = max( + matcher.get_matching_blocks(), + key=lambda block: block.size, + default=difflib.Match(0, 0, 0), + ) + contiguous_fraction = longest_block.size / len(expected_lines) + return contiguous_fraction * _format_multiplier(completion_lines) + except Exception: + # Preserve the source environment's fail-closed scoring behavior. + return 0.0 + + +def ascii_tree_reward(completion: str, answer: str) -> float: + """Combine global similarity (30%) and contiguous matching (70%).""" + return 0.3 * similarity_reward(completion, answer) + 0.7 * continuous_reward( + completion, + answer, + ) diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/submit.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/submit.py new file mode 100644 index 0000000000..caca4448e2 --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/submit.py @@ -0,0 +1,308 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Internal Evaluator submission and result-validation logic. + +``workflow.py`` calls ``submit_and_validate()`` after it has uploaded the custom +environment and configured inference. This module translates the prepared +dataset and FileSet into an Evaluator job payload, waits for that job, downloads +its result archive, and verifies that the persisted trial and reward agree. It +is not a standalone developer entry point; run ``run.py`` instead. +""" + +from __future__ import annotations + +import math +import shutil +import tarfile +import time +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from artifacts import read_jsonl, write_json +from nemo_evaluator.filesets import FilesetRef +from nemo_evaluator.jobs.agent_spec import GymRunnerTarget +from nemo_evaluator.shared.metric_bundles.bundles import bundle_metric +from nemo_evaluator.shared.metric_bundles.inline import InlineMetricBundlePackager +from nemo_evaluator_sdk.agent_eval.runtimes.gym import GymRewardMetric, discover_gym_tasks +from nemo_platform import NeMoPlatform +from nemo_platform_plugin.jobs.types import PlatformJobResponse + +JOB_TIMEOUT_SECONDS = 1200 +JOB_POLL_INTERVAL_SECONDS = 2 +JOB_PROGRESS_INTERVAL_SECONDS = 30 +TERMINAL_JOB_STATUSES = frozenset({"completed", "error", "cancelled"}) + + +@dataclass(frozen=True, slots=True) +class EvaluationSubmission: + """Inputs needed to submit one custom Gym evaluation.""" + + base_url: str + internal_api: str + workspace: str + fileset: str + model: str + dataset: Path + resources_server: str + evidence_directory: Path + + +@dataclass(frozen=True, slots=True) +class EvaluationResult: + """Validated identifiers and scores produced by the evaluation.""" + + job_name: str + verification: dict[str, Any] + + +def _post_evaluator_payload( + sdk: NeMoPlatform, + workspace: str, + path: str, + payload: Mapping[str, object], +) -> object: + """Post an untyped payload to an Evaluator API route.""" + return sdk.post( + f"/apis/evaluator/v2/workspaces/{workspace}/{path.lstrip('/')}", + cast_to=object, + body=dict(payload), + ) + + +def _require_job_name(payload: object) -> str: + """Extract the submitted Platform job name from an API response.""" + job_name = payload.get("name") if isinstance(payload, Mapping) else None + if not isinstance(job_name, str): + raise TypeError(f"unexpected job response: {payload!r}") + return job_name + + +def _wait_for_job( + sdk: NeMoPlatform, + job_name: str, + workspace: str, + progress: Callable[[str], None] | None = None, +) -> PlatformJobResponse: + """Poll until terminal, reporting status changes and occasional heartbeats.""" + started_at = time.monotonic() + deadline = started_at + JOB_TIMEOUT_SECONDS + next_progress_at = started_at + status_history: list[str] = [] + while True: + job = sdk.jobs.retrieve(job_name, workspace=workspace) + status = str(getattr(job.status, "value", job.status)).lower() + status_changed = not status_history or status_history[-1] != status + if status_changed: + status_history.append(status) + now = time.monotonic() + if progress is not None and (status_changed or now >= next_progress_at): + progress(f"{status} ({now - started_at:.0f}s elapsed)") + next_progress_at = now + JOB_PROGRESS_INTERVAL_SECONDS + if status in TERMINAL_JOB_STATUSES: + return job + if now >= deadline: + history = " -> ".join(status_history) + raise TimeoutError( + f"job {job_name!r} did not finish within {JOB_TIMEOUT_SECONDS} seconds; " + f"last status={status!r}, history={history}" + ) + time.sleep(JOB_POLL_INTERVAL_SECONDS) + + +def _task_payload(task: Any, reward_metric: dict[str, Any]) -> dict[str, Any]: + """Convert one discovered Gym task into the Evaluator API representation.""" + return { + "id": task.id, + "intent": task.intent, + "inputs": task.inputs or {}, + "reference": task.reference or {}, + "metrics": [reward_metric], + "metadata": [{"key": key, "value": value} for key, value in (task.metadata or {}).items()], + } + + +def _output_texts(trial: dict[str, Any]) -> list[str]: + """Extract assistant output text from a persisted Responses API trial.""" + trial_output = trial.get("output") + if not isinstance(trial_output, Mapping): + return [] + response = trial_output.get("response") + if not isinstance(response, Mapping): + return [] + response_output = response.get("output") + if not isinstance(response_output, list): + return [] + + texts: list[str] = [] + for output_item in response_output: + if not isinstance(output_item, Mapping): + continue + content_items = output_item.get("content") + if not isinstance(content_items, list): + continue + for content in content_items: + if not isinstance(content, Mapping) or content.get("type") != "output_text": + continue + text = content.get("text") + if isinstance(text, str) and text: + texts.append(text) + return texts + + +def _build_job_payload(submission: EvaluationSubmission) -> dict[str, Any]: + """Build the Evaluator job payload for one dataset task and one repetition.""" + discovered_tasks = discover_gym_tasks(submission.dataset)[:1] + if len(discovered_tasks) != 1: + raise RuntimeError(f"expected exactly one selected task, got {len(discovered_tasks)}") + + reward_metric = bundle_metric( + GymRewardMetric(), + InlineMetricBundlePackager(), + ).model_dump(mode="json") + target = GymRunnerTarget( + environment=FilesetRef(root=f"{submission.workspace}/{submission.fileset}"), + agent="simple_agent", + agent_config="responses_api_agents/simple_agent/configs/simple_agent.yaml", + resources_server=submission.resources_server, + num_repeats=1, + concurrency=1, + hydra_params={ + "policy_base_url": ( + f"{submission.internal_api}/apis/inference-gateway/v2/workspaces/" + f"{submission.workspace}/model/{submission.model}/-/v1" + ), + "policy_api_key": "not-used", + "policy_model_name": submission.model, + }, + ) + return { + "spec": { + "tasks": [_task_payload(discovered_tasks[0], reward_metric)], + "target": target.model_dump(mode="json"), + } + } + + +def _download_results( + sdk: NeMoPlatform, + submission: EvaluationSubmission, + job_name: str, +) -> Path: + """Download and safely extract the evaluation result archive.""" + archive_path = submission.evidence_directory / "agent-eval-results.tar.gz" + archive_response = sdk.jobs.results.download( + "agent-eval-results", + job=job_name, + workspace=submission.workspace, + ) + archive_contents = archive_response.read() + if not archive_contents: + raise RuntimeError(f"job {job_name} returned an empty result archive") + archive_path.write_bytes(archive_contents) + + extraction_directory = submission.evidence_directory / "agent-eval-results" + if extraction_directory.exists(): + shutil.rmtree(extraction_directory) + extraction_directory.mkdir(parents=True) + with tarfile.open(archive_path) as archive: + # The data filter prevents archive members from escaping the destination. + archive.extractall(extraction_directory, filter="data") + return extraction_directory + + +def _validate_results( + extraction_directory: Path, + submission: EvaluationSubmission, + job_name: str, + job_status: str, +) -> dict[str, Any]: + """Validate trial output, custom reward coverage, and score consistency.""" + trials_files = list(extraction_directory.rglob("trials.jsonl")) + scores_files = list(extraction_directory.rglob("scores.jsonl")) + if len(trials_files) != 1 or len(scores_files) != 1: + raise RuntimeError(f"expected one trials.jsonl and one scores.jsonl, got {trials_files=} {scores_files=}") + + trials = read_jsonl(trials_files[0]) + scores = read_jsonl(scores_files[0]) + if len(trials) != 1 or len(scores) != 1: + raise RuntimeError(f"expected one trial and one score, got {len(trials)=} {len(scores)=}") + + trial = trials[0] + score = scores[0] + if trial["error"] is not None: + raise RuntimeError(f"trial failed: {trial['error']}") + output_texts = _output_texts(trial) + if not output_texts: + raise RuntimeError(f"trial returned no assistant output text: {trial}") + + trial_reward = trial.get("metadata", {}).get("reward") + if not isinstance(trial_reward, int | float) or not math.isfinite(trial_reward): + raise RuntimeError(f"trial reward is not finite: {trial_reward!r}") + + if score.get("status") != "completed": + raise RuntimeError(f"metric did not complete: {score}") + metric_outputs = score.get("outputs", []) + if len(metric_outputs) != 1 or metric_outputs[0].get("name") != "reward": + raise RuntimeError(f"unexpected metric outputs: {metric_outputs}") + if metric_outputs[0].get("value") != trial_reward: + raise RuntimeError(f"metric reward does not match trial reward: {metric_outputs[0]} != {trial_reward}") + + verification = { + "job_name": job_name, + "job_status": job_status, + "trial_id": trial.get("id"), + "trial_status": trial.get("status"), + "score_status": score.get("status"), + "reward": trial_reward, + "resources_server": submission.resources_server, + "output_texts": output_texts, + } + write_json(submission.evidence_directory / "verification.json", verification) + return verification + + +def submit_and_validate( + submission: EvaluationSubmission, + *, + sdk: NeMoPlatform | None = None, + progress: Callable[[str], None] | None = None, +) -> EvaluationResult: + """Submit one job, wait for completion, download results, and validate them.""" + submission.evidence_directory.mkdir(parents=True, exist_ok=True) + platform = sdk or NeMoPlatform(base_url=submission.base_url, max_retries=2) + payload = _build_job_payload(submission) + response = _post_evaluator_payload( + platform, + submission.workspace, + "agent-evaluate/jobs", + payload, + ) + job_name = _require_job_name(response) + (submission.evidence_directory / "job-name.txt").write_text( + job_name + "\n", + encoding="utf-8", + ) + + job = _wait_for_job(platform, job_name, submission.workspace, progress) + if job.status.lower() != "completed": + messages = [ + f"[{entry.timestamp}] {entry.message}" + for entry in platform.jobs.get_logs( + job_name, + workspace=submission.workspace, + ) + ] + details = "\n".join(messages) + raise RuntimeError(f"job did not complete: {job.status}\n{details}") + + extraction_directory = _download_results(platform, submission, job_name) + verification = _validate_results( + extraction_directory, + submission, + job_name, + job.status, + ) + return EvaluationResult(job_name=job_name, verification=verification) diff --git a/plugins/nemo-evaluator/scripts/gym-custom-environment/workflow.py b/plugins/nemo-evaluator/scripts/gym-custom-environment/workflow.py new file mode 100644 index 0000000000..0f17381b8e --- /dev/null +++ b/plugins/nemo-evaluator/scripts/gym-custom-environment/workflow.py @@ -0,0 +1,550 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Internal orchestration for the developer-facing ``run.py`` script. + +``run.py`` constructs ``CustomGymEnvironmentWorkflow`` and calls ``run()``. +This module owns the ordered operation: prepare and upload an environment, +configure inference, submit an evaluation, verify generic runtime evidence, +and clean up. Submission details, input adapters, and prerequisite checks live +in dedicated modules so this file remains focused on the lifecycle. +""" + +from __future__ import annotations + +import http.client +import math +import subprocess +import time +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import ascii_tree_example +import prepare +from artifacts import read_json, write_json +from commands import CommandRunner +from config import Settings +from console import Console +from nemo_platform import NeMoPlatform +from prerequisites import DeploymentImages, PrerequisiteChecker +from submit import EvaluationSubmission, submit_and_validate + +TOTAL_STEPS = 7 + + +class ResultVerificationError(RuntimeError): + """Raised when a completed workflow does not meet an acceptance criterion.""" + + +def require_result(condition: bool, message: str) -> None: + """Raise a focused verification error when an acceptance criterion fails.""" + if not condition: + raise ResultVerificationError(message) + + +def nested_messages(value: Any) -> list[str]: + """Collect every string-valued ``message`` field from nested JSON data.""" + messages: list[str] = [] + if isinstance(value, dict): + message = value.get("message") + if isinstance(message, str): + messages.append(message) + for child in value.values(): + messages.extend(nested_messages(child)) + elif isinstance(value, list): + for child in value: + messages.extend(nested_messages(child)) + return messages + + +@dataclass(slots=True) +class CreatedResources: + """Track run-scoped resources that need cleanup.""" + + fileset: bool = False + inference_secret: bool = False + inference_provider: bool = False + + +class CustomGymEnvironmentWorkflow: + """Run the documented developer workflow and clean up temporary resources.""" + + def __init__( + self, + settings: Settings, + *, + runner: CommandRunner | None = None, + console: Console | None = None, + ) -> None: + """Create a workflow with injectable command and output collaborators.""" + self.settings = settings + self.runner = runner or CommandRunner(working_directory=settings.repo_root) + self.console = console or Console() + self.sdk = NeMoPlatform(base_url=settings.base_url, max_retries=2) + self.images: DeploymentImages | None = None + self.prepared_environment: prepare.PreparedEnvironment | None = None + self.job_name: str | None = None + self.created_resources = CreatedResources() + + def _kubectl(self, *arguments: str, output_path: Path | None = None) -> str: + """Run kubectl in the caller's current cluster context.""" + return self.runner.run(["kubectl", *arguments], output_path=output_path) + + def _require_images(self) -> DeploymentImages: + """Return resolved deployment images after prerequisite validation.""" + if self.images is None: + raise ResultVerificationError("deployment images have not been resolved") + return self.images + + def _require_prepared_environment(self) -> prepare.PreparedEnvironment: + """Return validated package metadata after preparation has completed.""" + if self.prepared_environment is None: + raise ResultVerificationError("custom environment has not been prepared") + return self.prepared_environment + + def verify_prerequisites(self) -> None: + """Run the dedicated read-only workstation and cluster checks.""" + self.console.step(1, TOTAL_STEPS, "Checking workstation and cluster prerequisites") + checker = PrerequisiteChecker(self.settings, self.runner, self.console) + self.images = checker.check() + + def prepare_custom_environment(self) -> None: + """Prepare and validate default or caller-provided environment inputs.""" + self.console.step(2, TOTAL_STEPS, "Preparing the custom Gym environment") + adapter_evidence: dict[str, Any] = {} + if self.settings.uses_default_example: + adapter_evidence = ascii_tree_example.prepare_example( + self.runner, + converter_environment=self.settings.paths.converter_environment, + converter_dataset=self.settings.paths.converter_dataset, + environment_output=self.settings.paths.environment, + dataset_output=self.settings.paths.dataset, + ) + prepared_environment = prepare.inspect_prepared_inputs( + self.settings.paths.environment, + self.settings.paths.dataset, + requested_resources_server=self.settings.requested_resources_server, + input_mode=str(adapter_evidence["input_mode"]), + ) + self.console.detail("Input", adapter_evidence["input_label"]) + else: + environment_source = self.settings.environment_source + dataset_source = self.settings.dataset_source + if environment_source is None or dataset_source is None: + raise ResultVerificationError("custom environment inputs are incomplete") + self.console.detail("Input", environment_source) + prepared_environment = prepare.prepare_custom_inputs( + environment_source, + dataset_source, + environment_output=self.settings.paths.environment, + dataset_output=self.settings.paths.dataset, + requested_resources_server=self.settings.requested_resources_server, + ) + + self.prepared_environment = prepared_environment + preparation_summary = { + **prepared_environment.evidence(), + **adapter_evidence, + } + write_json( + self.settings.paths.evidence / "preparation.json", + preparation_summary, + ) + self.console.detail("Environment format", "wheels-v1") + self.console.detail("Environment", prepared_environment.name) + self.console.detail("Dataset tasks", prepared_environment.task_count) + self.console.detail("Custom wheels", ", ".join(prepared_environment.wheel_names)) + self.console.detail("Resources server", prepared_environment.resources_server) + + @staticmethod + def _api_is_ready(port: int) -> bool: + """Return whether the local Platform readiness endpoint responds successfully.""" + connection = http.client.HTTPConnection("127.0.0.1", port, timeout=1) + try: + connection.request("GET", "/health/ready") + response = connection.getresponse() + response.read() + return response.status == 200 + except OSError: + return False + finally: + connection.close() + + @contextmanager + def platform_connection(self) -> Iterator[None]: + """Reuse a healthy API endpoint or own a temporary kubectl port-forward.""" + if self._api_is_ready(self.settings.local_port): + self.console.detail("Platform API", f"{self.settings.base_url} (existing)") + yield + return + + port_forward_log = self.settings.paths.evidence / "port-forward.log" + with port_forward_log.open("a", encoding="utf-8") as log: + process = self.runner.start( + [ + "kubectl", + "port-forward", + "-n", + self.settings.namespace, + f"service/{self.settings.platform_api_service}", + f"{self.settings.local_port}:8080", + ], + stdout=log, + ) + try: + for _ in range(30): + if self._api_is_ready(self.settings.local_port): + break + if process.poll() is not None: + raise ResultVerificationError(f"Platform API port-forward failed; see {port_forward_log}") + time.sleep(1) + else: + raise ResultVerificationError("Platform API port-forward did not become ready") + self.console.detail( + "Platform API", + f"{self.settings.base_url} (temporary port-forward)", + ) + yield + finally: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + + def _model_name(self) -> str: + """Return the workspace-relative model name required by API routes.""" + workspace_prefix = f"{self.settings.workspace}/" + require_result( + self.settings.model_entity_id.startswith(workspace_prefix), + (f"NMP_GYM_CUSTOM_MODEL_ENTITY_ID must start with {workspace_prefix}"), + ) + return self.settings.model_entity_id.removeprefix(workspace_prefix) + + def upload_custom_environment(self) -> None: + """Create a temporary FileSet and upload the custom environment package.""" + self.sdk.files.filesets.create( + name=self.settings.fileset, + workspace=self.settings.workspace, + purpose="environment", + description="Custom wheels-v1 Gym environment workflow", + ) + self.created_resources.fileset = True + self.sdk.files.upload( + local_path=f"{self.settings.paths.environment}/", + fileset=self.settings.fileset, + workspace=self.settings.workspace, + ) + fileset_listing = self.sdk.files.list( + fileset=self.settings.fileset, + workspace=self.settings.workspace, + ) + expected_paths = { + path.relative_to(self.settings.paths.environment).as_posix() + for path in self.settings.paths.environment.rglob("*") + if path.is_file() + } + uploaded_paths = {uploaded_file.path for uploaded_file in fileset_listing.data} + require_result( + uploaded_paths == expected_paths, + ( + "uploaded environment FileSet does not match the local package: " + f"missing={sorted(expected_paths - uploaded_paths)}, " + f"unexpected={sorted(uploaded_paths - expected_paths)}" + ), + ) + write_json( + self.settings.paths.evidence / "fileset-listing.json", + fileset_listing, + ) + self.console.detail("Environment FileSet", self.settings.fileset) + + def create_inference_provider(self, inference_api_key: str) -> None: + """Create a temporary NVIDIA Inference Hub Secret and provider.""" + require_result( + bool(inference_api_key), + "INFERENCE_NVIDIA_API_KEY is required", + ) + self.sdk.secrets.create( + name=self.settings.inference_secret, + value=inference_api_key, + workspace=self.settings.workspace, + description="Temporary key for the custom Gym environment workflow", + ) + self.created_resources.inference_secret = True + self.sdk.inference.providers.create( + name=self.settings.inference_provider, + workspace=self.settings.workspace, + host_url="https://inference-api.nvidia.com/v1", + api_key_secret_name=self.settings.inference_secret, + ) + self.created_resources.inference_provider = True + + deadline = time.monotonic() + 120 + while True: + provider = self.sdk.inference.providers.retrieve( + self.settings.inference_provider, + workspace=self.settings.workspace, + ) + served_model_ids = {model.model_entity_id for model in (provider.served_models or [])} + if self.settings.model_entity_id in served_model_ids: + break + if provider.status in {"ERROR", "DELETED", "LOST"}: + raise ResultVerificationError( + f"provider entered {provider.status}: {provider.status_message or 'no status message'}" + ) + if time.monotonic() >= deadline: + raise ResultVerificationError( + f"provider did not discover {self.settings.model_entity_id} within 120 seconds" + ) + time.sleep(2) + + write_json( + self.settings.paths.evidence / "provider.json", + provider, + ) + self.console.detail("Model", self.settings.model_entity_id) + + def configure_platform_resources(self, inference_api_key: str) -> None: + """Upload the environment and create temporary inference resources.""" + self.console.step(3, TOTAL_STEPS, "Creating temporary Platform resources") + self.upload_custom_environment() + self.create_inference_provider(inference_api_key) + + def smoke_test_model(self) -> None: + """Send a small environment-independent prompt through the selected model route.""" + self.console.step(4, TOTAL_STEPS, "Smoke-testing the selected model") + request_body: dict[str, object] = { + "model": self.settings.model_entity_id, + "messages": [{"role": "user", "content": "Reply with OK."}], + "max_tokens": 16, + "temperature": 0, + } + smoke_response = self.sdk.inference.gateway.model.post( + "v1/chat/completions", + name=self._model_name(), + workspace=self.settings.workspace, + body=request_body, + ) + require_result( + isinstance(smoke_response.get("choices"), list) and bool(smoke_response["choices"]), + "model smoke test returned no choices", + ) + write_json( + self.settings.paths.evidence / "inference-smoke.json", + smoke_response, + ) + self.console.detail("Model route", "ready") + + def submit_evaluation(self) -> None: + """Submit one trial and download its validated result artifacts.""" + self.console.step(5, TOTAL_STEPS, "Running the custom Gym evaluation") + prepared_environment = self._require_prepared_environment() + submission = EvaluationSubmission( + base_url=self.settings.base_url, + internal_api=self.settings.internal_api, + workspace=self.settings.workspace, + fileset=self.settings.fileset, + model=self._model_name(), + dataset=self.settings.paths.dataset, + resources_server=prepared_environment.resources_server, + evidence_directory=self.settings.paths.evidence, + ) + evaluation_result = submit_and_validate( + submission, + sdk=self.sdk, + progress=lambda message: self.console.detail("Job progress", message), + ) + self.job_name = evaluation_result.job_name + self.console.detail("Job", evaluation_result.job_name) + self.console.detail("Custom reward", evaluation_result.verification["reward"]) + + job_status = self.sdk.jobs.get_status( + evaluation_result.job_name, + workspace=self.settings.workspace, + ) + job_logs = self.sdk.jobs.get_logs( + evaluation_result.job_name, + workspace=self.settings.workspace, + ) + write_json( + self.settings.paths.evidence / "job-status.json", + job_status, + ) + write_json( + self.settings.paths.evidence / "job-logs.json", + {"data": list(job_logs)}, + ) + + def verify_results(self) -> dict[str, Any]: + """Verify runtime images, sandbox lifecycle, and complete custom scoring.""" + self.console.step(6, TOTAL_STEPS, "Verifying evaluation evidence") + images = self._require_images() + require_result(bool(self.job_name), "evaluation job name is missing") + + job_status = read_json(self.settings.paths.evidence / "job-status.json") + job_logs = read_json(self.settings.paths.evidence / "job-logs.json") + verification = read_json(self.settings.paths.evidence / "verification.json") + summary_files = list((self.settings.paths.evidence / "agent-eval-results").rglob("summary.json")) + require_result( + len(summary_files) == 1, + f"expected one evaluation summary.json, found {summary_files}", + ) + result_summary = read_json(summary_files[0]) + status_messages = nested_messages(job_status) + log_messages = nested_messages(job_logs) + + require_result( + job_status.get("status") == "completed", + "Platform job did not complete", + ) + require_result( + job_status.get("error_details") is None, + "Platform job contains error details", + ) + require_result( + sum(images.cpu_tasks in message for message in status_messages) >= 2, + "both job steps did not use the configured CPU Tasks image", + ) + require_result( + any(f"Creating sandbox with startup source: {images.gym_host}" in message for message in log_messages), + "OpenSandbox did not use the configured Gym host image", + ) + + # Together these markers show that execution entered and exited OpenSandbox cleanly. + for expected_message in ( + "Successfully created sandbox:", + "Collecting 1 example(s)", + "Successfully terminated sandbox:", + ): + require_result( + any(expected_message in message for message in log_messages), + f"missing job evidence: {expected_message}", + ) + + reward_coverage = result_summary.get("metric_coverage", {}).get("gym_reward", {}).get("reward") + require_result( + reward_coverage == {"failed": 0, "missing": 0, "scored": 1, "total": 1}, + "Gym reward coverage is incomplete", + ) + reward = verification.get("reward") + require_result( + isinstance(reward, int | float) and math.isfinite(reward), + "custom reward is not finite", + ) + require_result( + verification.get("trial_status") == "completed", + "trial did not complete", + ) + require_result( + verification.get("score_status") == "completed", + "metric did not complete", + ) + require_result( + bool(verification.get("output_texts")), + "model returned no output text", + ) + + run_summary = { + "status": "passed", + "job_name": self.job_name, + "image_registry": images.registry, + "image_tag": images.tag, + "cpu_tasks_image": images.cpu_tasks, + "gym_host_image": images.gym_host, + "model": self.settings.model_entity_id, + "evaluation": verification, + } + write_json(self.settings.paths.evidence / "run-summary.json", run_summary) + self.console.detail("CPU Tasks image", images.cpu_tasks) + self.console.detail("Gym host image", images.gym_host) + self.console.detail("Sandbox lifecycle", "created and terminated") + return run_summary + + def cleanup(self) -> None: + """Delete every run-scoped Platform resource that was created.""" + cleanup_errors: list[str] = [] + resources: tuple[tuple[str, bool, Callable[[], object]], ...] = ( + ( + "inference provider", + self.created_resources.inference_provider, + lambda: self.sdk.inference.providers.delete( + self.settings.inference_provider, + workspace=self.settings.workspace, + ), + ), + ( + "inference secret", + self.created_resources.inference_secret, + lambda: self.sdk.secrets.delete( + self.settings.inference_secret, + workspace=self.settings.workspace, + ), + ), + ( + "environment FileSet", + self.created_resources.fileset, + lambda: self.sdk.files.filesets.delete( + self.settings.fileset, + workspace=self.settings.workspace, + ), + ), + ) + created_count = sum(was_created for _, was_created, _ in resources) + if created_count == 0: + self.console.detail("Temporary resources", "none created") + return + + deleted_count = 0 + for resource_name, was_created, delete_resource in resources: + if not was_created: + continue + try: + delete_resource() + deleted_count += 1 + except Exception as error: + # Attempt all deletions so one failure does not leak unrelated resources. + cleanup_errors.append(f"{resource_name}: {error}") + if cleanup_errors: + raise RuntimeError("cleanup failed:\n" + "\n".join(cleanup_errors)) + self.console.detail("Temporary resources", f"{deleted_count} deleted") + + def run(self, *, inference_api_key: str) -> dict[str, Any]: + """Execute the complete workflow through one developer-facing invocation.""" + self.settings.paths.evidence.mkdir(parents=True, exist_ok=True) + self.verify_prerequisites() + self.prepare_custom_environment() + + workflow_failed = False + with self.platform_connection(): + try: + self.configure_platform_resources(inference_api_key) + self.smoke_test_model() + self.submit_evaluation() + run_summary = self.verify_results() + except Exception: + workflow_failed = True + raise + finally: + if workflow_failed: + self.console.warning("Workflow stopped; cleaning up resources created before the failure") + else: + self.console.step(7, TOTAL_STEPS, "Cleaning up temporary resources") + try: + self.cleanup() + except RuntimeError as cleanup_error: + if workflow_failed: + self.console.warning(str(cleanup_error)) + else: + raise + + self.console.success("Custom Gym environment evaluation passed") + self.console.detail("Job", run_summary["job_name"]) + self.console.detail("Model", run_summary["model"]) + self.console.detail("Reward", run_summary["evaluation"]["reward"]) + self.console.detail("Evidence", self.settings.paths.evidence) + return run_summary