feat(nemo-gym): serve --masked-as-error, and a verifiers contract job

.github/workflows/verifiers.yml

@@ -0,0 +1,68 @@
+name: verifiers contract
+
+# Rollouts through verifiers' NeMo Gym taskset, the one prime-rl trains through, against the resources server
+# agent_env.training.nemo_gym serves for a bundle (tst/verifiers), on verifiers 0.4.0 and on its main branch, which
+# prime-rl follows. verifiers pins mcp 2 and the framework mcp 1, so verifiers gets a venv of its own and only vf-eval
+# goes on PATH. It runs on pull requests that touch the helper, the OpenEnv server it plays or the tier, weekly, and on
+# demand; it isn't a required check.
+
+on:
+  pull_request:
+    paths:
+      - "src/agent_env/training/**"
+      - "src/agent_env/openenv/**"
+      - "tst/verifiers/**"
+      - "tst/nemo_gym/**"
+      - "tst/unit/training/**"
+      - "pyproject.toml"
+      - "uv.lock"
+      - ".github/workflows/verifiers.yml"
+  schedule:
+    - cron: "53 7 * * 1"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: verifiers-${{ github.event_name == 'pull_request' && github.ref || github.sha }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+env:
+  PIP_DISABLE_PIP_VERSION_CHECK: "1"
+
+jobs:
+  taskset:
+    name: taskset (verifiers ${{ matrix.verifiers }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - verifiers: "0.4.0"
+            requirement: "verifiers==0.4.0"
+          - verifiers: "main"
+            requirement: "verifiers @ git+https://github.com/PrimeIntellect-ai/verifiers@main"
+    steps:
+      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.13"
+      - name: Install from uv.lock with the nemo-gym extra
+        run: |
+          python -m pip install --quiet -r .github/ci-requirements.txt
+          uv sync --locked --extra dev --extra nemo-gym --python "$(command -v python)"
+      - name: Install verifiers in its own venv, with vf-eval on PATH
+        run: |
+          uv venv "$RUNNER_TEMP/verifiers" --python "$(command -v python)"
+          uv pip install --python "$RUNNER_TEMP/verifiers/bin/python" "${{ matrix.requirement }}"
+          mkdir -p "$RUNNER_TEMP/vf-bin"
+          ln -s "$RUNNER_TEMP/verifiers/bin/vf-eval" "$RUNNER_TEMP/vf-bin/vf-eval"
+          echo "$RUNNER_TEMP/vf-bin" >> "$GITHUB_PATH"
+          "$RUNNER_TEMP/verifiers/bin/python" -c "from importlib.metadata import version; print('verifiers', version('verifiers'))"
+      - name: verifiers tier
+        run: .venv/bin/python -m pytest tst/verifiers -rA --timeout=900 --junitxml=junit-verifiers.xml
+      - name: No skips
+        if: always()
+        run: .venv/bin/python .github/scripts/check_skip_policy.py junit-verifiers.xml

AGENTS.md

@@ -333,6 +333,9 @@ skips; a `changes` job decides, and a skipped `installer` counts as passed.
 requests that touch `agent_env/training`, `agent_env/openenv` or the tier, weekly, and on demand; it allows no skips
 and isn't a required check. `.github/workflows/nemo-gym.yml` runs `tst/nemo_gym` on Python 3.13 with the `nemo-gym`
 extra from `uv.lock` and with the latest nemo-gym, on the same triggers; it allows no skips and isn't a required check.
+`.github/workflows/verifiers.yml` runs `tst/verifiers`, `vf-eval` through verifiers' NeMo Gym taskset (prime-rl's)
+against `serve`, on verifiers 0.4.0 and its main branch, installed in a venv of its own since verifiers pins mcp 2, on
+the same triggers; it allows no skips and isn't a required check.
 `.github/workflows/plugin-api.yml` runs the `plugin-api` job on every pull
 request, title edits included, and on `main`: `.github/scripts/check_plugin_api.py` compares the
 plugin surface (its `BASES` and `USED`) between `HEAD^1` and `HEAD` with griffe, pinned in the `dev` extra, and fails on a break the

src/agent_env/training/nemo_gym/__main__.py

@@ -53,13 +53,17 @@ def write(source: str, root: str, name: str | None, sandbox: str | None, reward:
 @click.option("--host", default="127.0.0.1", show_default=True,
               help="The address to listen on. The server checks no token, so keep it off public interfaces.")
 @click.option("--port", type=int, required=True, help="The port to listen on.")
-def serve(folder: str, host: str, port: int) -> None:
+@click.option("--masked-as-error", is_flag=True,
+              help="Answer verify for a rollout the environment failed with HTTP 503 rather than mask_sample, for a "
+                   "client that doesn't read mask_sample, as verifiers' NeMo Gym taskset doesn't: it then fails the "
+                   "rollout, and prime-rl trains on none of it.")
+def serve(folder: str, host: str, port: int, masked_as_error: bool) -> None:
     """Runs the resources server written into FOLDER, a Gym root's resources_servers/<name>, on its own: without Gym's
     head server or the agent servers its config names. Point verifiers' NeMo Gym taskset at it with
     env.taskset.task.resources_url, and its env.taskset.dataset at the folder's data/<split>.jsonl."""
     if run_server is None:
         raise click.ClickException(UNSERVABLE)
-    run_server(folder, host=host, port=port)
+    run_server(folder, host=host, port=port, masked_as_error=masked_as_error)
 
 
 if __name__ == "__main__":

src/agent_env/training/nemo_gym/server.py

@@ -60,7 +60,10 @@ class AgentEnvResourcesServerConfig(BaseResourcesServerConfig):
     session may go without a request before its episode is ended, freeing its slot: Gym never verifies a rollout whose
     run failed or was cancelled. ``reward_components`` has verify also answer with each of the task's scores by step id,
     as NeMo RL's GDPO takes them, and makes the reward their sum, as NeMo RL requires of a reward with components; it
-    takes no ``reward``. It runs with one worker, since a rollout's session lives in its process."""
+    takes no ``reward``. ``masked_as_error`` answers verify for a masked rollout with HTTP 503 and why, rather than with
+    ``mask_sample``, for a client that doesn't read ``mask_sample``: verifiers' NeMo Gym taskset then fails the
+    rollout, which prime-rl leaves out of the loss and its group's baseline. It runs with one worker, since a rollout's
+    session lives in its process."""
 
     url: str | None = None
     bundle: str | None = None
@@ -71,6 +74,7 @@ class AgentEnvResourcesServerConfig(BaseResourcesServerConfig):
     reset_timeout: float = 600.0
     idle_timeout: float = 600.0
     reward_components: bool = False
+    masked_as_error: bool = False
 
     @model_validator(mode="after")
     def _checked(self) -> "AgentEnvResourcesServerConfig":
@@ -160,6 +164,8 @@ async def verify(self, request: Request, body: AgentEnvVerifyRequest) -> AgentEn
             reward, scores, failure = await session.result(_reply(body.response))
         if failure is not None:
             logger.warning("rollout masked: %s", failure)
+            if self.config.masked_as_error:
+                raise HTTPException(status_code=503, detail=failure)
         components = scores if self.config.reward_components and failure is None else None
         if components:
             reward = sum(components.values())
@@ -205,15 +211,19 @@ async def _shut_down(self) -> None:
                            "may be left running", served.url, STOP_SECONDS)
 
 
-def serve(folder: str | Path, *, host: str, port: int) -> None:
+def serve(folder: str | Path, *, host: str, port: int, masked_as_error: bool = False) -> None:
     """Runs the resources server ``write_server`` wrote into ``folder`` on its own, at ``host``:``port``, without Gym's
     head server or the agent servers its config names, for a trainer that plays the folder's rows against the server's
-    URL. As verifiers' launcher for Gym's own servers does, its client names a head server nothing calls."""
+    URL. ``host`` and ``port`` win over any the folder's config names, and ``masked_as_error``, as the config takes it,
+    is on when either it or the config sets it. As verifiers' launcher for Gym's own servers does, its client names a
+    head server nothing calls."""
     folder = Path(folder).resolve()
     written = yaml.safe_load((folder / "configs" / f"{folder.name}.yaml").read_text())
     config = written[folder.name]["resources_servers"][folder.name]
+    masked_as_error = masked_as_error or config.get("masked_as_error", False)
     resources = AgentEnvResourcesServer(
-        config=AgentEnvResourcesServerConfig(**config, name=folder.name, host=host, port=port),
+        config=AgentEnvResourcesServerConfig(**{**config, "name": folder.name, "host": host, "port": port,
+                                                "masked_as_error": masked_as_error}),
         server_client=ServerClient(head_server_config=BaseServerConfig(host=host, port=port),
                                    global_config_dict=OmegaConf.create({})),
     )

tst/nemo_gym/server_test.py

@@ -2,8 +2,9 @@
 cookie Gym keeps per rollout, over the bundle the unit tier serves on loopback: a row's seed starts its task's episode,
 the tool routes answer as text, verify ends the turns left open and scores the episode or masks the rollout, a rollout
 waits its turn at the server's limit, a closed or idle session ends its episode, the server takes one worker, a
-bundle named in the config is served for the server's lifetime, and ``serve`` runs a written server on its own. CI runs
-it on the nemo-gym ``uv.lock`` pins and on the latest nemo-gym."""
+bundle named in the config is served for the server's lifetime, and ``serve`` runs a written server on its own,
+answering a masked rollout's verify with an HTTP error when asked. CI runs it on the nemo-gym ``uv.lock`` pins and on
+the latest nemo-gym."""
 
 import asyncio
 import json
@@ -17,6 +18,7 @@
 import httpx
 import pydantic
 import pytest
+import yaml
 from fastapi.testclient import TestClient
 
 from agent_env.training.nemo_gym.prepare import write_server
@@ -178,26 +180,47 @@ def test_a_bundle_named_in_the_config_is_served_while_the_server_runs(tmp_path,
     assert len(teardowns.finished) == 2
 
 
-def test_serve_runs_a_written_server_on_its_own_at_the_port_it_is_given(serve, tmp_path, teardowns):
+@pytest.mark.parametrize(("flags", "written", "as_error"), [
+    ([], {}, False),
+    (["--masked-as-error"], {}, True),
+    ([], {"masked_as_error": True, "port": 1}, True),
+], ids=["default", "flag", "config"])
+def test_serve_runs_a_written_server_on_its_own_at_the_port_it_is_given_answering_a_masked_rollout_as_asked(
+        serve, tmp_path, teardowns, flags, written, as_error):
     bundle_server = serve({"a": PLAYED})
     folder = write_server(bundle_server.url, tmp_path)
+    configured = folder / "configs" / f"{folder.name}.yaml"
+    config = yaml.safe_load(configured.read_text())
+    config[folder.name]["resources_servers"][folder.name].update(written)
+    configured.write_text(yaml.safe_dump(config))
     row = json.loads((folder / "data" / "all.jsonl").read_text().splitlines()[0])
     port = _free_port()
 
-    served = subprocess.Popen([sys.executable, "-m", "agent_env.training.nemo_gym", "serve", ".", "--port", str(port)],
+    served = subprocess.Popen([sys.executable, "-m", "agent_env.training.nemo_gym", "serve", ".", "--port", str(port),
+                               *flags],
                               cwd=folder, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
     try:
         with httpx.Client(base_url=_up(served, f"http://127.0.0.1:{port}")) as client:
             assert client.post("/seed_session", json=row).json() == {}
             assert client.post("/add_item", json={"item": "crate"}).text == "add_item {'item': 'crate'}"
             verified = _verify(client, row, "Added a crate.")
+            client.post("/seed_session", json={**row, "task_id": "b"})
+            masked = client.post("/verify", json=_verified({**row, "task_id": "b"}, "Added a crate."))
     finally:
         served.terminate()
         served.wait(timeout=30)
     bundle_server.stop()
 
     assert (verified["reward"], verified["mask_sample"], verified["task_id"]) == (1.0, False, "a")
-    assert len(teardowns.finished) == 2
+    why = "the episode plays task 'a', not the row's 'b'"
+    if as_error:
+        assert masked.status_code == 503
+        assert why in masked.json()["detail"]
+    else:
+        assert masked.status_code == 200
+        assert (masked.json()["reward"], masked.json()["mask_sample"]) == (0.0, True)
+        assert why in masked.json()["failure_reason"]
+    assert len(teardowns.finished) == 3
 
 
 _EPISODE = {"rollout_id": "rollout"}

tst/util/capabilities.py

@@ -41,6 +41,10 @@
 #: isn't in the dev extra, and only that tier's CI job installs it.
 NEMO_GYM = "nemo_gym_installed"
 
+#: The verifiers tier (``tst/verifiers``) runs ``vf-eval`` from verifiers' own venv, since verifiers pins mcp 2 and the
+#: framework mcp 1; only that tier's CI job installs it and puts ``vf-eval`` on PATH.
+VERIFIERS = "verifiers_installed"
+
 #: ``collect_artifacts`` sizes a file with ``stat -c %s``, which only GNU stat takes (macOS ships BSD stat),
 #: so collecting off a local sandbox needs GNU stat first on PATH.
 GNU_STAT = "gnu_stat"

tst/verifiers/conftest.py

@@ -0,0 +1,7 @@
+"""The unit tier's bundle served on loopback over the Episode API's fake steps, which these tests play through
+verifiers' NeMo Gym taskset and AgentEnv's resources server."""
+
+from tst.unit.conftest import local_stores  # noqa: F401
+from tst.unit.openenv.environment_test import tools  # noqa: F401
+from tst.unit.task.episode.run_test import teardowns  # noqa: F401
+from tst.unit.training._support import serve  # noqa: F401

tst/verifiers/taskset_test.py

@@ -0,0 +1,115 @@
+"""verifiers' NeMo Gym taskset, prime-rl's, plays a written server's rows against ``serve`` with the null harness:
+``vf-eval`` seeds each row, calls its tools natively, and takes the reward from verify. A rollout the environment
+fails is masked, and with ``--masked-as-error`` it comes back failed, which prime-rl leaves out of training, rather
+than scored 0. The model is a streamed stand-in. CI runs it on verifiers 0.4.0 and on verifiers' main branch."""
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import pytest
+
+from agent_env.training.nemo_gym.prepare import write_server
+from tst.nemo_gym.server_test import _free_port, _up
+from tst.unit.training.nemo_gym_test import PLAYED
+from tst.util.capabilities import NEMO_GYM, VERIFIERS, missing_capability_reason
+
+pytest.importorskip("agent_env.training.nemo_gym.server", reason=missing_capability_reason(NEMO_GYM),
+                    exc_type=ImportError)
+VF_EVAL = shutil.which("vf-eval")
+pytestmark = pytest.mark.skipif(VF_EVAL is None, reason=missing_capability_reason(VERIFIERS))
+
+
+def test_vf_eval_scores_a_played_row_and_fails_a_masked_one(serve, tmp_path, teardowns):
+    bundle_server = serve({"a": PLAYED})
+    folder = write_server(bundle_server.url, tmp_path / "gym")
+    row = json.loads((folder / "data" / "all.jsonl").read_text().splitlines()[0])
+    (tmp_path / "rows.jsonl").write_text(json.dumps(row) + "\n" + json.dumps({**row, "task_id": "b"}) + "\n")
+    port, model = _free_port(), _model()
+
+    served = subprocess.Popen([sys.executable, "-m", "agent_env.training.nemo_gym", "serve", str(folder), "--port",
+                               str(port), "--masked-as-error"],
+                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    try:
+        url = _up(served, f"http://127.0.0.1:{port}")
+        (tmp_path / "eval.toml").write_text(_config(tmp_path, url, model.server_address[1]))
+        evaluated = subprocess.run([VF_EVAL, "@", str(tmp_path / "eval.toml"), "-n", "2", "-r", "1", "--no-push"],
+                                   env={**os.environ, "STAND_IN_API_KEY": "unused"}, capture_output=True, text=True,
+                                   timeout=600)
+    finally:
+        served.terminate()
+        served.wait(timeout=30)
+        model.shutdown()
+    bundle_server.stop()
+
+    assert evaluated.returncode == 0, evaluated.stdout + evaluated.stderr
+    traces = {trace["task"]["data"]["name"]: trace for trace in _traces(tmp_path / "out")}
+    played, masked = traces["rows:0"], traces["rows:1"]
+    assert (played["ok"], played.get("dropped"), played["rewards"]["nemo_gym"]["score"]) == (True, None, 1.0)
+    assert not masked["ok"]
+    assert "503 Service Unavailable" in masked["errors"][-1]["message"]
+    assert len(teardowns.finished) == 3
+
+
+def _config(tmp_path, url, model_port):
+    return f"""model = "stand-in"
+output_dir = "{tmp_path / 'out'}"
+
+[client]
+base_url = "http://127.0.0.1:{model_port}/v1"
+api_key_var = "STAND_IN_API_KEY"
+
+[env.taskset]
+id = "nemo-gym"
+dataset = "{tmp_path / 'rows.jsonl'}"
+
+[env.taskset.task]
+resources_url = "{url}"
+
+[env.agent.harness]
+id = "null"
+
+[env.agent.runtime]
+type = "subprocess"
+"""
+
+
+def _traces(out):
+    episodes = [json.loads(line) for path in out.rglob("traces.jsonl") for line in path.read_text().splitlines()]
+    return [trace for episode in episodes for trace in episode["traces"]]
+
+
+class _StandIn(BaseHTTPRequestHandler):
+    """An OpenAI-compatible model, streamed as verifiers asks: it adds a crate, then replies once the tool answered."""
+
+    def log_message(self, *args):
+        pass
+
+    def do_POST(self):
+        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+        answered = any(message.get("role") == "tool" for message in body["messages"])
+        delta = ({"role": "assistant", "content": "Added a crate."} if answered else
+                 {"role": "assistant", "tool_calls": [{"index": 0, "id": "call_add", "type": "function", "function": {
+                     "name": "add_item", "arguments": json.dumps({"item": "crate"})}}]})
+        head = {"id": "stand-in", "object": "chat.completion.chunk", "created": int(time.time()), "model": "stand-in"}
+        events = [head | {"choices": [{"index": 0, "delta": delta, "finish_reason": None}]},
+                  head | {"choices": [{"index": 0, "delta": {}, "finish_reason": "stop" if answered else "tool_calls"}]},
+                  head | {"choices": [], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}]
+        data = ("".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n").encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "text/event-stream")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+
+def _model():
+    server = ThreadingHTTPServer(("127.0.0.1", 0), _StandIn)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    return server
+