feat(opensandbox): keepalive-bounded transport + create hardening + resource requests/limits

nemo_gym/sandbox/providers/opensandbox/configs/opensandbox.yaml

@@ -26,10 +26,25 @@ sandbox:
       protocol: http
       request_timeout_s: 300
       use_server_proxy: true
+      # Must stay below the server's keep-alive idle timeout (uvicorn: ~5s),
+      # else pooled sockets are reused after the server has closed them.
+      keepalive_expiry_s: 3.0
+      # 0 disables connection reuse.
+      max_keepalive_connections: 20
+      # null for no cap.
+      max_connections: 100
+      connect_retries: 2
+      # "aiohttp" routes through the optional httpx-aiohttp bridge, falling
+      # back to httpx with a warning when it is not installed.
+      transport_backend: httpx
     create:
       request_timeout_s: 1200
-      timeout_s: 1200
-      skip_health_check: true
+      # Must exceed the spec's ready_timeout_s (1200 for mini_swe_agent_2): this
+      # bounds the whole create call, which now includes the readiness wait.
+      timeout_s: 1500
+      # Skipping the check lets the first command race a pod whose exec daemon
+      # is not listening yet, which returns a 502 and kills the rollout.
+      skip_health_check: false
       retries: 10
       retry_delay_s: 5.0
       retry_max_delay_s: 90.0
@@ -42,6 +57,8 @@ sandbox:
       retries: 5
       retry_delay_s: 1.0
       retry_max_delay_s: 45.0
+      # A retried command the server already started runs twice; agent commands
+      # are usually mutating. Raise only for idempotent workloads.
       command_retries: 0
       close_timeout_s: 30
     # Job attribution merged into every sandbox's metadata. OpenSandbox propagates

nemo_gym/sandbox/providers/opensandbox/provider.py

@@ -348,13 +348,27 @@ def _to_sandbox_status(state: Any) -> SandboxStatus:
 
 @dataclass(frozen=True)
 class OpenSandboxConnectionConfig:
-    """OpenSandbox server connection settings."""
+    """OpenSandbox server connection settings.
+
+    ``keepalive_expiry_s`` must stay below the server's own keep-alive idle
+    timeout (uvicorn defaults to 5s), or pooled sockets are reused after the
+    server has closed them; null falls back to the SDK's default transport.
+    ``transport_backend`` is "httpx" or "aiohttp" (via the optional
+    ``httpx-aiohttp`` bridge, falling back to httpx when it is absent).
+    The pool is shared, so ``max_connections`` also caps in-flight sandbox
+    operations per process; null means no cap.
+    """
 
     domain: str | None = None
     api_key: str | None = None
     protocol: str | None = None
     request_timeout_s: int | None = None
     use_server_proxy: bool = False
+    keepalive_expiry_s: float | None = 3.0
+    max_keepalive_connections: int = 20
+    max_connections: int | None = 100
+    connect_retries: int = 2
+    transport_backend: str = "httpx"
 
 
 @dataclass(frozen=True)
@@ -488,6 +502,9 @@ class OpenSandboxProviderOptions:
     volumes: tuple[Mapping[str, Any], ...] = ()
     skip_health_check: bool | None = None
     extensions: Mapping[str, str] = field(default_factory=dict)
+    # Scheduling requests (same keys as SandboxSpec.resources, which become the
+    # limits). Unset, the server applies the single resources map as both.
+    resource_requests: Mapping[str, Any] | None = None
 
     @classmethod
     def from_mapping(cls, options: Mapping[str, Any] | None) -> "OpenSandboxProviderOptions":
@@ -522,6 +539,9 @@ def from_mapping(cls, options: Mapping[str, Any] | None) -> "OpenSandboxProvider
         extensions = options.get("extensions", {})
         if not isinstance(extensions, Mapping):
             raise TypeError("OpenSandbox provider option 'extensions' must be a mapping")
+        resource_requests = options.get("resource_requests")
+        if resource_requests is not None and not isinstance(resource_requests, Mapping):
+            raise TypeError("OpenSandbox provider option 'resource_requests' must be a mapping")
 
         return cls(
             image_auth=dict(image_auth) if image_auth is not None else None,
@@ -530,6 +550,7 @@ def from_mapping(cls, options: Mapping[str, Any] | None) -> "OpenSandboxProvider
             volumes=tuple(dict(volume) for volume in volumes),
             skip_health_check=skip_health_check,
             extensions=_string_map(dict(extensions)),
+            resource_requests=dict(resource_requests) if resource_requests is not None else None,
         )
 
 
