feat(server): refuse several workers in servers that cannot share state

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.
 

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 "

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():

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)

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()

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

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))

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] = {}

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:

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

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)])

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)

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)

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

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 = {

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)
 

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

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)

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