[DO NOT MERGE] bxyu dev

benchmarks/deepswe/opencode.yaml

@@ -27,4 +27,4 @@ opencode_sandboxed_agent_deepswe:
           type: benchmark
           jsonl_fpath: benchmarks/deepswe/data/deepswe_benchmark.jsonl
           prepare_script: benchmarks/deepswe/prepare.py
-          num_repeats: 5
+          num_repeats: 10

benchmarks/nemotron_3.5_super/vllm_configs/nemotron_3.5_super_mtp.sh

@@ -0,0 +1,47 @@
+#!/bin/bash
+
+GYM_MODEL_PARAMS=(
+    "++policy_model.responses_api_models.vllm_model.sampling_overrides.temperature=1.0"
+    "++policy_model.responses_api_models.vllm_model.sampling_overrides.top_p=0.95"
+)
+
+# @bxyu-nvidia: `--skip-mm-profiling` Is needed to get Super VL checkpoint working, even with text benchmarks
+VLLM_COMMON_ARGS=(
+    --trust-remote-code
+    --disable-uvicorn-access-log
+    --gpu-memory-utilization 0.9
+    --distributed-executor-backend mp
+    --data-parallel-backend mp
+    --enable-auto-tool-choice
+    --tool-call-parser qwen3_coder
+    --reasoning-parser nemotron_v3
+    --enable-chunked-prefill
+    --enable-prefix-caching
+    --max-model-len 262144
+    --kv-cache-dtype fp8
+    --no-disable-hybrid-kv-cache-manager
+    --block-size 128
+    --mamba-cache-mode align
+    --mamba-ssm-cache-dtype float32
+    --model-loader-extra-config '{"enable_multithread_load": true, "num_threads": 96}'
+    --enable-expert-parallel
+    --skip-mm-profiling
+    --data-parallel-size 1
+    --api-server-count 1
+    --speculative-config '{"method":"mtp","num_speculative_tokens":5}'
+)
+VLLM_PREFILL_ARGS=(
+    --kv-transfer-config '{"kv_connector":"NixlConnector","kv_role":"kv_producer","kv_load_failure_policy":"fail"}'
+    --max-num-batched-tokens 135680
+    --max-num-seqs 1024
+    --data-parallel-size-local 1
+    --tensor-parallel-size 4
+)
+VLLM_DECODE_ARGS=(
+    --kv-transfer-config '{"kv_connector":"NixlConnector","kv_role":"kv_consumer","kv_load_failure_policy":"fail"}'
+    --compilation-config '{"cudagraph_mode":"FULL_DECODE_ONLY"}'
+    --max-num-batched-tokens 33920
+    --max-num-seqs 1024
+    --data-parallel-size-local 1
+    --tensor-parallel-size 4
+)

fern/versions/latest/pages/infrastructure/sandbox/opensandbox.mdx

@@ -238,3 +238,5 @@ Command failures return `SandboxExecResult` with the command's exit code. If Ope
 The provider's `create.image_pull_policy` defaults to `IfNotPresent`. Valid values are `Always`, `IfNotPresent`, and `Never`. The resolved policy is written into OpenSandbox create extensions as both `imagePullPolicy` and `opensandbox.extensions.image-pull-policy` unless those keys are already present in `provider_options.extensions`.
 
 Create retries handle transient allocation, connection, image pull, and server-side errors. Command retries are controlled separately by `operations.command_retries`.
+
+execd, the exec daemon inside each sandbox, holds every command response open for `EXECD_API_GRACE_SHUTDOWN` (default 1s) after the command's last event, which is fixed latency on each command and on the readiness probe. Set it in the sandbox spec's `env` to shorten it; the shipped SWE-bench and OpenCode configs use `50ms`, and the provider's `operations.background_poll_initial_s` (0.5) is sized to that.

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

@@ -68,7 +68,9 @@ sandbox:
       # Poll interval backs off from initial to interval.
       # @bxyu-nvidia: Too frequent of a poll interval can quickly reduce server performance.
       # e.g. on SWE Bench Verified 500 samples with 3 repeats = 1500 concurrent samples, we would do polling every `background_poll_interval_s` seconds
-      background_poll_initial_s: 1.0
+      # Assumes a short execd response grace (EXECD_API_GRACE_SHUTDOWN in the sandbox env; the
+      # shipped swebench and opencode configs set 50ms).
+      background_poll_initial_s: 0.5
       background_poll_interval_s: 30.0
       # Per-request budget for status polls, which are small idempotent GETs:
       # without their own short budget, each poll against an unreachable