@@ -552,6 +573,10 @@ def __init__(
         self._probe = _coerce_config(probe, OpenSandboxProbeConfig)
         self._operations = _coerce_config(operations, OpenSandboxOperationConfig)
         self._attribution = _coerce_config(attribution, OpenSandboxAttributionConfig)
+        # Shared injected transport. The SDK never closes transports it did not
+        # create, so the provider owns this one: built once, reused by every
+        # ConnectionConfig, closed in aclose().
+        self._transport: Any | None = None
 
     def _resolve_extensions(self, extensions: Mapping[str, str]) -> dict[str, str]:
         """Add the configured default image pull policy to SDK create extensions."""
@@ -587,11 +612,42 @@ def _connection_config(
             kwargs["request_timeout"] = timedelta(seconds=request_timeout_s)
         if self._connection.use_server_proxy:
             kwargs["use_server_proxy"] = True
+        if self._connection.keepalive_expiry_s is not None:
+            kwargs["transport"] = self._get_transport()
         return ConnectionConfig(**kwargs)
 
+    def _get_transport(self) -> Any:
+        """Return the provider-owned shared transport, building it on first use."""
+        if self._transport is None:
+            self._transport = self._build_transport()
+        return self._transport
+
+    def _build_transport(self) -> Any:
+        """Build the SDK transport with the configured pool limits."""
+        import httpx
+
+        limits = httpx.Limits(
+            max_connections=self._connection.max_connections,
+            max_keepalive_connections=self._connection.max_keepalive_connections,
+            keepalive_expiry=self._connection.keepalive_expiry_s,
+        )
+        if self._connection.transport_backend == "aiohttp":
+            try:
+                from httpx_aiohttp import AiohttpTransport
+
+                return AiohttpTransport(limits=limits, retries=self._connection.connect_retries)
+            except ImportError:
+                LOGGER.warning(
+                    "connection.transport_backend=aiohttp requested but httpx-aiohttp "
+                    "is not installed; falling back to the httpx transport"
+                )
+        return httpx.AsyncHTTPTransport(limits=limits, retries=self._connection.connect_retries)
+
     async def aclose(self) -> None:
         """Close provider-owned resources."""
-        return None
+        transport, self._transport = self._transport, None
+        if transport is not None:
+            await transport.aclose()
 
     async def serialize_handle(self, handle: SandboxHandle, *, scope: str | None = None) -> dict[str, Any]:
         """Return a descriptor for reattaching to this sandbox by id.
@@ -809,6 +865,8 @@ async def _create_once(self, spec: SandboxSpec) -> SandboxHandle:
             "extensions": self._resolve_extensions(options.extensions),
             "connection_config": self._connection_config(request_timeout_s=self._create.request_timeout_s),
         }
+        if options.resource_requests is not None:
+            kwargs["resource_requests"] = _resource_map(SandboxResources.from_mapping(options.resource_requests))
         if spec.image is not None:
             kwargs["image"] = _to_image_spec(spec.image, options.image_auth)
         if options.snapshot_id is not None:

pyproject.toml

@@ -235,9 +235,10 @@ sandbox = [
     "tenacity>=9.1.4",
 
     # OpenSandbox SDK: used by the OpenSandbox sandbox provider for create/exec/delete and SDK pool creation.
-    # Updated: Sat May 16, 2026 with opensandbox>=0.1.9
+    # Lower bound 0.1.15: first version exposing separate `resource_requests` on Sandbox.create.
+    # Updated: Thu Jul 30, 2026 with opensandbox>=0.1.15
     # License: Apache 2.0
-    "opensandbox>=0.1.9",
+    "opensandbox>=0.1.15",
 
     # OpenShell SDK: used by the OpenShell sandbox provider for gateway create/exec/delete over gRPC.
     # Lower bound 0.0.92: the version that made `workspace` a required argument on sandbox

responses_api_agents/mini_swe_agent_2/app.py

@@ -12,11 +12,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import asyncio
 import hashlib
 import json
 import os
 import sys
+import threading
 import time
 import traceback
 from asyncio import Semaphore
@@ -89,6 +89,10 @@ class MiniSWEAgentVerifyResponse(BaseVerifyResponse):
 
 
 @ray.remote(
+    # Rollout tasks spend nearly all their time waiting on LLM calls and
+    # sandbox I/O; reserving a full CPU per task caps concurrent rollouts at
+    # the Ray cluster's core count long before any real resource limit.
+    num_cpus=0.25,
     scheduling_strategy="SPREAD",
     runtime_env={
         "py_executable": sys.executable,
@@ -509,6 +513,11 @@ def _run_mini_swe_v2(**params: Any) -> dict[str, Any]:
     model_kwargs = model_config.setdefault("model_kwargs", {})
     model_kwargs["api_key"] = params["api_key"]
     model_kwargs["base_url"] = params["base_url"]
+    # Bounded retries for transient LLM-call failures (disconnects, resets):
+    # without any retry a single failed call kills the whole rollout, while a
+    # large value makes litellm retry silently for so long that the rollout
+    # looks hung. Config-provided model_kwargs take precedence.
+    model_kwargs.setdefault("num_retries", 5)
     model_kwargs.pop("api_base", None)
     max_output_tokens = model_kwargs.pop("max_output_tokens", None)
     if max_output_tokens is not None and "max_tokens" not in model_kwargs:
@@ -578,7 +587,18 @@ def _run_mini_swe_v2(**params: Any) -> dict[str, Any]:
         }
     finally:
         if env and hasattr(env, "cleanup"):
-            env.cleanup()
+            # Off the critical path: this finally block runs before the task's
+            # return value becomes fetchable, so an in-band stop() delays every
+            # finished result and, on failure, re-raises over it. Orphans are
+            # covered by the provider's sandbox TTL.
+            threading.Thread(target=_cleanup_env_best_effort, args=(env,), daemon=True).start()
+
+
+def _cleanup_env_best_effort(env: Any) -> None:
+    try:
+        env.cleanup()
+    except Exception as e:
+        print(f"[CLEANUP] best-effort sandbox teardown failed: {e}", flush=True)
 
 
 def run_mini_swe_with_sandbox(**params: Any) -> Any:
@@ -807,7 +827,12 @@ async def run(self, body: MiniSWEAgentRunRequest) -> MiniSWEAgentVerifyResponse:
                 if runtime_env.get("env_vars"):
                     runner = runner.options(runtime_env=runtime_env)
                 future = runner.remote(run_mini_swe_with_sandbox, params)
-                result = await asyncio.to_thread(ray.get, future)
+                # Ray ObjectRefs are awaitable: park on the event loop instead
+                # of pinning a thread in asyncio's default executor (capped at
+                # min(32, cpu+4) workers). With the thread-blocking ray.get,
+                # at most ~32 rollouts can be waiting on results at once and
+                # every other finished task queues behind them.
+                result = await future
                 result = result[instance_id]
                 input_messages = result["input_messages"]
                 response_output = result["response_output"]

responses_api_agents/mini_swe_agent_2/configs/mini_swe_agent_2.yaml

@@ -20,20 +20,29 @@ mini_swe_agent_2:
         type: responses_api_models
         name: policy_model
       concurrency: 64
+      # Declared (empty) so benchmark configs can inherit this server config and
+      # fill in their dataset list; the config merge rejects undeclared keys.
+      datasets: []
       env: sandbox
       # Name of the sandbox to use; include a provider config that defines a
       # `sandbox` block (e.g. nemo_gym/sandbox/providers/opensandbox/configs/opensandbox.yaml).
       sandbox_provider: sandbox
       sandbox_spec:
         ttl_s: 18000
         ready_timeout_s: 1200
+        # Limits (burst ceiling); memory-spiky test suites OOM-kill below this.
         resources:
           cpu: 2
           memory_mib: 8192
           # 30 GiB works on every provider: within Fargate's 21-200 GiB ephemeral range
           # (an explicit 20 is rejected there) and fine as an ephemeral request elsewhere.
           disk_gib: 30
-        provider_options: {}
+        provider_options:
+          # Scheduling requests, kept below the limits so sandboxes pack densely.
+          resource_requests:
+            cpu: 0.5
+            memory_mib: 2048
+            disk_gib: 30
         metadata:
           benchmark: swebench-verified
           harness: mini-swe-agent