diff --git a/.github/workflows/verifiers.yml b/.github/workflows/verifiers.yml new file mode 100644 index 00000000..5b12bb42 --- /dev/null +++ b/.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 diff --git a/AGENTS.md b/AGENTS.md index 5b813d6b..daeb2724 100644 --- a/AGENTS.md +++ b/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 diff --git a/src/agent_env/training/nemo_gym/__main__.py b/src/agent_env/training/nemo_gym/__main__.py index 416add33..50d69691 100644 --- a/src/agent_env/training/nemo_gym/__main__.py +++ b/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/, 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/.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__": diff --git a/src/agent_env/training/nemo_gym/server.py b/src/agent_env/training/nemo_gym/server.py index 48ea6d21..b40e926f 100644 --- a/src/agent_env/training/nemo_gym/server.py +++ b/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({})), ) diff --git a/tst/nemo_gym/server_test.py b/tst/nemo_gym/server_test.py index 103d8ade..677b91eb 100644 --- a/tst/nemo_gym/server_test.py +++ b/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"} diff --git a/tst/util/capabilities.py b/tst/util/capabilities.py index 9a663a86..46f6d4e2 100644 --- a/tst/util/capabilities.py +++ b/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" diff --git a/tst/verifiers/__init__.py b/tst/verifiers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tst/verifiers/conftest.py b/tst/verifiers/conftest.py new file mode 100644 index 00000000..68973307 --- /dev/null +++ b/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 diff --git a/tst/verifiers/taskset_test.py b/tst/verifiers/taskset_test.py new file mode 100644 index 00000000..c3fbc9e2 --- /dev/null +++ b/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 +