nemo_gym/server_utils.py

@@ -746,13 +746,18 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
         received_messages: asyncio.Queue[Message] = asyncio.Queue()
         client_disconnected = asyncio.Event()
+        response_complete = asyncio.Event()
 
         async def receive_message() -> Message:
             return await received_messages.get()
 
         async def send_message(message: Message) -> None:
-            if not client_disconnected.is_set():
-                await send(message)
+            if client_disconnected.is_set():
+                return
+
+            await send(message)
+            if message["type"] == "http.response.body" and not message.get("more_body", False):
+                response_complete.set()
 
         # The listener is the sole reader of the original ASGI receive channel.
         # Forwarding request messages keeps the body available to the app while
@@ -768,10 +773,17 @@ async def run_app() -> None:
             async def listen_for_disconnect() -> None:
                 while True:
                     message = await receive()
-                    await received_messages.put(message)
                     if message["type"] != "http.disconnect":
+                        await received_messages.put(message)
                         continue
 
+                    # Uvicorn returns http.disconnect from receive() once the response is complete,
+                    # even if the peer did not disconnect early. Only cancel requests whose response
+                    # has not finished being sent.
+                    if response_complete.is_set():
+                        return
+
+                    await received_messages.put(message)
                     client_disconnected.set()
                     self.num_cancelled += 1
                     if is_global_aiohttp_client_request_debug_enabled() or self.num_cancelled % 100 == 0:

resources_servers/swebench/configs/swebench.yaml

@@ -20,6 +20,10 @@ swebench_resources_server:          # instance name — how agents/CLI refer to
         ttl_s: 18000
         ready_timeout_s: 1200
         derive_cpu_env: true
+        env:
+          # execd holds each finished command's response open for this long (default 1s);
+          # a short grace keeps every command and the readiness probe from paying that tail.
+          EXECD_API_GRACE_SHUTDOWN: "50ms"
         resources:
           cpu: 2
           memory_mib: 16384

resources_servers/terminal_bench_2_1/configs/terminal_bench_2_1.yaml

@@ -15,6 +15,10 @@ terminal_bench_2_1_resources_server:          # instance name — how agents/CLI
         ttl_s: 18000
         ready_timeout_s: 1200
         derive_cpu_env: true
+        env:
+          # execd holds each finished command's response open for this long (default 1s);
+          # a short grace keeps every command and the readiness probe from paying that tail.
+          EXECD_API_GRACE_SHUTDOWN: "50ms"
         resources:
           cpu: 4
           memory_mib: 16384

responses_api_agents/opencode_sandboxed_agent/app.py

@@ -476,6 +476,7 @@ async def _start_sandbox(self, sandbox_id: Optional[str] = None) -> AsyncSandbox
         resources = SandboxResources.from_mapping(self.config.sandbox_config.get("resources", {}))
         # TODO @bxyu-nvidia: Refactor this after swapping to PTY as this should be set on the SWE Bench resources server side
         env = cpu_cap_env(resources.cpu) if self.config.sandbox_config.get("derive_cpu_env", True) else {}
