From 0ca5d400dfd46f658887fb3ea73925e1904ca7fe Mon Sep 17 00:00:00 2001 From: Ananth Subramaniam Date: Wed, 4 Nov 2026 13:02:29 -0800 Subject: [PATCH] feat(server): refuse several workers in servers that cannot share state Some servers keep state that session routing cannot reach, or own a resource that one process must hold, so they cannot run with several workers. They now override multiple_workers_unsupported_reason(), and run_webserver refuses num_workers > 1 with that reason before uvicorn starts: - aviary and toolsandbox keep environments in memory by the env_id in the request body, and their agents send step, verify and close without the session cookie; - critpt pools verify calls in memory into batches submitted together; - osworld owns the Docker host capacity, startup cleanup and its state file; - longmt_eval builds a Ray GPU actor pool per worker; - gdpval claims deliverable directories in memory by task and rollout ID; - terminal_bench_4 deduplicates seeds by the client_session_id and rollout_id in the request body; - vlm_eval_kit waits in memory for every row of an MMBench group; - voicebench's legacy scoring draws every rule-scored row from one random stream shared by the whole server; - enterpriseops_gym sessions and the partner_if mock-tool episodes are process-local, and agentic_if sessions hold their sandbox connections in one process; - the Legal Agent Bench Harbor bridge has an entrypoint outside its server directory, which uvicorn workers cannot import. speed_bench, wmt_translation, proof_judge, genrm_compare, ns_tools and terminus_2_sandboxed_agent refuse only in the configurations that need one process, and get the module-level app otherwise: speed_bench with the Prometheus metrics source, which measures from a snapshot in memory; wmt_translation with COMET, which builds a Ray GPU actor pool per worker; proof_judge with zero_reward_incorrect_groups; genrm_compare with cohort verification; ns_tools with the HTTP PythonTool sidecar on a fixed port; and terminus_2_sandboxed_agent without a resources server, which serves only process-local episode sessions. proof_judge's debug log also takes a file lock around each append, and still writes the record on a filesystem without file locks. The checks that did this job in model_post_init, in config validators and as num_workers: Literal[1], including those main added to voicebench, enterpriseops_gym, agentic_if, the partner_if mock tools and terminus_2_sandboxed_agent, become the method with the same conditions. gdpval, voicebench, the partner_if mock tools and speed_bench check the refusal at the start of model_post_init as well, before they install dependencies, load their backend or take a metrics snapshot, so a misconfigured server fails on the worker count rather than on that setup. A unit test reads every server entrypoint without importing it, including the entrypoints configs name outside app.py, and fails when a server has neither the module-level app nor a refusal. It fails on the parent commit. Signed-off-by: Ananth Subramaniam --- .../resources-server-apis.mdx | 2 +- resources_servers/agentic_if/app.py | 5 +- .../agentic_if/tests/test_app.py | 6 +- resources_servers/aviary/app.py | 7 + resources_servers/critpt/app.py | 6 + resources_servers/enterpriseops_gym/app.py | 5 +- .../enterpriseops_gym/tests/test_app.py | 5 +- resources_servers/gdpval/app.py | 13 +- .../gdpval/tests/test_sessions.py | 11 +- resources_servers/genrm_compare/app.py | 21 +- .../genrm_compare/tests/test_review_fixes.py | 22 +- .../legal_agent_bench/harbor_bridge.py | 6 + resources_servers/longmt_eval/app.py | 6 + resources_servers/ns_tools/app.py | 12 + resources_servers/ns_tools/tests/test_app.py | 19 ++ resources_servers/osworld/app.py | 7 + resources_servers/osworld/config.py | 4 - .../partner_if/mock_tools_app.py | 9 +- .../partner_if/tests/test_mock_tools_app.py | 2 +- resources_servers/proof_judge/app.py | 23 ++ .../proof_judge/tests/test_app.py | 73 ++++ resources_servers/speed_bench/app.py | 18 + .../speed_bench/tests/test_app.py | 7 + resources_servers/terminal_bench_4/app.py | 10 +- resources_servers/toolsandbox/app.py | 7 + resources_servers/vlm_eval_kit/app.py | 8 +- .../vlm_eval_kit/tests/test_app.py | 9 + resources_servers/voicebench/app.py | 11 +- .../voicebench/tests/test_app.py | 11 +- resources_servers/wmt_translation/app.py | 13 + .../wmt_translation/tests/test_app.py | 11 + .../terminus_2_sandboxed_agent/app.py | 13 +- .../tests/test_app.py | 13 +- .../test_multiple_workers_contract.py | 312 ++++++++++++++++++ 34 files changed, 653 insertions(+), 54 deletions(-) create mode 100644 tests/unit_tests/test_multiple_workers_contract.py diff --git a/fern/versions/latest/pages/environment-tutorials/resources-server-apis.mdx b/fern/versions/latest/pages/environment-tutorials/resources-server-apis.mdx index 36037655bee..a9d195c11a5 100644 --- a/fern/versions/latest/pages/environment-tutorials/resources-server-apis.mdx +++ b/fern/versions/latest/pages/environment-tutorials/resources-server-apis.mdx @@ -448,7 +448,7 @@ elif is_nemo_gym_fastapi_entrypoint(__file__): app = StatefulServer.run_webserver() # noqa: F401 ``` -A server whose state routing cannot reach, such as a registry keyed by another ID in the request body, overrides `multiple_workers_unsupported_reason()` instead and returns why it needs one worker. `run_webserver` then refuses `num_workers` above 1 with that reason before Uvicorn starts, and a dry run reports it too. +A server whose state routing cannot reach, such as a registry keyed by another ID in the request body, overrides `multiple_workers_unsupported_reason()` instead and returns why it needs one worker. `run_webserver` then refuses `num_workers` above 1 with that reason before Uvicorn starts, and a dry run reports it too. A unit test checks that every server does one or the other. A server that keeps no per-session state, or shares it outside the process, can opt out of routing with `routes_sessions_to_owner = False` on its class. diff --git a/resources_servers/agentic_if/app.py b/resources_servers/agentic_if/app.py index c6ec6429c76..f38634a34be 100644 --- a/resources_servers/agentic_if/app.py +++ b/resources_servers/agentic_if/app.py @@ -220,10 +220,11 @@ class AgenticIFResourcesServer(SimpleResourcesServer): _grade_jobs: set[asyncio.Task[IFResult]] = PrivateAttr(default_factory=set) _shutting_down: bool = PrivateAttr(default=False) + def multiple_workers_unsupported_reason(self) -> str | None: + return "agentic IF sessions hold sandbox connections in one process" + def model_post_init(self, context: Any) -> None: super().model_post_init(context) - if self.config.num_workers not in (None, 1): - raise ValueError("agentic IF sessions hold sandbox connections in one process: num_workers must be 1") if missing := [str(path) for path in self.config.opencode_runtime_files if not path.is_file()]: raise ValueError( f"opencode_runtime_files are missing: {missing}; for the benchmark, run " diff --git a/resources_servers/agentic_if/tests/test_app.py b/resources_servers/agentic_if/tests/test_app.py index 01c30f295c3..9a735ef7a49 100644 --- a/resources_servers/agentic_if/tests/test_app.py +++ b/resources_servers/agentic_if/tests/test_app.py @@ -851,8 +851,10 @@ async def test_a_session_less_verify_only_reads_its_own_persisted_episode(world, def test_a_session_server_refuses_more_than_one_worker(tmp_path): - with pytest.raises(ValueError, match="num_workers"): - make_server(tmp_path, Backends(), num_workers=2) + from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers + + with pytest.raises(ValueError, match="needs num_workers=1: agentic IF sessions hold sandbox connections"): + refuse_unsupported_num_workers(make_server(tmp_path, Backends(), num_workers=2)) def test_the_server_declares_stateless_reverification_and_no_ray(): diff --git a/resources_servers/aviary/app.py b/resources_servers/aviary/app.py index f1564a26a4f..0dfbb22e3c1 100644 --- a/resources_servers/aviary/app.py +++ b/resources_servers/aviary/app.py @@ -98,6 +98,13 @@ class AviaryResourcesServer(SimpleResourcesServer, Generic[TEnv, TDataset], ABC) env_id_to_env: dict[str, TEnv] = Field(default_factory=dict) env_id_to_total_reward: dict[str, float] = Field(default_factory=lambda: defaultdict(float)) + def multiple_workers_unsupported_reason(self) -> str | None: + return ( + "environments are kept in this process's memory by the env_id in the request body, " + "and the aviary agent sends its step, verify and close calls without the session cookie, " + "so they can reach a worker that does not hold the environment" + ) + def setup_webserver(self) -> FastAPI: app = super().setup_webserver() app.post("/step")(self.step) diff --git a/resources_servers/critpt/app.py b/resources_servers/critpt/app.py index 11825596f2f..be7e60ce5ad 100644 --- a/resources_servers/critpt/app.py +++ b/resources_servers/critpt/app.py @@ -236,6 +236,12 @@ class CritPtVerifyResponse(BaseVerifyResponse): class CritPtResourcesServer(SimpleResourcesServer): config: CritPtResourcesServerConfig + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return ( + "verify calls are pooled in this process's memory into batches of distinct problems that are submitted together, " + "and the batches, submission counters and API key rotation must be shared by every request" + ) + def model_post_init(self, context: Any) -> None: super().model_post_init(context) self._lock = asyncio.Lock() diff --git a/resources_servers/enterpriseops_gym/app.py b/resources_servers/enterpriseops_gym/app.py index 8220f9c7b8b..41fe06d10fe 100644 --- a/resources_servers/enterpriseops_gym/app.py +++ b/resources_servers/enterpriseops_gym/app.py @@ -192,10 +192,11 @@ class EnterpriseOpsGymResourcesServer(SimpleResourcesServer): def model_post_init(self, context: Any) -> None: super().model_post_init(context) - if self.config.num_workers not in (None, 1): - raise ValueError("EnterpriseOps-Gym sessions are process-local and require num_workers=1") self._fence = ResourcesSessionFence(self.config.session_close_retry_window_seconds) + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return "EnterpriseOps-Gym sessions are process-local" + def setup_webserver(self) -> FastAPI: app = super().setup_webserver() parent_lifespan = app.router.lifespan_context diff --git a/resources_servers/enterpriseops_gym/tests/test_app.py b/resources_servers/enterpriseops_gym/tests/test_app.py index f2387b1dba1..e5a87c7375e 100644 --- a/resources_servers/enterpriseops_gym/tests/test_app.py +++ b/resources_servers/enterpriseops_gym/tests/test_app.py @@ -20,6 +20,7 @@ ResourcesSeedSessionRequest, check_resources_session_contract, ) +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from resources_servers.enterpriseops_gym import app as app_module from resources_servers.enterpriseops_gym.app import ( EnterpriseOpsGymResourcesServer, @@ -567,5 +568,5 @@ def test_shutdown_logs_a_sandbox_it_cannot_stop(archive, gyms, sandboxes, caplog def test_sessions_live_in_one_worker(archive): - with pytest.raises(ValueError, match="num_workers=1"): - _server(archive, num_workers=2) + with pytest.raises(ValueError, match="needs num_workers=1: EnterpriseOps-Gym sessions are process-local"): + refuse_unsupported_num_workers(_server(archive, num_workers=2)) diff --git a/resources_servers/gdpval/app.py b/resources_servers/gdpval/app.py index 2b2e48002e7..f4ef7401616 100644 --- a/resources_servers/gdpval/app.py +++ b/resources_servers/gdpval/app.py @@ -73,6 +73,7 @@ SimpleResourcesServer, lost_session_verification, ) +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from resources_servers.gdpval import persisted_layout from resources_servers.gdpval.judge_panel import ( ResolvedJudge, @@ -620,12 +621,20 @@ class GDPValResourcesServer(SimpleResourcesServer): config: GDPValResourcesServerConfig + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return ( + "a seed claims the persisted deliverables directory of its task and repeat in this process's memory, " + "keyed by the task ID and rollout ID in the request body, " + "so sessions on different workers could write the same directory" + ) + def model_post_init(self, context: Any) -> None: + # Refuse several workers before installing LibreOffice and the other task dependencies, + # so the worker-count error is the one reported. + refuse_unsupported_num_workers(self) self._judge_prompt_fpath: str = self.config.judge_prompt_template_fpath or _DEFAULT_JUDGE_PROMPT_FPATH self._judge_telemetry = JudgeTelemetrySink(self.config.judge_telemetry_output_dir) self._tavily = TavilySearch(api_keys=self.config.tavily_api_key, max_sweeps=self.config.tavily_max_sweeps) - if self.config.num_workers not in (None, 1): - raise ValueError("GDPVal sessions are process-local and require num_workers=1") self._sessions: Dict[str, _TaskSession] = {} # (task_id, repeat) of the persisted directory each live session will write, to the session holding it. self._claimed_directories: Dict[Tuple[str, int], str] = {} diff --git a/resources_servers/gdpval/tests/test_sessions.py b/resources_servers/gdpval/tests/test_sessions.py index 038cad4d6f2..51723f63180 100644 --- a/resources_servers/gdpval/tests/test_sessions.py +++ b/resources_servers/gdpval/tests/test_sessions.py @@ -555,9 +555,14 @@ def test_shutdown_stops_the_sandboxes_of_open_sessions(server, root, tmp_path, s assert len(stopped) == 1 -def test_sessions_require_a_single_worker(tmp_path): - with pytest.raises(ValueError, match="num_workers=1"): - _server(tmp_path, num_workers=2) +def test_sessions_require_a_single_worker(tmp_path, monkeypatch): + def install(*args, **kwargs): + raise AssertionError("LibreOffice was set up before the worker count was refused") + + monkeypatch.setattr("resources_servers.gdpval.setup_libreoffice.ensure_libreoffice", install) + # The refusal comes before dependency setup, as a dry run with several workers would see it. + with pytest.raises(ValueError, match="needs num_workers=1: a seed claims the persisted deliverables directory"): + _server(tmp_path, num_workers=2, libreoffice_command=["soffice"]) def _restart(client: TestClient, tmp_path: Path) -> TestClient: diff --git a/resources_servers/genrm_compare/app.py b/resources_servers/genrm_compare/app.py index 8f3235d0ba0..f7fa45a8c3b 100644 --- a/resources_servers/genrm_compare/app.py +++ b/resources_servers/genrm_compare/app.py @@ -61,6 +61,7 @@ ReverifyMode, SimpleResourcesServer, ) +from nemo_gym.utils import is_nemo_gym_fastapi_entrypoint from nemo_gym.verification import JUDGE_FAILURES, JudgeError, is_judge_failure from resources_servers.genrm_compare.utils import ( GenRMOutputParseError, @@ -286,14 +287,6 @@ class GenRMCompareConfig(BaseResourcesServerConfig): genrm_parse_retries: int = 3 genrm_parse_retry_sleep_s: float = 0.2 - @model_validator(mode="after") - def _validate_cohort_workers(self): - if (self.comparison_mode == "fixed_baseline" or self.num_rollouts_per_prompt > 1) and ( - self.num_workers or 1 - ) > 1: - raise ValueError("GenRM cohort verification requires one HTTP worker because group state is process-local") - return self - class GenRMCompareVerifyRequest(BaseVerifyRequest): """Verify request with optional principle for cohort-based GenRM comparison.""" @@ -428,6 +421,15 @@ class GenRMCompareResourcesServer(SimpleResourcesServer): _cohort_tasks: set[asyncio.Task] = PrivateAttr(default_factory=set) _closed: bool = PrivateAttr(default=False) + def multiple_workers_unsupported_reason(self) -> Optional[str]: + if self.config.comparison_mode == "rollout_cohort" and self.config.num_rollouts_per_prompt <= 1: + return None + return ( + "cohort verification (comparison_mode=fixed_baseline or num_rollouts_per_prompt > 1) collects " + "the rollouts of a prompt in this process's memory, keyed by the request body, " + "so a cohort split across workers never completes" + ) + def _own_task(self, coro, *, name: str) -> asyncio.Task: task = asyncio.create_task(coro, name=name, context=Context()) self._cohort_tasks.add(task) @@ -1731,3 +1733,6 @@ async def call(): if __name__ == "__main__": GenRMCompareResourcesServer.run_webserver() +elif is_nemo_gym_fastapi_entrypoint(__file__): + # With num_workers > 1, uvicorn imports this module in each worker and serves its module-level `app`. + app = GenRMCompareResourcesServer.run_webserver() # noqa: F401 diff --git a/resources_servers/genrm_compare/tests/test_review_fixes.py b/resources_servers/genrm_compare/tests/test_review_fixes.py index 323a21a9aaa..6ccad159316 100644 --- a/resources_servers/genrm_compare/tests/test_review_fixes.py +++ b/resources_servers/genrm_compare/tests/test_review_fixes.py @@ -23,7 +23,6 @@ from aiohttp import ClientConnectionError, ClientPayloadError, ClientResponseError, RequestInfo from fastapi import HTTPException from multidict import CIMultiDict, CIMultiDictProxy -from pydantic import ValidationError from yarl import URL import resources_servers.genrm_compare.app as genrm @@ -36,6 +35,7 @@ ) from nemo_gym.data import TaskDataValidator, load_task_data_schema from nemo_gym.metrics import RewardProfiler +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from nemo_gym.verification import judge_failsafe from resources_servers.genrm_compare.tests.support import request_with_session from resources_servers.genrm_compare.tests.test_cohort_lifecycle import collected_comparisons, member @@ -502,14 +502,22 @@ async def test_reordered_rubric_ids_share_cohort_and_completed_replay( assert error.value.status_code == 409 -@pytest.mark.parametrize("mode,allowed", [("fixed_baseline", False), ("rollout_cohort", True)]) -def test_singleton_worker_guard(config: genrm.GenRMCompareConfig, mode: str, allowed: bool) -> None: - values = config.model_dump() | {"comparison_mode": mode, "num_rollouts_per_prompt": 1, "num_workers": 2} +@pytest.mark.parametrize( + "mode,rollouts,allowed", + [("fixed_baseline", 1, False), ("rollout_cohort", 2, False), ("rollout_cohort", 1, True)], +) +def test_several_workers_are_refused_only_for_cohort_verification( + config: genrm.GenRMCompareConfig, mode: str, rollouts: int, allowed: bool +) -> None: + values = config.model_dump() | {"comparison_mode": mode, "num_rollouts_per_prompt": rollouts, "num_workers": 2} + server = genrm.GenRMCompareResourcesServer.model_construct( + config=genrm.GenRMCompareConfig.model_validate(values), server_client=MagicMock() + ) if allowed: - assert genrm.GenRMCompareConfig.model_validate(values).num_workers == 2 + refuse_unsupported_num_workers(server) else: - with pytest.raises(ValidationError, match="one HTTP worker"): - genrm.GenRMCompareConfig.model_validate(values) + with pytest.raises(ValueError, match="needs num_workers=1: cohort verification"): + refuse_unsupported_num_workers(server) @pytest.mark.parametrize("ids,clean", [([1, 2], True), ("1,2", False)]) diff --git a/resources_servers/legal_agent_bench/harbor_bridge.py b/resources_servers/legal_agent_bench/harbor_bridge.py index 039bf170391..841e6401217 100644 --- a/resources_servers/legal_agent_bench/harbor_bridge.py +++ b/resources_servers/legal_agent_bench/harbor_bridge.py @@ -20,6 +20,12 @@ class LegalAgentBenchHarborBridge(HarborAgent): ray_enabled = True + def multiple_workers_unsupported_reason(self) -> str | None: + return ( + "this entrypoint lives outside the harbor_agent directory, " + "so uvicorn workers cannot import it by the module path Gym gives them" + ) + def setup_webserver(self) -> FastAPI: app = super().setup_webserver() app.post("/aggregate_metrics")(self.aggregate_metrics) diff --git a/resources_servers/longmt_eval/app.py b/resources_servers/longmt_eval/app.py index 141acdbaaa7..5691195daf1 100644 --- a/resources_servers/longmt_eval/app.py +++ b/resources_servers/longmt_eval/app.py @@ -141,6 +141,12 @@ class LongmtEvalServer(SimpleResourcesServer): ray_enabled = True config: LongmtEvalConfig + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return ( + "each worker builds its own Ray GPU actor pool, " + "so several workers multiply the GPU actors and can fail to schedule them" + ) + _segale_actors: List[Any] = PrivateAttr(default_factory=list) _actor_lock: Any = PrivateAttr(default=None) _actor_idx: int = PrivateAttr(default=0) diff --git a/resources_servers/ns_tools/app.py b/resources_servers/ns_tools/app.py index cf883e8d307..6179bcf8b8c 100644 --- a/resources_servers/ns_tools/app.py +++ b/resources_servers/ns_tools/app.py @@ -48,6 +48,7 @@ BaseVerifyResponse, SimpleResourcesServer, ) +from nemo_gym.utils import is_nemo_gym_fastapi_entrypoint logger = logging.getLogger(__name__) @@ -133,6 +134,14 @@ class NSToolsResourcesServer(SimpleResourcesServer): _timing_by_session: Dict[str, list] = {} # session_id -> list of timing records _uses_python_tool_sidecar: bool = False + def multiple_workers_unsupported_reason(self) -> Optional[str]: + if not any(self._tool_uses_python_tool_sidecar(tool_spec) for tool_spec in self.config.nemo_skills_tools): + return None + return ( + "the HTTP PythonTool starts a sidecar on the fixed python_tool_port in every worker, " + "so a second worker cannot bind it" + ) + def setup_webserver(self) -> FastAPI: app = super().setup_webserver() @@ -485,3 +494,6 @@ async def shutdown(self): if __name__ == "__main__": NSToolsResourcesServer.run_webserver() +elif is_nemo_gym_fastapi_entrypoint(__file__): + # With num_workers > 1, uvicorn imports this module in each worker and serves its module-level `app`. + app = NSToolsResourcesServer.run_webserver() # noqa: F401 diff --git a/resources_servers/ns_tools/tests/test_app.py b/resources_servers/ns_tools/tests/test_app.py index 986795d90f9..f1874b5f5a9 100644 --- a/resources_servers/ns_tools/tests/test_app.py +++ b/resources_servers/ns_tools/tests/test_app.py @@ -29,6 +29,7 @@ from nemo_gym.config import ResourcesServerRef from nemo_gym.core import NeMoGymResponse from nemo_gym.server_runtime import SimpleResourcesServer +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers class TestApp: @@ -42,6 +43,24 @@ def test_sanity(self) -> None: ) NSToolsResourcesServer(config=config, server_client=MagicMock(spec=ServerClient)) + @pytest.mark.parametrize("uses_sidecar", [True, False]) + def test_several_workers_are_refused_only_with_the_http_python_tool(self, uses_sidecar: bool) -> None: + config = NSToolsConfig( + host="0.0.0.0", + port=8080, + entrypoint="", + name="ns_tools", + num_workers=2, + nemo_skills_tools=["nemo_skills.mcp.servers.python_tool::PythonTool"], + ) + server = NSToolsResourcesServer(config=config, server_client=MagicMock(spec=ServerClient)) + with patch.object(NSToolsResourcesServer, "_tool_uses_python_tool_sidecar", return_value=uses_sidecar): + if uses_sidecar: + with pytest.raises(ValueError, match="needs num_workers=1: the HTTP PythonTool starts a sidecar"): + refuse_unsupported_num_workers(server) + else: + refuse_unsupported_num_workers(server) + def test_config_with_verifiers(self) -> None: """Test configuration with verifiers.""" verifiers = { diff --git a/resources_servers/osworld/app.py b/resources_servers/osworld/app.py index 503b74bf7da..99beee217a2 100644 --- a/resources_servers/osworld/app.py +++ b/resources_servers/osworld/app.py @@ -42,6 +42,13 @@ class OSWorldResourcesServer(SimpleResourcesServer): config: OSWorldResourcesServerConfig _manager: OSWorldSessionManager = PrivateAttr() + def multiple_workers_unsupported_reason(self) -> str | None: + return ( + "remote Docker worker capacity, startup cleanup of orphaned containers and the sessions.json state file " + "are each owned by one process, so several workers would overfill the Docker hosts, " + "remove each other's containers and overwrite each other's state file" + ) + def model_post_init(self, _context) -> None: self._manager = OSWorldSessionManager(self.config) diff --git a/resources_servers/osworld/config.py b/resources_servers/osworld/config.py index 1b903895e31..a363b4bd51f 100644 --- a/resources_servers/osworld/config.py +++ b/resources_servers/osworld/config.py @@ -105,10 +105,6 @@ def validate_resources_server(self) -> "OSWorldResourcesServerConfig": raise ValueError("worker names must be unique") if self.enable_proxy and not self.proxy_config_file: raise ValueError("proxy_config_file is required when enable_proxy=true") - if self.num_workers not in {None, 1}: - raise ValueError( - "OSWorld Resources Server keeps live DesktopEnv objects in process; num_workers must be 1" - ) return self @property diff --git a/resources_servers/partner_if/mock_tools_app.py b/resources_servers/partner_if/mock_tools_app.py index 9aba62e399e..dc9e89621cc 100644 --- a/resources_servers/partner_if/mock_tools_app.py +++ b/resources_servers/partner_if/mock_tools_app.py @@ -37,6 +37,7 @@ ReverifyMode, SimpleResourcesServer, ) +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from nemo_gym.utils import is_nemo_gym_fastapi_entrypoint from nemo_gym.verification import VerifierFixture from resources_servers.partner_if.app import ( @@ -186,10 +187,14 @@ class MockToolsResourcesServer(PartnerIFGradingMixin, PartnerIFMetricsMixin, Sim _session_locks: dict[str | None, asyncio.Lock] = PrivateAttr(default_factory=dict) _closed: ClosedResourcesSessionRecords[EpisodeId | None] = PrivateAttr() + def multiple_workers_unsupported_reason(self) -> str | None: + return "mock-tool episodes live in the server's memory" + def model_post_init(self, context: object) -> None: + # Refuse several workers before loading the tool backend and the task file, + # so the worker-count error is the one reported. + refuse_unsupported_num_workers(self) super().model_post_init(context) - if self.config.num_workers not in (None, 1): - raise ValueError("Mock-tool episodes live in the server's memory, so it requires num_workers=1") self._closed = ClosedResourcesSessionRecords(self.config.session_close_retry_window_seconds) if self.config.backend is not None: self._backend = load_backend(self.config.backend)(self.config.fixture_path) diff --git a/resources_servers/partner_if/tests/test_mock_tools_app.py b/resources_servers/partner_if/tests/test_mock_tools_app.py index d58b79d6193..6d71a103e47 100644 --- a/resources_servers/partner_if/tests/test_mock_tools_app.py +++ b/resources_servers/partner_if/tests/test_mock_tools_app.py @@ -571,7 +571,7 @@ def test_an_oversized_artifact_is_a_capture_error_for_the_items_that_need_it(ser def test_episodes_live_in_one_worker_and_the_server_never_joins_ray(tmp_path, plugin): for workers in (None, 1): make_server(tmp_path, num_workers=workers) - with pytest.raises(ValueError, match="num_workers=1"): + with pytest.raises(ValueError, match="needs num_workers=1: mock-tool episodes live in the server's memory"): make_server(tmp_path, num_workers=2) # Declared on the class the entrypoint runs, where Gym reads it before it decides whether to start Ray. assert api.MockToolsResourcesServer.ray_enabled is False diff --git a/resources_servers/proof_judge/app.py b/resources_servers/proof_judge/app.py index ae8c7544bc5..8caa2ea477a 100644 --- a/resources_servers/proof_judge/app.py +++ b/resources_servers/proof_judge/app.py @@ -10,6 +10,7 @@ # 2. External: reads JUDGE_SERVER_ARGS env var → AsyncOpenAI /v1/chat/completions # (judge servers managed by nemo-skill pipeline as SLURM het groups) import asyncio +import fcntl import json import logging import os @@ -37,6 +38,7 @@ BaseVerifyResponse, SimpleResourcesServer, ) +from nemo_gym.utils import is_nemo_gym_fastapi_entrypoint from nemo_gym.verification import reraise_judge_errors @@ -233,6 +235,14 @@ class ProofWithJudgeResourcesServer(SimpleResourcesServer): default_factory=lambda: defaultdict(IncorrectGroupCoordinator) ) + def multiple_workers_unsupported_reason(self) -> Optional[str]: + if not self.config.zero_reward_incorrect_groups: + return None + return ( + "zero_reward_incorrect_groups waits in this process's memory for every rollout of a problem, " + "keyed by the problem text, so a group split across workers never completes" + ) + async def verify(self, body: ProofWithJudgeVerifyRequest) -> ProofWithJudgeVerifyResponse: problem = getattr(body, "problem", "") or (body.model_dump().get("problem") or "") full_response = self._extract_assistant_text(body.response) @@ -274,6 +284,16 @@ async def _append_log_jsonl( } async with self._log_lock: with open(log_path, "a", encoding="utf-8") as f: + # The asyncio lock covers this worker only. + # With num_workers > 1, every worker appends to this file, + # and a long record can reach it in several writes. + # The file lock keeps records from different workers from interleaving. + # A filesystem without file locks (NFS without lockd, Lustre mounted noflock) raises here; + # the record is still written, unlocked, as it was before workers shared the file. + try: + fcntl.flock(f, fcntl.LOCK_EX) + except OSError: + pass f.write(json.dumps(record, ensure_ascii=False) + "\n") except Exception as e: LOG.warning("[proof_judge] Failed to append log_jsonl %s: %s", log_path, e) @@ -492,3 +512,6 @@ async def _judge_single(self, problem: str, full_response: str) -> tuple[float, if __name__ == "__main__": ProofWithJudgeResourcesServer.run_webserver() +elif is_nemo_gym_fastapi_entrypoint(__file__): + # With num_workers > 1, uvicorn imports this module in each worker and serves its module-level `app`. + app = ProofWithJudgeResourcesServer.run_webserver() # noqa: F401 diff --git a/resources_servers/proof_judge/tests/test_app.py b/resources_servers/proof_judge/tests/test_app.py index ecda6415dfc..40d253986ee 100644 --- a/resources_servers/proof_judge/tests/test_app.py +++ b/resources_servers/proof_judge/tests/test_app.py @@ -1,16 +1,35 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio +import fcntl +import json +import threading +import time from unittest.mock import MagicMock +import pytest + from nemo_gym.client import ServerClient from nemo_gym.config import ModelServerRef +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from resources_servers.proof_judge.app import ( ProofWithJudgeResourcesServer, ProofWithJudgeResourcesServerConfig, ) +def _config(**overrides) -> ProofWithJudgeResourcesServerConfig: + return ProofWithJudgeResourcesServerConfig( + host="0.0.0.0", + port=8080, + entrypoint="", + name="proof_judge", + judge_model_server=ModelServerRef(type="responses_api_models", name="judge"), + **overrides, + ) + + class TestApp: def test_sanity(self) -> None: config = ProofWithJudgeResourcesServerConfig( @@ -21,3 +40,57 @@ def test_sanity(self) -> None: judge_model_server=ModelServerRef(type="responses_api_models", name="judge"), ) ProofWithJudgeResourcesServer(config=config, server_client=MagicMock(spec=ServerClient)) + + def test_log_appends_wait_for_another_workers_lock(self, tmp_path) -> None: + """With several workers appending to one debug log, an append waits for a lock another process holds.""" + server = ProofWithJudgeResourcesServer(config=_config(), server_client=MagicMock(spec=ServerClient)) + log = tmp_path / "log.jsonl" + other_worker = open(log, "a") + fcntl.flock(other_worker, fcntl.LOCK_EX) + released_at: list[float] = [] + + def release() -> None: + time.sleep(0.3) + released_at.append(time.monotonic()) + other_worker.close() + + releaser = threading.Thread(target=release) + releaser.start() + asyncio.run( + server._append_log_jsonl(log_path=str(log), problem="p", generated_sequence="g", reward=1.0, details={}) + ) + appended_at = time.monotonic() + releaser.join() + + assert appended_at >= released_at[0] + assert json.loads(log.read_text())["reward"] == 1.0 + + def test_several_workers_are_refused_only_for_group_rewards(self) -> None: + grouped = ProofWithJudgeResourcesServer( + config=_config(num_workers=2, zero_reward_incorrect_groups=True, expected_group_size=4), + server_client=MagicMock(spec=ServerClient), + ) + with pytest.raises(ValueError, match="needs num_workers=1: zero_reward_incorrect_groups waits"): + refuse_unsupported_num_workers(grouped) + + stateless = ProofWithJudgeResourcesServer( + config=_config(num_workers=2), server_client=MagicMock(spec=ServerClient) + ) + refuse_unsupported_num_workers(stateless) + + +def test_log_appends_without_file_locks_still_write_the_record(tmp_path, monkeypatch) -> None: + """A filesystem without file locks makes flock raise; the record is written anyway.""" + import errno + + def no_locks(*args) -> None: + raise OSError(errno.ENOLCK, "No locks available") + + monkeypatch.setattr(fcntl, "flock", no_locks) + server = ProofWithJudgeResourcesServer(config=_config(), server_client=MagicMock(spec=ServerClient)) + log = tmp_path / "log.jsonl" + asyncio.run( + server._append_log_jsonl(log_path=str(log), problem="p", generated_sequence="g", reward=1.0, details={}) + ) + + assert json.loads(log.read_text())["reward"] == 1.0 diff --git a/resources_servers/speed_bench/app.py b/resources_servers/speed_bench/app.py index 18ecbce1e52..ea17dda1f37 100644 --- a/resources_servers/speed_bench/app.py +++ b/resources_servers/speed_bench/app.py @@ -41,6 +41,8 @@ BaseVerifyResponse, SimpleResourcesServer, ) +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers +from nemo_gym.utils import is_nemo_gym_fastapi_entrypoint LOG = logging.getLogger(__name__) @@ -488,6 +490,16 @@ def _compute_std_metrics( class SpeedBenchResourcesServer(SimpleResourcesServer): config: SpeedBenchResourcesServerConfig + def multiple_workers_unsupported_reason(self) -> Optional[str]: + # Only the Prometheus source measures against a snapshot; the response source scores each request on its own. + if self.config.metrics_source != "prometheus": + return None + return ( + "with metrics_source=prometheus, the speculative decoding measurement window starts from a metrics " + "snapshot kept in this process's memory, so each worker would measure from a different start and report " + "deltas that miss the other workers' traffic" + ) + # State populated lazily on the first verify() call so the benchmark window # is bounded by the first model traffic our server sees, not by server # init time. Wrapped in an asyncio.Lock to avoid two concurrent first calls @@ -498,6 +510,9 @@ class SpeedBenchResourcesServer(SimpleResourcesServer): _init_lock: Optional[asyncio.Lock] = None def model_post_init(self, __context: Any) -> None: + # Refuse several workers before taking the startup metrics snapshot, + # so the worker-count error is the one reported. + refuse_unsupported_num_workers(self) super().model_post_init(__context) self._init_lock = asyncio.Lock() if self.config.metrics_source == "prometheus" and self.config.snapshot_at_init: @@ -741,3 +756,6 @@ def get_key_metrics(self, agent_metrics: Dict[str, Any]) -> Dict[str, Any]: if __name__ == "__main__": SpeedBenchResourcesServer.run_webserver() +elif is_nemo_gym_fastapi_entrypoint(__file__): + # With num_workers > 1, uvicorn imports this module in each worker and serves its module-level `app`. + app = SpeedBenchResourcesServer.run_webserver() # noqa: F401 diff --git a/resources_servers/speed_bench/tests/test_app.py b/resources_servers/speed_bench/tests/test_app.py index f2399a8f097..b0d1a9f2ceb 100644 --- a/resources_servers/speed_bench/tests/test_app.py +++ b/resources_servers/speed_bench/tests/test_app.py @@ -41,6 +41,7 @@ ) from nemo_gym.client import ServerClient +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers # ────────────────────────────────────────────────────────────────────────────── @@ -804,3 +805,9 @@ def test_compute_metrics_omits_std_metrics_when_single_seed(): m = s.compute_metrics(tasks) assert "spec_acceptance_length_avg" not in m assert "spec_acceptance_length_std_dev_across_runs" not in m + + +def test_several_workers_are_refused_only_for_the_prometheus_source(): + with pytest.raises(ValueError, match="needs num_workers=1: with metrics_source=prometheus"): + _make_server(num_workers=2) + refuse_unsupported_num_workers(_make_server(num_workers=2, metrics_source="response")) diff --git a/resources_servers/terminal_bench_4/app.py b/resources_servers/terminal_bench_4/app.py index 9acb2192b58..ea519f1bf9b 100644 --- a/resources_servers/terminal_bench_4/app.py +++ b/resources_servers/terminal_bench_4/app.py @@ -13,7 +13,7 @@ from datetime import datetime, timedelta, timezone from pathlib import Path, PurePosixPath from time import monotonic -from typing import ClassVar, Literal +from typing import ClassVar from uuid import uuid4 from fastapi import HTTPException, Request @@ -69,7 +69,6 @@ class TerminalBench4Config(BaseResourcesServerConfig): - num_workers: Literal[1] = 1 REVERIFY_MODE: ClassVar[ReverifyMode] = ReverifyMode.UNSUPPORTED manifest_path: Path = BENCHMARK / "manifest.json" artifacts_dir: Path = Path("results/terminal_bench_4/resources") @@ -185,6 +184,13 @@ def _scored_verifier_failure(diagnostics) -> str | None: class TerminalBench4ResourcesServer(SimpleResourcesServer): config: TerminalBench4Config + def multiple_workers_unsupported_reason(self) -> str | None: + return ( + "seeds are deduplicated in this process's memory by the client_session_id and rollout_id " + "in the request body, which session routing does not route, " + "so a retried seed without the session cookie can reach another worker and provision a second sandbox" + ) + def model_post_init(self, context): super().model_post_init(context) manifest_path = self.config.manifest_path diff --git a/resources_servers/toolsandbox/app.py b/resources_servers/toolsandbox/app.py index 2991175a998..7bf25c57f4b 100644 --- a/resources_servers/toolsandbox/app.py +++ b/resources_servers/toolsandbox/app.py @@ -190,6 +190,13 @@ class ToolSandboxResourcesServer(SimpleResourcesServer): # cached at /close time. `reward` is the combined milestone similarity. scoring: Dict[str, Dict[str, Any]] = Field(default_factory=dict) + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return ( + "episodes are kept in this process's memory by the env_id in the request body, " + "and toolsandbox_agent sends its step, close and verify calls without the session cookie, " + "so they can reach a worker that does not hold the episode" + ) + # Deterministic (name, Scenario) list; built once, indexed by task_idx. _scenarios: Optional[List[Tuple[str, Scenario]]] = PrivateAttr(default=None) diff --git a/resources_servers/vlm_eval_kit/app.py b/resources_servers/vlm_eval_kit/app.py index fd947c9382d..05adbf44d0a 100644 --- a/resources_servers/vlm_eval_kit/app.py +++ b/resources_servers/vlm_eval_kit/app.py @@ -16,7 +16,7 @@ from collections import defaultdict from pathlib import Path from subprocess import run -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from pydantic import BaseModel, ConfigDict, Field @@ -57,6 +57,12 @@ class VlmEvalKitResourcesServer(SimpleResourcesServer): MMBench_DEV_EN_V11_sets: Dict[str, Coordinator] = Field(default_factory=lambda: defaultdict(Coordinator)) + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return ( + "MMBench scoring waits in this process's memory for every row of a group, " + "keyed by the group in the request body, so a group split across workers never completes" + ) + def setup_webserver(self): self.setup_VLMEvalKit() diff --git a/resources_servers/vlm_eval_kit/tests/test_app.py b/resources_servers/vlm_eval_kit/tests/test_app.py index e96826ef979..3a464574b50 100644 --- a/resources_servers/vlm_eval_kit/tests/test_app.py +++ b/resources_servers/vlm_eval_kit/tests/test_app.py @@ -14,7 +14,10 @@ # limitations under the License. from unittest.mock import MagicMock +import pytest + from nemo_gym.client import ServerClient +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from resources_servers.vlm_eval_kit.app import VlmEvalKitResourcesServer, VlmEvalKitResourcesServerConfig @@ -27,3 +30,9 @@ def test_sanity(self) -> None: name="", ) VlmEvalKitResourcesServer(config=config, server_client=MagicMock(spec=ServerClient)) + + def test_several_workers_are_refused(self) -> None: + config = VlmEvalKitResourcesServerConfig(host="0.0.0.0", port=8080, entrypoint="", name="", num_workers=2) + server = VlmEvalKitResourcesServer(config=config, server_client=MagicMock(spec=ServerClient)) + with pytest.raises(ValueError, match="needs num_workers=1: MMBench scoring waits"): + refuse_unsupported_num_workers(server) diff --git a/resources_servers/voicebench/app.py b/resources_servers/voicebench/app.py index 7bb0372555d..bcd845fc2ec 100644 --- a/resources_servers/voicebench/app.py +++ b/resources_servers/voicebench/app.py @@ -10,7 +10,7 @@ import random from collections.abc import Mapping, Sequence from pathlib import Path -from typing import ClassVar, cast +from typing import ClassVar, Optional, cast from pydantic import ( ConfigDict, @@ -31,6 +31,7 @@ ReverifyMode, SimpleResourcesServer, ) +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from resources_servers.voicebench.legacy import replay_rule_rows from resources_servers.voicebench.metrics import MetricValue, compute_voicebench_metrics from resources_servers.voicebench.scorers.base import ScoringResult @@ -56,8 +57,6 @@ def validate_profile(self) -> "VoiceBenchConfig": raise ValueError(f"subset {self.subset} requires scoring_type={SUBSET_SCORERS[self.subset]}") if self.scoring_version != "legacy_v1": raise ValueError("supported scoring_version is legacy_v1") - if self.num_workers not in (None, 1): - raise ValueError("legacy scoring requires one worker for the shared random stream") if self.scoring_type in ("open", "reference_qa"): if self.judge_model_server is None or self.judge_prompt_file is None: raise ValueError("judge scoring requires judge_model_server and judge_prompt_file") @@ -169,7 +168,13 @@ class VoiceBenchServer(SimpleResourcesServer): _expected_sample_ids: set[str] | None = PrivateAttr(default=None) _rule_rng: random.Random = PrivateAttr(default_factory=random.Random) + def multiple_workers_unsupported_reason(self) -> Optional[str]: + return "legacy scoring draws every rule-scored row from one random stream shared by the whole server" + def model_post_init(self, context: object) -> None: + # Refuse several workers before downloading the IFEval NLTK data, + # so the worker-count error is the one reported. + refuse_unsupported_num_workers(self) self._rule_rng = random.Random(self.config.seed) if self.config.expected_sample_ids_file: self._expected_sample_ids = _expected_ids(self.config.expected_sample_ids_file, self.config.subset) diff --git a/resources_servers/voicebench/tests/test_app.py b/resources_servers/voicebench/tests/test_app.py index 1cb25595555..2b5ff63f615 100644 --- a/resources_servers/voicebench/tests/test_app.py +++ b/resources_servers/voicebench/tests/test_app.py @@ -94,10 +94,19 @@ def test_types_reject_wrong_subset_scorer_and_missing_reference(tmp_path): request(subset="ifeval", instruction_id_list=["x"], kwargs=[]) with pytest.raises(ValidationError): request(subset="ifeval", instruction_id_list=["x"], kwargs=[None]) - with pytest.raises(ValidationError): + with pytest.raises(ValueError, match="needs num_workers=1: legacy scoring draws every rule-scored row"): server(tmp_path, num_workers=2) +def test_several_workers_are_refused_before_the_ifeval_data_is_downloaded(tmp_path, monkeypatch): + def download(data_dir): + raise AssertionError("NLTK data was downloaded before the worker count was refused") + + monkeypatch.setattr("resources_servers.voicebench.app._ensure_nltk", download) + with pytest.raises(ValueError, match="needs num_workers=1"): + server(tmp_path, subset="ifeval", scoring_type="ifeval", num_workers=2) + + def test_nltk_cache_existing_permissions_are_preserved_and_default_is_private(tmp_path): existing = tmp_path / "shared" existing.mkdir(mode=0o755) diff --git a/resources_servers/wmt_translation/app.py b/resources_servers/wmt_translation/app.py index e5465d81f0a..92b00bd0132 100644 --- a/resources_servers/wmt_translation/app.py +++ b/resources_servers/wmt_translation/app.py @@ -53,6 +53,7 @@ BaseVerifyResponse, SimpleResourcesServer, ) +from nemo_gym.utils import is_nemo_gym_fastapi_entrypoint LOG = logging.getLogger(__name__) @@ -317,6 +318,15 @@ class WmtTranslationResourcesServer(SimpleResourcesServer): ray_enabled = True config: WmtTranslationResourcesServerConfig + def multiple_workers_unsupported_reason(self) -> Optional[str]: + # Without COMET, scoring builds no actor pool and keeps nothing between requests. + if not self.config.compute_comet: + return None + return ( + "with compute_comet, each worker builds its own Ray GPU actor pool for COMET scoring, " + "so several workers multiply the GPU actors and can fail to schedule them" + ) + # COMET actor pool state — populated lazily during compute_metrics() so # actor creation happens after Ray is fully up and `extra_gpu` is # advertised. Pydantic PrivateAttr keeps these out of the config schema. @@ -917,3 +927,6 @@ def get_key_metrics(self, agent_metrics: Dict[str, Any]) -> Dict[str, Any]: if __name__ == "__main__": WmtTranslationResourcesServer.run_webserver() +elif is_nemo_gym_fastapi_entrypoint(__file__): + # With num_workers > 1, uvicorn imports this module in each worker and serves its module-level `app`. + app = WmtTranslationResourcesServer.run_webserver() # noqa: F401 diff --git a/resources_servers/wmt_translation/tests/test_app.py b/resources_servers/wmt_translation/tests/test_app.py index 978bdf9bc84..6e8aa270fd1 100644 --- a/resources_servers/wmt_translation/tests/test_app.py +++ b/resources_servers/wmt_translation/tests/test_app.py @@ -33,6 +33,7 @@ from nemo_gym.client import ServerClient from nemo_gym.core import NeMoGymResponse +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers def _make_response(text: str) -> NeMoGymResponse: @@ -1152,3 +1153,13 @@ def test_raises_if_sys_executable_missing(self, monkeypatch: pytest.MonkeyPatch, with pytest.raises(RuntimeError, match="sys.executable doesn't exist"): _build_comet_actor_class() + + +def test_several_workers_are_refused_only_with_comet() -> None: + comet = _make_server(compute_comet=True) + comet.config.num_workers = 2 + with pytest.raises(ValueError, match="needs num_workers=1: with compute_comet"): + refuse_unsupported_num_workers(comet) + no_comet = _make_server(compute_comet=False) + no_comet.config.num_workers = 2 + refuse_unsupported_num_workers(no_comet) diff --git a/responses_api_agents/terminus_2_sandboxed_agent/app.py b/responses_api_agents/terminus_2_sandboxed_agent/app.py index 65d118c8341..77f1cc42938 100644 --- a/responses_api_agents/terminus_2_sandboxed_agent/app.py +++ b/responses_api_agents/terminus_2_sandboxed_agent/app.py @@ -616,16 +616,19 @@ async def _summarize(self, *args, **kwargs): class Terminus2Agent(SimpleResponsesAPIAgent): config: Terminus2AgentConfig + def multiple_workers_unsupported_reason(self) -> Optional[str]: + # A deployment without a resources server exists only to serve episode sessions, + # so a hand-written config with several workers fails at `gym env start` rather than on every seed. + # A benchmark composition keeps the legacy /run route, where workers are independent. + if self.config.resources_server is not None: + return None + return "Terminus 2 episode sessions are process-local" + def model_post_init(self, context: Any, /) -> None: super().model_post_init(context) if self.config.resources_server is not None: _warn_deprecated_once() self._session_sandboxes: dict[str, AsyncSandbox] = {} - # Episode sessions are process-local. A deployment without a resources server exists only to serve - # sessions, so a hand-written config with several workers fails at `gym env start` rather than on - # every seed; a benchmark composition keeps the legacy /run route, where workers are independent. - if self.config.resources_server is None and self.config.num_workers not in (None, 1): - raise ValueError("Process-local Terminus 2 episode sessions require num_workers=1") if not self.config.debug: harbor_logger.setLevel(logging.WARNING) diff --git a/responses_api_agents/terminus_2_sandboxed_agent/tests/test_app.py b/responses_api_agents/terminus_2_sandboxed_agent/tests/test_app.py index 8e8c4357fe6..6714295320f 100644 --- a/responses_api_agents/terminus_2_sandboxed_agent/tests/test_app.py +++ b/responses_api_agents/terminus_2_sandboxed_agent/tests/test_app.py @@ -26,6 +26,7 @@ NeMoGymResponseOutputTokensDetails, NeMoGymResponseUsage, ) +from nemo_gym.server_runtime.base_server import refuse_unsupported_num_workers from nemo_gym.trajectory import TrajectoryRecord, TrajectoryTurn from nemo_gym.utils import IS_NEMO_GYM_FASTAPI_WORKER_KEY_NAME from responses_api_agents.terminus_2_sandboxed_agent import app as app_module @@ -684,10 +685,10 @@ def test_multiple_workers_are_refused_at_startup(self): # Session state is process-local: a hand-written config fails at `gym env start`, not per episode. config = _native_config() config.num_workers = 4 - with pytest.raises(ValueError, match="num_workers=1"): - Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient)) + with pytest.raises(ValueError, match="needs num_workers=1: Terminus 2 episode sessions are process-local"): + refuse_unsupported_num_workers(Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient))) config.num_workers = 1 - Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient)) + refuse_unsupported_num_workers(Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient))) def test_benchmark_composition_keeps_its_legacy_workers(self): """The shipped template (num_workers: 4) composed onto a benchmark still starts: /run is per request.""" @@ -718,11 +719,11 @@ def test_benchmark_composition_keeps_its_legacy_workers(self): ) assert config.num_workers == 4 assert config.resources_server.name == "terminal_bench_2_1_terminus_2_sandboxed_resources_server" - Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient)) + refuse_unsupported_num_workers(Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient))) # The same block without a resources server is session-only, and four workers are refused. config.resources_server = None - with pytest.raises(ValueError, match="num_workers=1"): - Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient)) + with pytest.raises(ValueError, match="needs num_workers=1"): + refuse_unsupported_num_workers(Terminus2Agent(config=config, server_client=MagicMock(spec=ServerClient))) def test_close_carries_the_runs_observations(self, monkeypatch): """seed -> responses -> close: the harness bundle (invocation record, compactions) rides on the close.""" diff --git a/tests/unit_tests/test_multiple_workers_contract.py b/tests/unit_tests/test_multiple_workers_contract.py new file mode 100644 index 00000000000..006acab88fc --- /dev/null +++ b/tests/unit_tests/test_multiple_workers_contract.py @@ -0,0 +1,312 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Every resources, agent and environment server either runs with several workers or refuses them. + +With `num_workers > 1`, `run_webserver` hands uvicorn the import string `:app`, +and each worker imports the entrypoint and serves its module-level `app`. +An entrypoint without that `app` makes every worker exit at startup. +A server that cannot run with several workers must say so instead, +by overriding `multiple_workers_unsupported_reason` so that `run_webserver` refuses before uvicorn starts. + +These checks read source only, +because each server runs in its own virtual environment and its dependencies are not installed here. +""" + +import ast +import warnings +from functools import cache +from pathlib import Path + +import pytest +import yaml + +from nemo_gym import PARENT_DIR + + +SERVER_KINDS = ("resources_servers", "responses_api_agents", "environment_servers") +REASON_METHOD = "multiple_workers_unsupported_reason" + +# Entrypoints that configs name but that start no server, so neither the adapter nor a refusal applies. +NOT_SERVERS = { + # The base class of Gymnasium-style servers; its config only declares the module so its tests run. + "resources_servers/gymnasium/base.py", + # The local SEC search library that finance_sec_search and finance_agent_v2 import; its config says so. + "resources_servers/sec_local_index/local_edgar_search.py", +} + + +def _parse(path: Path) -> ast.Module: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", SyntaxWarning) + return ast.parse(path.read_text()) + + +def _config_entrypoints(node: object, found: set[tuple[str, str, str]]) -> None: + if isinstance(node, dict): + for key, value in node.items(): + if key in SERVER_KINDS and isinstance(value, dict): + for server_dir, block in value.items(): + if isinstance(block, dict) and isinstance(block.get("entrypoint"), str): + found.add((key, server_dir, block["entrypoint"])) + _config_entrypoints(value, found) + elif isinstance(node, list): + for item in node: + _config_entrypoints(item, found) + + +def entrypoint_modules(root: Path) -> dict[Path, bool]: + """Every server entrypoint module, mapped to whether a config names it. + + That is each module a config names, and each `app.py` in a server directory, + since a server can also be started from a config outside the repository. + """ + root = root.resolve() + modules = {path: False for kind in SERVER_KINDS for path in (root / kind).glob("*/app.py")} + found: set[tuple[str, str, str]] = set() + # Configs live beside their servers and also under benchmarks/, environments/ and elsewhere. + for config in root.rglob("*.y*ml"): + if any(part.startswith(".") or part in ("node_modules", "cache") for part in config.relative_to(root).parts): + continue + try: + documents = list(yaml.safe_load_all(config.read_text())) + except (yaml.YAMLError, UnicodeDecodeError): + continue + for document in documents: + _config_entrypoints(document, found) + for kind, server_dir, entrypoint in found: + module = (root / kind / server_dir / entrypoint).resolve() + # Configs that name a server directory or module that does not exist cannot start, so they are skipped. + if module.suffix == ".py" and module.is_file(): + modules[module] = True + return { + module: named for module, named in sorted(modules.items()) if str(module.relative_to(root)) not in NOT_SERVERS + } + + +def _served_class_names(tree: ast.Module) -> set[str]: + return { + node.func.value.id + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "run_webserver" + and isinstance(node.func.value, ast.Name) + } + + +def has_adapter(tree: ast.Module) -> bool: + """Whether the module assigns a module-level `app` under `is_nemo_gym_fastapi_entrypoint(__file__)`.""" + branches = [node for node in tree.body if isinstance(node, ast.If)] + while branches: + branch = branches.pop() + branches.extend(node for node in branch.orelse if isinstance(node, ast.If)) + test = branch.test + if not ( + isinstance(test, ast.Call) + and isinstance(test.func, ast.Name) + and test.func.id == "is_nemo_gym_fastapi_entrypoint" + ): + continue + for statement in branch.body: + if isinstance(statement, ast.Assign) and any( + isinstance(target, ast.Name) and target.id == "app" for target in statement.targets + ): + return True + return False + + +@cache +def _module_tree(path: Path) -> ast.Module: + return _parse(path) + + +def _module_path(root: Path, module: str) -> Path | None: + base = root.joinpath(*module.split(".")) + for candidate in (base.with_suffix(".py"), base / "__init__.py"): + if candidate.is_file(): + return candidate + return None + + +def _find_class(root: Path, path: Path, name: str, seen: set[tuple[Path, str]]) -> tuple[Path, ast.ClassDef] | None: + """The class `name` refers to in the module at `path`, following imports within the repository.""" + if (path, name) in seen: + return None + seen.add((path, name)) + tree = _module_tree(path) + for node in tree.body: + if isinstance(node, ast.ClassDef) and node.name == name: + return path, node + for node in tree.body: + if not isinstance(node, ast.ImportFrom) or node.module is None or node.level: + continue + for alias in node.names: + if (alias.asname or alias.name) == name: + source = _module_path(root, node.module) + if source is not None: + return _find_class(root, source, alias.name, seen) + return None + + +def refuses_with_a_reason(root: Path, path: Path, name: str) -> bool: + """Whether the class, or a parent class among the servers in the repository, overrides the reason method. + + Parents in `nemo_gym` do not count: `SimpleServer` defines the method only to return None. + """ + found = _find_class(root, path, name, set()) + if found is None: + return False + class_path, class_node = found + if not any(class_path.is_relative_to(root / kind) for kind in SERVER_KINDS): + return False + if any(isinstance(item, ast.FunctionDef) and item.name == REASON_METHOD for item in class_node.body): + return True + for base in class_node.bases: + if isinstance(base, ast.Subscript): + base = base.value + if isinstance(base, ast.Name) and refuses_with_a_reason(root, class_path, base.id): + return True + return False + + +def contract_violations(root: Path) -> list[str]: + violations = [] + for path, named_by_config in entrypoint_modules(root).items(): + tree = _module_tree(path) + relative = path.relative_to(root.resolve()) + class_names = _served_class_names(tree) + if not class_names: + # An app.py that starts nothing is a library module; a module a config names must start a server. + if named_by_config: + violations.append( + f"{relative} is a configured entrypoint but calls no `.run_webserver()`, " + "so this check cannot tell which server it starts. Add it to NOT_SERVERS if it starts none." + ) + continue + if has_adapter(tree): + continue + for name in sorted(class_names): + if not refuses_with_a_reason(root.resolve(), path, name): + violations.append( + f"{relative}: {name} has neither the module-level `app` that multi-worker uvicorn imports " + f"nor a refusal. If the server is safe with several workers, end the module with\n" + f" elif is_nemo_gym_fastapi_entrypoint(__file__):\n" + f" app = {name}.run_webserver() # noqa: F401\n" + f"Otherwise override `{REASON_METHOD}` on {name} and return why it needs num_workers=1." + ) + return violations + + +def test_every_server_runs_with_several_workers_or_refuses_them() -> None: + violations = contract_violations(PARENT_DIR) + assert not violations, "\n\n".join(violations) + + +def test_the_scan_covers_entrypoints_outside_app_py() -> None: + modules = {str(path.relative_to(PARENT_DIR.resolve())) for path in entrypoint_modules(PARENT_DIR)} + + assert { + "responses_api_agents/langgraph_agent/rewoo_agent.py", + "responses_api_agents/langchain_deepagents_agent/reasoning_search_agent.py", + "resources_servers/aviary/gsm8k_app.py", + "resources_servers/aviary/notebook_app.py", + "resources_servers/legal_agent_bench/harbor_bridge.py", + } <= modules + assert not modules & NOT_SERVERS + + +def _write(root: Path, relative: str, source: str) -> None: + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source) + + +@pytest.fixture +def repo(tmp_path: Path) -> Path: + for kind in (*SERVER_KINDS, "benchmarks"): + (tmp_path / kind).mkdir() + return tmp_path + + +class TestContractChecker: + def test_server_without_adapter_or_refusal_is_reported(self, repo: Path) -> None: + _write(repo, "resources_servers/plain/app.py", "class Plain:\n pass\n\nPlain.run_webserver()\n") + + [violation] = contract_violations(repo) + + assert violation.startswith("resources_servers/plain/app.py: Plain has neither") + assert "app = Plain.run_webserver() # noqa: F401" in violation + assert REASON_METHOD in violation + + def test_adapter_satisfies_the_contract(self, repo: Path) -> None: + _write( + repo, + "responses_api_agents/safe/app.py", + "class Safe:\n pass\n\n" + "if __name__ == '__main__':\n Safe.run_webserver()\n" + "elif is_nemo_gym_fastapi_entrypoint(__file__):\n app = Safe.run_webserver()\n", + ) + + assert contract_violations(repo) == [] + + def test_refusal_inherited_from_a_parent_server_in_another_module(self, repo: Path) -> None: + _write( + repo, + "resources_servers/family/app.py", + "class Parent:\n def multiple_workers_unsupported_reason(self):\n return 'state'\n", + ) + _write( + repo, + "resources_servers/family/child_app.py", + "from resources_servers.family.app import Parent as Base\n\n" + "class Child(Base[int]):\n pass\n\nChild.run_webserver()\n", + ) + _write(repo, "resources_servers/family/configs/family.yaml", _config("family", "child_app.py")) + + assert contract_violations(repo) == [] + + def test_a_refusal_on_a_gym_base_class_does_not_count(self, repo: Path) -> None: + _write( + repo, + "nemo_gym/server_runtime/base_server.py", + "class SimpleServer:\n def multiple_workers_unsupported_reason(self):\n return None\n", + ) + _write( + repo, + "environment_servers/env/app.py", + "from nemo_gym.server_runtime.base_server import SimpleServer\n\n" + "class Env(SimpleServer):\n pass\n\nEnv.run_webserver()\n", + ) + + [violation] = contract_violations(repo) + + assert violation.startswith("environment_servers/env/app.py: Env has neither") + + def test_a_configured_entrypoint_that_starts_no_server_is_reported(self, repo: Path) -> None: + _write(repo, "resources_servers/lib/app.py", "class Library:\n pass\n") + _write(repo, "resources_servers/lib/helpers.py", "def helper():\n pass\n") + _write(repo, "resources_servers/lib/configs/lib.yaml", _config("lib", "helpers.py")) + + [violation] = contract_violations(repo) + + assert violation.startswith("resources_servers/lib/helpers.py is a configured entrypoint") + + def test_entrypoints_named_by_configs_outside_the_server_directory_are_checked(self, repo: Path) -> None: + _write(repo, "responses_api_agents/multi/app.py", "class A:\n pass\n\nA.run_webserver()\n") + _write(repo, "responses_api_agents/multi/second_agent.py", "class B:\n pass\n\nB.run_webserver()\n") + _write( + repo, + "benchmarks/bench/config.yaml", + "bench_agent:\n responses_api_agents:\n multi:\n entrypoint: second_agent.py\n", + ) + + violations = contract_violations(repo) + + assert [violation.split(":")[0] for violation in violations] == [ + "responses_api_agents/multi/app.py", + "responses_api_agents/multi/second_agent.py", + ] + + +def _config(server_dir: str, entrypoint: str) -> str: + return f"{server_dir}:\n resources_servers:\n {server_dir}:\n entrypoint: {entrypoint}\n"