+        env |= dict(self.config.sandbox_config.get("env", {}))  # explicit keys win over the derived caps
 
         # TODO @bxyu-nvidia: Refactor this after Hemil's swap from Python dataclass to Pydantic BaseModel
         sandbox_spec = SandboxSpec(

responses_api_agents/opencode_sandboxed_agent/configs/opencode_sandboxed_agent.yaml

@@ -147,6 +147,10 @@ opencode_sandboxed_agent:
       sandbox_config:
         ttl_s: 18000
         ready_timeout_s: 1200
+        env:
+          # execd holds each finished command's response open for this long (default 1s);
+          # a short grace keeps every command and the readiness probe from paying that tail.
+          EXECD_API_GRACE_SHUTDOWN: "50ms"
         resources:
           cpu: 0.25
           memory_mib: 512

responses_api_agents/opencode_sandboxed_agent/tests/test_app.py

@@ -107,6 +107,14 @@ async def created_spec(sandbox_config: Dict[str, Any]) -> Any:
         assert (await created_spec({"resources": {"cpu": 2}, "derive_cpu_env": False})).env == {}
         assert (await created_spec({"resources": {"memory_mib": 8192}})).env == {}
 
+        # Explicit sandbox_config.env is passed through and wins over the derived caps.
+        spec = await created_spec(
+            {"resources": {"cpu": 2}, "env": {"EXECD_API_GRACE_SHUTDOWN": "50ms", "OMP_NUM_THREADS": "4"}}
+        )
+        assert spec.env["EXECD_API_GRACE_SHUTDOWN"] == "50ms"
+        assert spec.env["OMP_NUM_THREADS"] == "4"
+        assert all(spec.env[name] == "2" for name in CPU_CAP_ENV_VARS if name != "OMP_NUM_THREADS")
+
     @fixture
     def opencode_export_test_data(self) -> Dict[str, Any]:
         test_data_path = Path(__file__).parent / "opencode_export_test_data.json"

responses_api_agents/terminus_2_sandboxed_agent/app.py

@@ -15,7 +15,7 @@
 
 from fastapi import Request
 from harbor.agents.terminus_2 import Terminus2
-from harbor.llms.base import BaseLLM, LLMResponse
+from harbor.llms.base import BaseLLM, ContextLengthExceededError, LLMResponse
 from harbor.models.agent.context import AgentContext
 from harbor.models.metric.usage_info import UsageInfo
 from harbor.utils.logger import logger as harbor_logger
@@ -92,6 +92,7 @@ class Terminus2AgentVerifyResponse(BaseVerifyResponse):
     terminus2_time_taken: float
     model_calls_gt_10min: int
     num_compactions: int
+    error: Optional[str]
 
 
 class NeMoGymSandboxEnvironment:
@@ -193,8 +194,8 @@ async def call(self, prompt: str, **kwargs: Any) -> LLMResponse:
         input_items = self._input_items(message_history, prompt)
         response = None
         start_time = perf_counter()
-        max_attempts = 3  # Hardcode 3 attempts for now
-        for attempt in range(max_attempts):
+        max_attempts = 10  # Hardcode 10 attempts for now
+        for _ in range(max_attempts):
             try:
                 async with asyncio.timeout(delay=60 * 10):  # Hardcoded to match litellm default timeout
                     response = NeMoGymResponse.model_validate(
@@ -206,10 +207,7 @@ async def call(self, prompt: str, **kwargs: Any) -> LLMResponse:
                     break
             except TimeoutError:
                 self._model_calls_gt_10min += 1
-                print(
-                    f"Hit LiteLLM default 10min timeout on model call, attempt {attempt + 1} / {max_attempts}",
-                    file=sys.stderr,
-                )
+
         self._times_spent.append(perf_counter() - start_time)
         if not response:
             raise TimeoutError(f"Failed to query model endpoint due to timeouts after {max_attempts} attempts!")
@@ -231,6 +229,11 @@ async def call(self, prompt: str, **kwargs: Any) -> LLMResponse:
                 cost_usd=0.0,
             )
         content, reasoning_content = self._response_text(response)
+
+        # @bxyu-nvidia: Gym will return an empty model response when context length is exceeded
+        if not (content or reasoning_content):
+            raise ContextLengthExceededError
+
         return LLMResponse(
             content=content,
             reasoning_content=reasoning_content,
@@ -259,9 +262,6 @@ def __init__(self, *args: Any, llm: NeMoGymLLM, dump_trajectory: bool, **kwargs:
     def _init_llm(self, *args: Any, **kwargs: Any) -> BaseLLM:
         return self._nemo_gym_llm
 
-    def _count_total_tokens(self, chat: Any) -> int:
-        return sum(len(str(message.get("content", ""))) // 4 for message in chat.messages)
-
     def _dump_trajectory_with_continuation_index(self, continuation_index: int) -> None:
         if self._dump_trajectory_enabled:
             super()._dump_trajectory_with_continuation_index(continuation_index)
@@ -354,10 +354,13 @@ async def _execute(
                 async with asyncio.timeout(self.config.sandbox_timeout):
                     await agent.run(instruction, environment, context)
                 terminus2_completed = True
+                error = None
             except TimeoutError:
                 terminus2_completed = False
+                error = format_exc()
             except:
                 terminus2_completed = False
+                error = format_exc()
                 print(f"Hit exception while running Terminus2: {format_exc()}", file=sys.stderr)
             finally:
                 pass
@@ -397,6 +400,7 @@ async def _execute(
             "terminus2_time_taken": total_time,
             "model_calls_gt_10min": llm._model_calls_gt_10min,
             "num_compactions": agent._num_compactions,
+            "error": error,
         }
         return response, metrics
 

responses_api_agents/terminus_2_sandboxed_agent/infra_error_analysis_rollouts.py

@@ -0,0 +1,40 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+import orjson
+
+
+fpath = "temp.jsonl"
+
+gt_10 = 0
+num_stuck = 0
+num_is_covered = 0
+num_model_call_long = 0
+total = 0
+with open(fpath, "rb") as f:
+    for i, line in enumerate(f):
+        row = orjson.loads(line)
+
+        count = 0
+        stuck_count = 0
+        for output_item in row["response"]["output"]:
+            if output_item.get("role") == "user":
+                content = output_item["content"]
+                count += "No valid JSON found in response" in content
+            elif output_item.get("type") == "reasoning":
+                content = output_item["summary"][0]["text"]
+                stuck_count += "stuck" in content or "unresponsive" in content
+
+        is_gt_10 = count > 10
+        was_stuck = stuck_count > 10
+        model_call_long = row["model_calls_gt_10min"] >= 6
+        is_covered = is_gt_10 or was_stuck or model_call_long
+        if not is_covered:
+            print(f"{i + 1}: {count > 10=} {was_stuck=} {model_call_long=}")
+
+        num_model_call_long += model_call_long
+        num_stuck += was_stuck
+        gt_10 += is_gt_10
+        num_is_covered += is_covered
+        total += 1
+
+print(gt_10, num_stuck, num_model_call_long, num_is_covered, total)

tests/unit_tests/test_server_utils.py

@@ -34,6 +34,7 @@
 from nemo_gym.server_utils import (
     BaseServer,
     BaseServerConfig,
+    ClientDisconnectCancellationMiddleware,
     ConnectionError,
     DictConfig,
     GlobalAIOHTTPAsyncClientConfig,
@@ -628,6 +629,69 @@ async def send(message):
         assert handler_cancelled.is_set()
         assert sent_messages == []
 
+    async def test_cancellation_middleware_ignores_disconnect_after_response_completion(self) -> None:
+        response_sent = asyncio.Event()
+        finish_cleanup = asyncio.Event()
+        cleanup_completed = asyncio.Event()
+        handler_cancelled = asyncio.Event()
+
+        async def inner_app(scope, receive, send) -> None:
+            assert await receive() == {"type": "http.request", "body": b"", "more_body": False}
+            await send({"type": "http.response.start", "status": 200, "headers": []})
+            await send({"type": "http.response.body", "body": b"ok", "more_body": False})
+            try:
+                await finish_cleanup.wait()
+            except asyncio.CancelledError:
+                handler_cancelled.set()
+                raise
+            cleanup_completed.set()
+
+        middleware = ClientDisconnectCancellationMiddleware(inner_app)
+        request_delivered = False
+
+        async def receive():
+            nonlocal request_delivered
+            if not request_delivered:
+                request_delivered = True
+                return {"type": "http.request", "body": b"", "more_body": False}
+
+            await response_sent.wait()
+            return {"type": "http.disconnect"}
+
+        sent_messages = []
+
+        async def send(message):
+            sent_messages.append(message)
+            if message["type"] == "http.response.body" and not message.get("more_body", False):
+                response_sent.set()
+
+        scope = {
+            "type": "http",
+            "asgi": {"version": "3.0"},
+            "http_version": "1.1",
+            "method": "POST",
+            "scheme": "http",
+            "path": "/work",
+            "raw_path": b"/work",
+            "query_string": b"",
+            "headers": [],
+            "client": ("127.0.0.1", 1234),
+            "server": ("testserver", 80),
+        }
+        app_task = asyncio.create_task(middleware(scope, receive, send))
+        await asyncio.wait_for(response_sent.wait(), timeout=1)
+        await asyncio.sleep(0)
+        finish_cleanup.set()
+        await asyncio.wait_for(app_task, timeout=1)
+
+        assert cleanup_completed.is_set()
+        assert not handler_cancelled.is_set()
+        assert middleware.num_cancelled == 0
+        assert sent_messages == [
+            {"type": "http.response.start", "status": 200, "headers": []},
+            {"type": "http.response.body", "body": b"ok", "more_body": False},
+        ]
+
     def test_upstream_error_log_has_bounded_body_and_redacted_url(self) -> None:
         request_info = RequestInfo(
             url=URL("http://policy.test/v1/responses?api_key=secret"),