diff --git a/pyproject.toml b/pyproject.toml index 78168edea3b..98c909546bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -656,6 +656,12 @@ include = [ # SWE harnesses create mutable runtime trees inside their package directory. # Package discovery must never race or ship those caches and rollout outputs. exclude = [ + # Sandboxed harness agents keep one results directory per rollout; a long run leaves tens of + # thousands, and walking them stalls every package build. + "responses_api_agents.claude_code_sandboxed_agent.results", + "responses_api_agents.claude_code_sandboxed_agent.results.*", + "responses_api_agents.opencode_sandboxed_agent.results", + "responses_api_agents.opencode_sandboxed_agent.results.*", "responses_api_agents.swe_agents.results", "responses_api_agents.swe_agents.results.*", "responses_api_agents.swe_agents.swe_openhands_setup", diff --git a/resources_servers/scale_swe/configs/scale_swe.yaml b/resources_servers/scale_swe/configs/scale_swe.yaml index 955ca4a8a55..cab6fe652eb 100644 --- a/resources_servers/scale_swe/configs/scale_swe.yaml +++ b/resources_servers/scale_swe/configs/scale_swe.yaml @@ -6,7 +6,7 @@ scale_swe_resources_server: resources_servers: scale_swe: - allowed_agents: [opencode_sandboxed_agent] + allowed_agents: [opencode_sandboxed_agent, claude_code_sandboxed_agent] entrypoint: app.py domain: coding verified: false diff --git a/resources_servers/scale_swe/configs/scale_swe_claude_code.yaml b/resources_servers/scale_swe/configs/scale_swe_claude_code.yaml new file mode 100644 index 00000000000..31ecba07b2b --- /dev/null +++ b/resources_servers/scale_swe/configs/scale_swe_claude_code.yaml @@ -0,0 +1,39 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Wires the Claude Code sandboxed agent to scale_swe_resources_server, as +# scale_swe_opencode.yaml does for the OpenCode agent. Rows route here with +# agent_ref.name = scale_swe_claude_code_sandboxed_agent. +# +# gym env start --config resources_servers/scale_swe/configs/scale_swe_claude_code.yaml + +config_paths: + - responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml + - resources_servers/scale_swe/configs/scale_swe.yaml + +scale_swe_claude_code_sandboxed_agent: + _inherit_from: claude_code_sandboxed_agent + responses_api_agents: + claude_code_sandboxed_agent: + resources_server: + type: resources_servers + name: scale_swe_resources_server + datasets: + - name: scale_swe + type: train + jsonl_fpath: resources_servers/scale_swe/data/scale_swe_training.jsonl + num_repeats: 1 + # AweAI-Team/Scale-SWE has no declared license on the Hub (no license tag, empty card + # data) -- TBD rather than guessing one of the enum's real licenses. + license: TBD + +# Rollout collection reaches an agent through its environment server. Named after the agent, not +# the environment: scale_swe_opencode.yaml's scale_swe_environment_server fronts the OpenCode +# agent, and a run loads both. +scale_swe_claude_code_environment_server: + environment_servers: + legacy_agent: + entrypoint: app.py + agent_server: + type: responses_api_agents + name: scale_swe_claude_code_sandboxed_agent diff --git a/resources_servers/swe_next/configs/swe_next.yaml b/resources_servers/swe_next/configs/swe_next.yaml index 4878fbf6591..0c027892365 100644 --- a/resources_servers/swe_next/configs/swe_next.yaml +++ b/resources_servers/swe_next/configs/swe_next.yaml @@ -6,7 +6,7 @@ swe_next_resources_server: resources_servers: swe_next: - allowed_agents: [opencode_sandboxed_agent] + allowed_agents: [opencode_sandboxed_agent, claude_code_sandboxed_agent] entrypoint: app.py domain: coding verified: false diff --git a/resources_servers/swe_next/configs/swe_next_claude_code.yaml b/resources_servers/swe_next/configs/swe_next_claude_code.yaml new file mode 100644 index 00000000000..654dbbd35f2 --- /dev/null +++ b/resources_servers/swe_next/configs/swe_next_claude_code.yaml @@ -0,0 +1,37 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Wires the Claude Code sandboxed agent to swe_next_resources_server, as +# swe_next_opencode.yaml does for the OpenCode agent. Rows route here with +# agent_ref.name = swe_next_claude_code_sandboxed_agent. +# +# gym env start --config resources_servers/swe_next/configs/swe_next_claude_code.yaml + +config_paths: + - responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml + - resources_servers/swe_next/configs/swe_next.yaml + +swe_next_claude_code_sandboxed_agent: + _inherit_from: claude_code_sandboxed_agent + responses_api_agents: + claude_code_sandboxed_agent: + resources_server: + type: resources_servers + name: swe_next_resources_server + datasets: + - name: swe_next + type: train + jsonl_fpath: resources_servers/swe_next/data/swe_next_training.jsonl + num_repeats: 1 + license: MIT + +# Rollout collection reaches an agent through its environment server. Named after the agent, not +# the environment: swe_next_opencode.yaml's swe_next_environment_server fronts the OpenCode +# agent, and a run loads both. +swe_next_claude_code_environment_server: + environment_servers: + legacy_agent: + entrypoint: app.py + agent_server: + type: responses_api_agents + name: swe_next_claude_code_sandboxed_agent diff --git a/resources_servers/swe_rebench/configs/swe_rebench.yaml b/resources_servers/swe_rebench/configs/swe_rebench.yaml index b597770b5dd..98c03706366 100644 --- a/resources_servers/swe_rebench/configs/swe_rebench.yaml +++ b/resources_servers/swe_rebench/configs/swe_rebench.yaml @@ -6,7 +6,7 @@ swe_rebench_resources_server: resources_servers: swe_rebench: - allowed_agents: [opencode_sandboxed_agent] + allowed_agents: [opencode_sandboxed_agent, claude_code_sandboxed_agent] entrypoint: app.py domain: coding verified: false diff --git a/resources_servers/swe_rebench/configs/swe_rebench_claude_code.yaml b/resources_servers/swe_rebench/configs/swe_rebench_claude_code.yaml new file mode 100644 index 00000000000..b0ab422d13a --- /dev/null +++ b/resources_servers/swe_rebench/configs/swe_rebench_claude_code.yaml @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Wires the Claude Code sandboxed agent to swe_rebench_resources_server, as +# swe_rebench_opencode.yaml does for the OpenCode agent. Rows route here with +# agent_ref.name = swe_rebench_claude_code_sandboxed_agent. +# +# gym env start --config resources_servers/swe_rebench/configs/swe_rebench_claude_code.yaml + +config_paths: + - responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml + - resources_servers/swe_rebench/configs/swe_rebench.yaml + +swe_rebench_claude_code_sandboxed_agent: + _inherit_from: claude_code_sandboxed_agent + responses_api_agents: + claude_code_sandboxed_agent: + resources_server: + type: resources_servers + name: swe_rebench_resources_server + datasets: + - name: swe_rebench_v2 + type: train + jsonl_fpath: resources_servers/swe_rebench/data/swe_rebench_training.jsonl + num_repeats: 1 + # nebius/SWE-rebench-V2's own Hub metadata license (cc-by-4.0). + license: Creative Commons Attribution 4.0 International + +# Rollout collection reaches an agent through its environment server. Named after the agent, not +# the environment: swe_rebench_opencode.yaml's swe_rebench_environment_server fronts the OpenCode +# agent, and a run loads both. +swe_rebench_claude_code_environment_server: + environment_servers: + legacy_agent: + entrypoint: app.py + agent_server: + type: responses_api_agents + name: swe_rebench_claude_code_sandboxed_agent diff --git a/resources_servers/swemer_v1/configs/swemer_v1.yaml b/resources_servers/swemer_v1/configs/swemer_v1.yaml index b2d5f068a7b..d1968004092 100644 --- a/resources_servers/swemer_v1/configs/swemer_v1.yaml +++ b/resources_servers/swemer_v1/configs/swemer_v1.yaml @@ -10,7 +10,7 @@ swemer_v1_resources_server: resources_servers: swemer_v1: - allowed_agents: [opencode_sandboxed_agent] + allowed_agents: [opencode_sandboxed_agent, claude_code_sandboxed_agent] entrypoint: app.py domain: coding verified: false diff --git a/resources_servers/swemer_v1/configs/swemer_v1_claude_code.yaml b/resources_servers/swemer_v1/configs/swemer_v1_claude_code.yaml new file mode 100644 index 00000000000..5dbba08dd12 --- /dev/null +++ b/resources_servers/swemer_v1/configs/swemer_v1_claude_code.yaml @@ -0,0 +1,37 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Wires the Claude Code sandboxed agent to swemer_v1_resources_server, as +# swemer_v1_opencode.yaml does for the OpenCode agent. Rows route here with +# agent_ref.name = swemer_v1_claude_code_sandboxed_agent. +# +# gym env start --config resources_servers/swemer_v1/configs/swemer_v1_claude_code.yaml + +config_paths: + - responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml + - resources_servers/swemer_v1/configs/swemer_v1.yaml + +swemer_v1_claude_code_sandboxed_agent: + _inherit_from: claude_code_sandboxed_agent + responses_api_agents: + claude_code_sandboxed_agent: + resources_server: + type: resources_servers + name: swemer_v1_resources_server + datasets: + - name: swemer_v1 + type: train + jsonl_fpath: resources_servers/swemer_v1/data/swemer_v1_training.jsonl + num_repeats: 1 + license: NVIDIA Internal Use Only, Do Not Distribute + +# Rollout collection reaches an agent through its environment server. Named after the agent, not +# the environment: swemer_v1_opencode.yaml's swemer_v1_environment_server fronts the OpenCode +# agent, and a run loads both. +swemer_v1_claude_code_environment_server: + environment_servers: + legacy_agent: + entrypoint: app.py + agent_server: + type: responses_api_agents + name: swemer_v1_claude_code_sandboxed_agent diff --git a/resources_servers/swemer_v2/configs/swemer_v2.yaml b/resources_servers/swemer_v2/configs/swemer_v2.yaml index 28fc9d5e0d2..6ad5ed816ae 100644 --- a/resources_servers/swemer_v2/configs/swemer_v2.yaml +++ b/resources_servers/swemer_v2/configs/swemer_v2.yaml @@ -13,7 +13,7 @@ swemer_v2_resources_server: resources_servers: swemer_v2: - allowed_agents: [opencode_sandboxed_agent] + allowed_agents: [opencode_sandboxed_agent, claude_code_sandboxed_agent] entrypoint: app.py domain: coding verified: false diff --git a/resources_servers/swemer_v2/configs/swemer_v2_claude_code.yaml b/resources_servers/swemer_v2/configs/swemer_v2_claude_code.yaml new file mode 100644 index 00000000000..61eb5c168e1 --- /dev/null +++ b/resources_servers/swemer_v2/configs/swemer_v2_claude_code.yaml @@ -0,0 +1,37 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Wires the Claude Code sandboxed agent to swemer_v2_resources_server, as +# swemer_v2_opencode.yaml does for the OpenCode agent. Rows route here with +# agent_ref.name = swemer_v2_claude_code_sandboxed_agent. +# +# gym env start --config resources_servers/swemer_v2/configs/swemer_v2_claude_code.yaml + +config_paths: + - responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml + - resources_servers/swemer_v2/configs/swemer_v2.yaml + +swemer_v2_claude_code_sandboxed_agent: + _inherit_from: claude_code_sandboxed_agent + responses_api_agents: + claude_code_sandboxed_agent: + resources_server: + type: resources_servers + name: swemer_v2_resources_server + datasets: + - name: swemer_v2 + type: train + jsonl_fpath: resources_servers/swemer_v2/data/swemer_v2_training.jsonl + num_repeats: 1 + license: NVIDIA Internal Use Only, Do Not Distribute + +# Rollout collection reaches an agent through its environment server. Named after the agent, not +# the environment: swemer_v2_opencode.yaml's swemer_v2_environment_server fronts the OpenCode +# agent, and a run loads both. +swemer_v2_claude_code_environment_server: + environment_servers: + legacy_agent: + entrypoint: app.py + agent_server: + type: responses_api_agents + name: swemer_v2_claude_code_sandboxed_agent diff --git a/responses_api_agents/claude_code_sandboxed_agent/.gitignore b/responses_api_agents/claude_code_sandboxed_agent/.gitignore new file mode 100644 index 00000000000..1a06816d838 --- /dev/null +++ b/responses_api_agents/claude_code_sandboxed_agent/.gitignore @@ -0,0 +1 @@ +results diff --git a/responses_api_agents/claude_code_sandboxed_agent/README.md b/responses_api_agents/claude_code_sandboxed_agent/README.md new file mode 100644 index 00000000000..272eb38aaad --- /dev/null +++ b/responses_api_agents/claude_code_sandboxed_agent/README.md @@ -0,0 +1,94 @@ +# Claude Code Sandboxed Agent + +Runs the Claude Code CLI (`claude -p`) inside the task's own sandbox, the way +`opencode_sandboxed_agent` runs OpenCode: the resources server seeds the sandbox, this agent +attaches to it, runs one command under `sandbox_timeout`, reads the transcript the CLI left behind, +calls `/verify` and stops the sandbox. Gym stays on the host; the sandbox needs only the binary. + +Claude Code talks to the Gym model server's `/v1/messages` route. Every Gym model server serves +that route, so the agent works with any of them. With `token_id_capture: true` (the default) each +call carries the rollout's training-capture URL prefix, so training gets the exact sampled tokens. + +```bash +gym env start \ + --config responses_api_models/vllm_model/configs/vllm_model.yaml \ + --config nemo_gym/sandbox/providers/opensandbox/configs/opensandbox.yaml \ + --config resources_servers/swe_rebench/configs/swe_rebench_claude_code.yaml +``` + +Each SWE resources server that has an OpenCode wiring config (`_opencode.yaml`) has a +Claude Code one next to it (`_claude_code.yaml`) with the agent instance +`_claude_code_sandboxed_agent` and the environment server +`_claude_code_environment_server` that fronts it; rows route to it with +`agent_ref.name = _claude_code_sandboxed_agent`. A run may load both wiring configs. + +## Installing Claude Code in the sandbox + +Stage the release binary where every sandbox can read it, for example on a read-only bucket +mount, and set `remote_claude_code_binary_path` (and `remote_claude_code_musl_binary_path` for +musl-based images). The agent copies it to a writable directory, makes it executable and checks +that it reports `claude_code_version`. Without a staged binary the official installer runs inside +the sandbox, which needs network access. + +```bash +VERSION=2.1.287 +curl -fsSL -o claude https://downloads.claude.ai/claude-code-releases/$VERSION/linux-x64/claude +curl -fsSL -o manifest.json https://downloads.claude.ai/claude-code-releases/$VERSION/manifest.json +grep -q "$(sha256sum claude | cut -d' ' -f1)" manifest.json && echo MATCH || echo MISMATCH +``` + +## What the agent sets + +- A fresh `CLAUDE_CONFIG_DIR`, binary and transcript directory per rollout under + `remote_work_dir` (`/tmp`), outside the task repository; stdin closed. +- The full harness: `bare: false`. `--bare` is Claude Code's minimal mode, with a one-line system + prompt and only the Bash, Read and Edit tools. +- `--setting-sources user`: only the rollout's empty user settings and `--settings` load, so a task + repository's `.claude/` settings and hooks cannot change the run. +- No auto-memory (`CLAUDE_CODE_DISABLE_AUTO_MEMORY=1`): each rollout starts from an empty config + directory, and notes written mid-session would change the context of later calls. +- `--dangerously-skip-permissions` with `IS_SANDBOX=1`. +- `--disallowedTools` for what does not fit a one-shot run in a task sandbox: web access and skills + (as for the OpenCode agent), worktrees (they would move the edits out of the repository the + verifier diffs), scheduled prompts, wakeups, workflows, agent messaging and the code-review report + tool. Agent (subagents), Bash, Read, Edit, Write, NotebookEdit and TaskStop stay. +- Every model role (`ANTHROPIC_MODEL`, the opus/sonnet/haiku defaults, subagents) points at `model`. +- No update checks, telemetry, error reports, experimental betas or per-request attribution text. + The attribution text would change the system prompt on every call and break capture's call chain. +- `CLAUDE_STREAM_FIRST_BYTE_TIMEOUT_MS=1800000` and `API_TIMEOUT_MS=3600000`. A Gym model server + sends a streamed reply's headers only after generation finishes, and Claude Code's default wait + of 3 to 5 minutes for headers would abort and retry long calls. +- `auto_compact: false` sets `DISABLE_COMPACT`, `DISABLE_AUTO_COMPACT` and the `autoCompactEnabled` + and `precomputeCompactionEnabled` settings to off. Compaction rewrites the conversation, which + token capture cannot follow; training must turn it off. With it off, Claude Code sends requests + until the model server rejects one, like OpenCode with compaction off. +- The CLI runs under `timeout` until `harness_timeout_margin_s` before `sandbox_timeout`, so it + cannot keep editing the repository or calling the model while the verifier runs. + +`claude_code_env` and `claude_code_settings` are layered over all of the above. + +## Output + +The stream-json output is downloaded to `results//stream.jsonl` (or under +`artifacts_dir`) and converted to Responses items: the main conversation's reasoning, text, tool +calls and tool results. Subagent traffic and Claude Code's own error notices are left out. A +`generation.json` receipt with the response and the execution fields is written before grading. +When observations are collected, the session transcripts are archived and read with +`claude_code_agent.observability`, which also reports any compaction. + +The `/run` result carries, besides the verifier's fields: + +- `claude_code_finished`, `claude_code_exit_code`, `claude_code_error_type`, `claude_code_failed` + (exec error, nonzero exit, missing transcript or error result), `claude_code_export_found`, + `claude_code_results_fpath`, `claude_code_run_stdout`, `claude_code_run_stderr`. +- How the session ended, from its final `result` event: `claude_code_result_success`, + `claude_code_result_error`, `claude_code_result_missing` (the deadline or a crash cut it off), + `claude_code_context_overflow` (the last request no longer fit the context), and + `claude_code_duration_s`. +- What it did, main conversation and subagents together: `claude_code_main_turns`, + `claude_code_subagent_turns`, `claude_code_subagent_calls`, `claude_code_tool_calls`, + `claude_code_tool_errors` (calls Claude Code refused) and `claude_code_unknown_tool_calls` + (refused for naming a tool that does not exist). + +With `execution_failure_reward_zero: true` a failed run is graded as an empty response instead of +whatever patch it left behind; the verifier must score an empty response as an unmasked zero. diff --git a/responses_api_agents/claude_code_sandboxed_agent/__init__.py b/responses_api_agents/claude_code_sandboxed_agent/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/responses_api_agents/claude_code_sandboxed_agent/app.py b/responses_api_agents/claude_code_sandboxed_agent/app.py new file mode 100644 index 00000000000..914e95a10aa --- /dev/null +++ b/responses_api_agents/claude_code_sandboxed_agent/app.py @@ -0,0 +1,1016 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. +"""Runs the Claude Code CLI inside the task's own sandbox. + +Same shape as ``opencode_sandboxed_agent``: the resources server seeds the task and hands back its +sandbox, this agent attaches to it, runs one command under ``sandbox_timeout``, reads the +transcript the CLI left behind, calls ``/verify`` and stops the sandbox. Gym stays on the host; +nothing but the Claude Code binary is needed inside the sandbox. The command: + +1. installs Claude Code. A staged binary is copied out of its read-only mount and made executable, + since S3 mounts keep no file modes. Without a staged binary the official installer runs, which + needs network access from the sandbox. +2. checks that the binary reports ``claude_code_version``. +3. writes a settings file and runs ``claude -p`` in the task's working directory. The stream-json + output goes to a file outside the task repository, so a verifier that diffs the repository + never sees it, and stdin is closed so the CLI does not wait for piped input. + +Model calls go to the Gym model server's ``/v1/messages`` route under this rollout's URL prefix, so +training-token capture records every call. The transcript built here is what the verifier and the +rollout record see; training rebuilds ``response.output`` from the captured tokens. +""" + +import json +import sys +import tarfile +import tempfile +from asyncio import Semaphore +from copy import deepcopy +from dataclasses import dataclass +from pathlib import Path +from shlex import quote +from time import time +from traceback import format_exc +from typing import Any, Dict, Iterable, List, Optional +from uuid import uuid4 + +from fastapi import Request +from pydantic import ConfigDict, Field + +from nemo_gym.base_resources_server import BaseRunRequest, BaseVerifyRequest, BaseVerifyResponse +from nemo_gym.base_responses_api_agent import ( + BaseResponsesAPIAgentConfig, + Body, + SimpleResponsesAPIAgent, +) +from nemo_gym.config_types import ModelServerRef, ResourcesServerRef +from nemo_gym.global_config import get_global_config_dict +from nemo_gym.openai_utils import ( + NeMoGymFunctionCallOutput, + NeMoGymResponse, + NeMoGymResponseCreateParamsNonStreaming, + NeMoGymResponseFunctionToolCall, + NeMoGymResponseInputTokensDetails, + NeMoGymResponseOutputItem, + NeMoGymResponseOutputMessage, + NeMoGymResponseOutputText, + NeMoGymResponseOutputTokensDetails, + NeMoGymResponseReasoningItem, + NeMoGymResponseUsage, + NeMoGymSummary, +) +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ObservationGap, + SandboxObservation, + ToolCallObservation, +) +from nemo_gym.sandbox import AsyncSandbox, SandboxResources, SandboxSpec, create_provider +from nemo_gym.sandbox.agent_tools import sandbox_server_url, verify_agent_response +from nemo_gym.sandbox.config import resolve_provider_config, resolve_provider_metadata +from nemo_gym.sandbox.utils import cpu_cap_env +from nemo_gym.server_utils import SESSION_ID_KEY, is_nemo_gym_fastapi_entrypoint, raise_for_status +from responses_api_agents.claude_code_agent.observability import extract_claude_code_observations + + +STREAM_FNAME = "stream.jsonl" +TRANSCRIPTS_TAR_FNAME = "transcripts.tar" +FINISHED_MARKER = "Claude Code run finished" +OFFICIAL_INSTALLER_URL = "https://claude.ai/install.sh" +# The Gym model server does not authenticate; Claude Code only needs some credential to start. +PLACEHOLDER_API_KEY = "nemo-gym" # pragma: allowlist secret +# Claude Code's model name for messages it makes up itself, such as API error notices. +SYNTHETIC_MODEL = "" +# The tools that start a subagent (Task is the pre-2.1 name Claude Code still accepts). +SUBAGENT_TOOLS = frozenset({"Agent", "Task"}) +# How Claude Code reports a tool call it refused to run, and the refusal for a tool name that does +# not exist (for example a lowercase ``read`` from a model used to another harness). +TOOL_ERROR_TAG = "" +UNKNOWN_TOOL_MESSAGE = "No such tool available" +# The error text of a ``result`` event whose last request no longer fit the model's context. +CONTEXT_OVERFLOW_MESSAGE = "Prompt is too long" + + +class ClaudeCodeSandboxedAgentConfig(BaseResponsesAPIAgentConfig): + resources_server: ResourcesServerRef + model_server: ModelServerRef + + claude_code_version: str + # Binaries staged inside the sandbox, e.g. on a read-only bucket mount. The musl build is only + # needed for musl-based task images. With no staged binary the official installer runs. + remote_claude_code_binary_path: Optional[str] = None + remote_claude_code_musl_binary_path: Optional[str] = None + # Each rollout's binary, settings and transcripts go in a fresh directory under this one, + # which must be writable, allow executables, and lie outside the task repository. + remote_work_dir: str = "/tmp" + + # The model name Claude Code sends, for every role (main loop, subagents, background tasks). A + # Gym model server serves its own configured model whatever the name. + model: str = "policy" + # Claude Code's automatic context compaction. Compaction rewrites the conversation, which + # training-token capture cannot follow, so training turns it off. + auto_compact: bool = True + # Auto-memory keeps notes across sessions; every rollout starts from an empty config directory, + # and notes written mid-session would change the context of later calls. + auto_memory: bool = False + # `--bare` is Claude Code's minimal mode (CLAUDE_CODE_SIMPLE): a one-line system prompt and only + # the Bash, Read and Edit tools. Off, the agent runs the full harness. + bare: bool = False + # Settings sources Claude Code loads besides --settings (user, project, local). "user" alone is + # this rollout's empty config directory, so a task repository's .claude/ settings and hooks + # cannot change the run. Null loads Claude Code's default sources. + setting_sources: Optional[str] = "user" + allowed_tools: Optional[str] = None + disallowed_tools: Optional[str] = None + max_turns: Optional[int] = None + append_system_prompt: Optional[str] = None + # Per-call output cap (CLAUDE_CODE_MAX_OUTPUT_TOKENS). Claude Code clamps it to its own limit. + max_output_tokens: Optional[int] = None + # Claude Code's request timeout (API_TIMEOUT_MS) and how long it waits for response headers + # (CLAUDE_STREAM_FIRST_BYTE_TIMEOUT_MS) before it aborts and retries the call. A Gym model + # server sends the headers of a streamed reply only after generation finishes. + api_timeout_ms: Optional[int] = None + stream_first_byte_timeout_ms: Optional[int] = None + # The CLI is stopped this many seconds before sandbox_timeout so that it cannot keep editing + # the repository or calling the model after the agent gives up on it. + harness_timeout_margin_s: float = 120 + + # Layered over the settings and environment derived from the fields above. + claude_code_settings: Dict[str, Any] = Field(default_factory=dict) + claude_code_env: Dict[str, str] = Field(default_factory=dict) + + concurrency: int = Field(default=64, gt=0) + # Grade a failed run (exec error, nonzero exit, missing transcript, error result) as an empty + # response instead of whatever patch it left behind. The verifier must score an empty + # response as an unmasked zero. + execution_failure_reward_zero: bool = False + # Where each rollout's transcript and generation receipt are kept; this agent's ``results`` + # directory by default. + artifacts_dir: Optional[str] = None + + # Sandbox config + sandbox_provider: str + sandbox_config: Dict[str, Any] + sandbox_timeout: float + + debug: bool = False + + +class ClaudeCodeSandboxedAgentRunRequest(BaseRunRequest): + # Allow for benchmark params to propagate properly + model_config = ConfigDict(extra="allow") + + +class ClaudeCodeSandboxedAgentVerifyRequest(BaseVerifyRequest): + # Allow for benchmark params to propagate properly + model_config = ConfigDict(extra="allow") + + +class ClaudeCodeSandboxedAgentVerifyResponse(BaseVerifyResponse): + # Allow for benchmark params to propagate properly + model_config = ConfigDict(extra="allow") + + claude_code_results_fpath: str + claude_code_run_stdout: str + claude_code_run_stderr: str + claude_code_finished: bool + claude_code_export_found: bool + claude_code_exit_code: Optional[int] = None + claude_code_error_type: Optional[str] = None + claude_code_failed: bool = False + # How the session ended, from its final ``result`` event: finished on its own, ended with an + # error (``claude_code_context_overflow`` names the one that matters most), or never wrote one + # (the deadline or a crash cut it off). + claude_code_result_success: bool = False + claude_code_result_error: bool = False + claude_code_result_missing: bool = True + claude_code_context_overflow: bool = False + claude_code_duration_s: Optional[float] = None + # What the session did: model turns of the main conversation and of subagents, subagent + # launches, tool calls, and tool calls Claude Code refused, of which the ones naming a tool that + # does not exist. + claude_code_main_turns: int = 0 + claude_code_subagent_turns: int = 0 + claude_code_subagent_calls: int = 0 + claude_code_tool_calls: int = 0 + claude_code_tool_errors: int = 0 + claude_code_unknown_tool_calls: int = 0 + ng_agent_observations: Optional[AgentObservationBundle] = Field( + default=None, + exclude_if=lambda value: value is None, + ) + + +@dataclass(frozen=True) +class ClaudeCodeRunPaths: + """Where one rollout's files live inside the sandbox.""" + + root: str + + @property + def binary(self) -> str: + return f"{self.root}/bin/claude" + + @property + def config_dir(self) -> str: + return f"{self.root}/config" + + @property + def settings(self) -> str: + return f"{self.root}/settings.json" + + @property + def stream(self) -> str: + return f"{self.root}/{STREAM_FNAME}" + + @property + def transcripts_tar(self) -> str: + return f"{self.root}/{TRANSCRIPTS_TAR_FNAME}" + + +@dataclass +class ClaudeCodeStream: + """What the stream-json output of one ``claude -p`` run says about the run.""" + + output: List[NeMoGymResponseOutputItem] + usage: Optional[NeMoGymResponseUsage] + result: Optional[Dict[str, Any]] + compaction_attempts: List[Dict[str, str]] + compaction_boundaries: int + invalid_lines: int + # Distinct assistant message ids, in the main conversation and under a parent_tool_use_id. + main_turns: int = 0 + subagent_turns: int = 0 + # Tool calls of the main conversation that started a subagent. + subagent_calls: int = 0 + # Tool calls, main conversation and subagents alike, and the ones Claude Code refused to run. + tool_calls: int = 0 + tool_errors: int = 0 + unknown_tool_calls: int = 0 + + +def _tool_result_text(content: Any) -> str: + """Flatten a tool_result's content the way the Gym Messages converter does.""" + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" if content is None else json.dumps(content) + return "\n".join( + block.get("text", "") for block in content if isinstance(block, dict) and block.get("type") == "text" + ) + + +def _token_count(value: Any) -> int: + return int(value) if isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 else 0 + + +def _usage_from_messages_usage(usages: Iterable[Dict[str, Any]]) -> Optional[NeMoGymResponseUsage]: + """Sum Anthropic Messages usage blocks into one Responses usage. + + Anthropic counts cache reads and writes outside ``input_tokens``; they are prompt tokens too. + """ + found = False + input_tokens = cached_tokens = output_tokens = 0 + for usage in usages: + if not isinstance(usage, dict): + continue + found = True + cache_read = _token_count(usage.get("cache_read_input_tokens")) + cache_creation = _token_count(usage.get("cache_creation_input_tokens")) + input_tokens += _token_count(usage.get("input_tokens")) + cache_read + cache_creation + cached_tokens += cache_read + output_tokens += _token_count(usage.get("output_tokens")) + if not found: + return None + return NeMoGymResponseUsage( + input_tokens=input_tokens, + input_tokens_details=NeMoGymResponseInputTokensDetails(cached_tokens=cached_tokens), + output_tokens=output_tokens, + output_tokens_details=NeMoGymResponseOutputTokensDetails(reasoning_tokens=0), + total_tokens=input_tokens + output_tokens, + ) + + +def _content_blocks(content: Any) -> List[Dict[str, Any]]: + if isinstance(content, str): + return [{"type": "text", "text": content}] + if isinstance(content, list): + return [block for block in content if isinstance(block, dict)] + return [] + + +def parse_claude_code_stream(lines: Iterable[str]) -> ClaudeCodeStream: + """Convert ``claude -p --output-format stream-json --verbose`` output into Responses items. + + Only the main conversation is converted. Subagent traffic carries a ``parent_tool_use_id`` + and stays out of the transcript; the main conversation sees a subagent only through its tool + call and result. Messages Claude Code makes up itself (model ````, e.g. API error + notices) are not model output and are left out too. Claude Code emits one ``assistant`` event + per content block, all with the same message id, so usage is taken once per message id + unless the final ``result`` event reports the session total. The counters cover the main + conversation and subagents alike. + """ + output: List[NeMoGymResponseOutputItem] = [] + usage_by_message: Dict[str, Dict[str, Any]] = {} + result: Optional[Dict[str, Any]] = None + compacting: set[str] = set() + compaction_attempts: List[Dict[str, str]] = [] + compaction_boundaries = 0 + invalid_lines = 0 + main_messages: set[str] = set() + subagent_messages: set[str] = set() + subagent_calls = tool_calls = tool_errors = unknown_tool_calls = 0 + + for line in lines: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except (json.JSONDecodeError, RecursionError): + invalid_lines += 1 + continue + if not isinstance(event, dict): + invalid_lines += 1 + continue + + event_type = event.get("type") + if event_type == "result": + result = event + continue + if event_type == "system": + session_id = event.get("session_id") + if event.get("subtype") == "compact_boundary": + compaction_boundaries += 1 + elif event.get("subtype") == "status" and isinstance(session_id, str) and session_id: + if event.get("status") == "compacting": + compacting.add(session_id) + elif event.get("compact_result") in {"failed", "success"}: + if event["compact_result"] == "failed": + compaction_attempts.append({"invocation_id": session_id, "outcome": "failed"}) + compacting.discard(session_id) + continue + + message = event.get("message") + if not isinstance(message, dict): + continue + in_subagent = bool(event.get("parent_tool_use_id")) + blocks = _content_blocks(message.get("content")) + if event_type == "assistant": + if message.get("model") == SYNTHETIC_MODEL: + continue + message_id = message.get("id") + if isinstance(message_id, str): + (subagent_messages if in_subagent else main_messages).add(message_id) + if isinstance(message.get("usage"), dict): + usage_by_message[message_id] = message["usage"] + for block in blocks: + if block.get("type") == "tool_use": + tool_calls += 1 + if not in_subagent and block.get("name") in SUBAGENT_TOOLS: + subagent_calls += 1 + elif event_type == "user": + for block in blocks: + if block.get("type") != "tool_result": + continue + text = _tool_result_text(block.get("content")) + if block.get("is_error") is True or TOOL_ERROR_TAG in text: + tool_errors += 1 + if UNKNOWN_TOOL_MESSAGE in text: + unknown_tool_calls += 1 + if in_subagent: + continue + + if event_type == "assistant": + for block in blocks: + block_type = block.get("type") + if block_type == "thinking" and block.get("thinking"): + output.append( + NeMoGymResponseReasoningItem( + id=f"rs_{uuid4().hex}", + summary=[NeMoGymSummary(text=block["thinking"], type="summary_text")], + type="reasoning", + ) + ) + elif block_type == "text" and block.get("text"): + text = NeMoGymResponseOutputText(text=block["text"], annotations=[], type="output_text") + output.append( + NeMoGymResponseOutputMessage( + id=f"msg_{uuid4().hex}", + content=[text], + role="assistant", + status="completed", + type="message", + ) + ) + elif block_type == "tool_use" and isinstance(block.get("id"), str): + output.append( + NeMoGymResponseFunctionToolCall( + # Serialized as the Gym Messages converter does for the next request. + arguments=json.dumps(block.get("input", {})), + call_id=block["id"], + name=str(block.get("name") or ""), + type="function_call", + status="completed", + ) + ) + elif event_type == "user": + for block in blocks: + if block.get("type") == "tool_result" and isinstance(block.get("tool_use_id"), str): + output.append( + NeMoGymFunctionCallOutput( + call_id=block["tool_use_id"], + output=_tool_result_text(block.get("content")), + type="function_call_output", + ) + ) + + compaction_attempts.extend({"invocation_id": session_id, "outcome": "unknown"} for session_id in compacting) + result_usage = result.get("usage") if result is not None else None + usage = _usage_from_messages_usage([result_usage] if isinstance(result_usage, dict) else usage_by_message.values()) + return ClaudeCodeStream( + output=output, + usage=usage, + result=result, + compaction_attempts=compaction_attempts, + compaction_boundaries=compaction_boundaries, + invalid_lines=invalid_lines, + main_turns=len(main_messages), + subagent_turns=len(subagent_messages), + subagent_calls=subagent_calls, + tool_calls=tool_calls, + tool_errors=tool_errors, + unknown_tool_calls=unknown_tool_calls, + ) + + +def context_overflow(result: Optional[Dict[str, Any]]) -> bool: + """Whether the session ended because a request no longer fit the model's context.""" + if result is None or result.get("is_error") is not True: + return False + text = result.get("result") + return isinstance(text, str) and text.lstrip().startswith(CONTEXT_OVERFLOW_MESSAGE) + + +def invocation_outcome(result: Optional[Dict[str, Any]]) -> tuple[str, Optional[str]]: + """Map the final ``result`` event to an invocation status and error type. + + Claude Code reports an error either as an ``error_*`` subtype or, for an API error such as a + context overflow, as subtype ``success`` with ``is_error`` set and the error text in ``result``. + """ + if result is None: + return "incomplete", "result_missing" + subtype = result.get("subtype") + if subtype == "error_max_turns": + return "incomplete", subtype + if context_overflow(result): + return "failed", "context_overflow" + if isinstance(subtype, str) and subtype.startswith("error"): + return "failed", subtype + if result.get("is_error") is True: + return "failed", "agent_error" + if subtype == "success": + return "completed", None + return "incomplete", "result_unrecognized" + + +class ClaudeCodeSandboxedAgent(SimpleResponsesAPIAgent): + ray_enabled = False + config: ClaudeCodeSandboxedAgentConfig + + def model_post_init(self, context: Any, /) -> None: + super().model_post_init(context) + + self._sem = Semaphore(self.config.concurrency) + self._sandbox_id_to_sandbox: Dict[str, AsyncSandbox] = dict() + self._sandbox_id_to_run_result: Dict[str, Dict[str, Any]] = dict() + + async def _start_sandbox(self, sandbox_id: Optional[str] = None, workdir: Optional[str] = None) -> AsyncSandbox: + global_config_dict = get_global_config_dict() + resolved_sandbox_provider = create_provider( + resolve_provider_config(self.config.sandbox_provider, global_config_dict) + ) + provider_default_metadata = resolve_provider_metadata(self.config.sandbox_provider, global_config_dict) + + if sandbox_id: + return await AsyncSandbox.connect( + {"sandbox_id": sandbox_id, "workdir": workdir}, provider=resolved_sandbox_provider + ) + + if self.config.debug: + print("Creating new sandbox since one wasn't provided", file=sys.stderr) + + resources = SandboxResources.from_mapping(self.config.sandbox_config.get("resources", {})) + 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 + + sandbox_spec = SandboxSpec( + image=self.config.sandbox_config.get("image", "swebench/sweb.eval.x86_64.astropy_1776_astropy-12907"), + ttl_s=self.config.sandbox_config.get("ttl_s", None), + ready_timeout_s=self.config.sandbox_config.get("ready_timeout_s", None), + workdir=self.config.sandbox_config.get("workdir"), + env=env, + files=dict(self.config.sandbox_config.get("files", {})), + metadata=provider_default_metadata + | self.config.sandbox_config.get("metadata", {}) + | { + "nemo_gym_agent": self.config.name, + }, + resources=resources, + entrypoint=self.config.sandbox_config.get("entrypoint"), + provider_options=deepcopy(self.config.sandbox_config.get("provider_options", {})), + ) + sandbox = AsyncSandbox(resolved_sandbox_provider) + await sandbox.start(sandbox_spec) + + return sandbox + + def _agent_sandbox_observation( + self, + *, + sandbox: AsyncSandbox, + return_code: Any, + error_type: Any, + finished: bool, + ) -> SandboxObservation: + handle = getattr(sandbox, "_handle", None) + handle_provider = getattr(handle, "provider_name", None) + handle_sandbox_id = getattr(handle, "sandbox_id", None) + normalized_error = error_type.lower() if isinstance(error_type, str) else "" + if "timeout" in normalized_error or (not normalized_error and return_code == 124): + outcome = "timeout" + elif normalized_error: + outcome = "sandbox_error" + elif return_code == 0 and finished: + outcome = "completed" + elif isinstance(return_code, int): + outcome = "failed" if return_code != 0 else "unknown" + else: + outcome = "unknown" + return SandboxObservation( + role="agent", + provider=handle_provider if isinstance(handle_provider, str) else None, + sandbox_id=handle_sandbox_id if isinstance(handle_sandbox_id, str) else None, + outcome=outcome, + exit_code=return_code if not normalized_error and isinstance(return_code, int) else None, + error_type=error_type if isinstance(error_type, str) else None, + ) + + def _new_run_paths(self) -> ClaudeCodeRunPaths: + work_dir = self.config.remote_work_dir.rstrip("/") + return ClaudeCodeRunPaths(root=f"{work_dir}/nemo-gym-claude-code-{uuid4().hex}") + + def _results_dir(self, request: Request) -> Path: + results_root = ( + Path(self.config.artifacts_dir) if self.config.artifacts_dir else Path(__file__).parent / "results" + ) + results_dir = results_root / request.session[SESSION_ID_KEY] + results_dir.mkdir(parents=True, exist_ok=True) + return results_dir + + def _settings(self) -> Dict[str, Any]: + settings: Dict[str, Any] = {} + if not self.config.auto_compact: + settings["autoCompactEnabled"] = False + # Background summaries for a compaction that will never happen. + settings["precomputeCompactionEnabled"] = False + return settings | self.config.claude_code_settings + + def _env(self, base_url: str, paths: ClaudeCodeRunPaths) -> Dict[str, str]: + model = self.config.model + env = { + "CLAUDE_CONFIG_DIR": paths.config_dir, + # Claude Code appends /v1/messages. + "ANTHROPIC_BASE_URL": base_url, + "ANTHROPIC_API_KEY": PLACEHOLDER_API_KEY, + "ANTHROPIC_MODEL": model, + "ANTHROPIC_DEFAULT_OPUS_MODEL": model, + "ANTHROPIC_DEFAULT_SONNET_MODEL": model, + "ANTHROPIC_DEFAULT_HAIKU_MODEL": model, + "CLAUDE_CODE_SUBAGENT_MODEL": model, + # As root, Claude Code accepts --dangerously-skip-permissions only in a declared sandbox. + "IS_SANDBOX": "1", + # No update checks, telemetry, error reports or feature-flag fetches. + "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", + "DISABLE_AUTOUPDATER": "1", + "DISABLE_TELEMETRY": "1", + "DISABLE_ERROR_REPORTING": "1", + # Plain Messages requests: no beta parameters or headers the Gym converter would drop. + "CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "1", + # Otherwise every request's system prompt carries a per-request attribution line, and a + # system prompt that changes between calls breaks the capture's call chain. + "CLAUDE_CODE_ATTRIBUTION_HEADER": "0", + } + if self.config.max_output_tokens is not None: + env["CLAUDE_CODE_MAX_OUTPUT_TOKENS"] = str(self.config.max_output_tokens) + if self.config.api_timeout_ms is not None: + env["API_TIMEOUT_MS"] = str(self.config.api_timeout_ms) + if self.config.stream_first_byte_timeout_ms is not None: + env["CLAUDE_STREAM_FIRST_BYTE_TIMEOUT_MS"] = str(self.config.stream_first_byte_timeout_ms) + if not self.config.auto_compact: + # DISABLE_COMPACT also removes manual /compact and Claude Code's own context cap, so + # requests run until the model server rejects one, as with compaction-free harnesses. + env["DISABLE_COMPACT"] = "1" + env["DISABLE_AUTO_COMPACT"] = "1" + if not self.config.auto_memory: + env["CLAUDE_CODE_DISABLE_AUTO_MEMORY"] = "1" + return env | self.config.claude_code_env + + def _install_command(self, paths: ClaudeCodeRunPaths) -> str: + version = self.config.claude_code_version + binary = quote(paths.binary) + if self.config.remote_claude_code_binary_path: + glibc = quote(self.config.remote_claude_code_binary_path) + if self.config.remote_claude_code_musl_binary_path: + musl_branch = f"src={quote(self.config.remote_claude_code_musl_binary_path)}" + else: + musl_branch = ( + 'echo "Claude Code: musl-based sandbox, but remote_claude_code_musl_binary_path is not set" >&2' + " && false" + ) + install = ( + "if [ -f /lib/libc.musl-x86_64.so.1 ] || [ -f /lib/libc.musl-aarch64.so.1 ]; " + f"then {musl_branch}; else src={glibc}; fi" + f' && cp "$src" {binary} && chmod 755 {binary}' + ) + else: + print( + "Installing Claude Code from the official installer inside the sandbox. " + "Consider staging the binary and setting remote_claude_code_binary_path instead!", + file=sys.stderr, + ) + installer = quote(f"{paths.root}/install.sh") + install = ( + f"curl -fsSL {quote(OFFICIAL_INSTALLER_URL)} -o {installer}" + f" && bash {installer} {quote(version)}" + f' && cp "$HOME/.local/bin/claude" {binary} && chmod 755 {binary}' + ) + # `claude --version` prints " (Claude Code)". + version_check = ( + f'case " $({binary} --version) " in *{quote(" " + version + " ")}*) ;; ' + f'*) echo "Claude Code: {binary} does not report version {version}" >&2 && false ;; esac' + ) + return f"{install} && {version_check}" + + def _claude_args(self, paths: ClaudeCodeRunPaths, query: str) -> List[str]: + args = [ + "-p", + "--output-format", + "stream-json", + "--verbose", + "--dangerously-skip-permissions", + "--settings", + paths.settings, + "--model", + self.config.model, + ] + if self.config.bare: + args.append("--bare") + if self.config.setting_sources is not None: + args += ["--setting-sources", self.config.setting_sources] + if self.config.allowed_tools: + args += ["--allowedTools", self.config.allowed_tools] + if self.config.disallowed_tools: + args += ["--disallowedTools", self.config.disallowed_tools] + if self.config.max_turns is not None: + args += ["--max-turns", str(self.config.max_turns)] + if self.config.append_system_prompt: + args += ["--append-system-prompt", self.config.append_system_prompt] + # `--` ends the options, so a variadic option above cannot swallow the prompt. + return args + ["--", query] + + def build_command(self, base_url: str, query: str, paths: ClaudeCodeRunPaths) -> str: + env = " ".join(f"{name}={quote(value)}" for name, value in self._env(base_url, paths).items()) + args = " ".join(quote(arg) for arg in self._claude_args(paths, query)) + deadline_s = int(self.config.sandbox_timeout - self.config.harness_timeout_margin_s) + limit = ( + f"if command -v timeout >/dev/null 2>&1; then limit='timeout -k 30 {deadline_s}'; else limit=''; fi && " + if deadline_s > 0 + else "limit='' && " + ) + return ( + 'echo "Shell: $SHELL"' + f" && mkdir -p {quote(paths.root + '/bin')} {quote(paths.config_dir)}" + f" && {self._install_command(paths)}" + ' && echo "Installed Claude Code"' + f" && printf '%s' {quote(json.dumps(self._settings()))} > {quote(paths.settings)}" + f" && {limit}{env} $limit {quote(paths.binary)} {args} < /dev/null > {quote(paths.stream)}" + f' && echo "{FINISHED_MARKER}"' + ) + + async def _model_base_url(self, request: Request) -> str: + """The model server URL for this rollout; Claude Code appends ``/v1/messages`` itself.""" + return self.base_url_for_run( + base_url=sandbox_server_url(self.config.model_server.name), + body=await request.json(), + ) + + async def _download_stream( + self, sandbox: AsyncSandbox, paths: ClaudeCodeRunPaths, local_path: Path + ) -> Optional[ClaudeCodeStream]: + try: + await sandbox.download(paths.stream, local_path) + except Exception: + print(f"Failed to download the Claude Code stream {paths.stream}", format_exc(), file=sys.stderr) + return None + if not local_path.is_file(): + return None + with local_path.open(encoding="utf-8", errors="replace") as lines: + return parse_claude_code_stream(lines) + + async def _collect_observations( + self, + sandbox: AsyncSandbox, + paths: ClaudeCodeRunPaths, + stream: Optional[ClaudeCodeStream], + observation_invocation_id: Optional[str], + ) -> AgentObservationBundle: + result = stream.result if stream is not None else None + status, error_type = invocation_outcome(result) + duration_ms = result.get("duration_ms") if result is not None else None + try: + archived = await sandbox.exec( + command=f"tar -C {quote(paths.config_dir)} -cf {quote(paths.transcripts_tar)} projects" + ) + if archived.return_code != 0 or archived.error_type is not None: + raise RuntimeError(f"archiving the Claude Code transcripts failed: {archived.stderr}") + with tempfile.TemporaryDirectory(prefix="nemo-gym-claude-code-") as local_dir: + local_tar = Path(local_dir) / TRANSCRIPTS_TAR_FNAME + config_dir = Path(local_dir) / "config" + await sandbox.download(paths.transcripts_tar, local_tar) + with tarfile.open(local_tar) as archive: + archive.extractall(config_dir, filter="data") + return extract_claude_code_observations( + config_dir, + model_ref=self.config.model_server, + root_status=status, + root_duration_ms=duration_ms if isinstance(duration_ms, (int, float)) else None, + root_error_type=error_type, + compaction_attempts=stream.compaction_attempts if stream is not None else None, + ) + except Exception: + print("Failed to capture Claude Code observations", format_exc(), file=sys.stderr) + return AgentObservationBundle( + source="claude_code", + records=( + [AgentInvocation(invocation_id=observation_invocation_id)] if observation_invocation_id else [] + ), + gaps=[ + ObservationGap(code="agent_artifact_unavailable"), + ObservationGap(code="agent_transcript_unavailable"), + ObservationGap(code="observation_capture_failed"), + ], + ) + + async def responses( + self, + request: Request, + body: NeMoGymResponseCreateParamsNonStreaming = Body(), + ) -> NeMoGymResponse: + if "sandbox_id" not in request.cookies: + raise ValueError("Use /run: Claude Code needs the sandbox the resources server seeded") + sandbox = self._sandbox_id_to_sandbox[request.cookies["sandbox_id"]] + + query = None + # This can be modified to handle system/developer prompts too. + for input_item in body.input: + if input_item.role == "user": + assert not query, body.input + if isinstance(input_item.content, str): + query = input_item.content + elif isinstance(input_item.content, list): + assert len(input_item.content) == 1, body.input + query = input_item.content[0]["text"] + + assert query, body.input + + paths = self._new_run_paths() + command = self.build_command(await self._model_base_url(request), query, paths) + if self.config.debug: + print(f"Running command:\n```bash\n{command}\n```\n", file=sys.stderr) + + run_error_type = None + try: + result = await sandbox.exec( + command=command, + timeout_s=self.config.sandbox_timeout, + ) + except Exception as exc: + result = None + run_error_type = type(exc).__name__ + print("Claude Code exec hit error.", format_exc(), file=sys.stderr) + + if self.config.debug and result: + print("Claude Code install and run stdout:\n", result.stdout, file=sys.stderr) + print("Claude Code install and run stderr:\n", result.stderr, file=sys.stderr) + + results_dir = self._results_dir(request) + local_stream = results_dir / STREAM_FNAME + local_stream.unlink(missing_ok=True) + stream = await self._download_stream(sandbox, paths, local_stream) + if stream is not None: + if stream.compaction_boundaries and not self.config.auto_compact: + print( + f"Claude Code compacted {stream.compaction_boundaries} time(s) although auto_compact is off", + file=sys.stderr, + ) + if stream.invalid_lines: + print(f"Claude Code stream had {stream.invalid_lines} unparseable line(s)", file=sys.stderr) + + observation_invocation_id = getattr(request.state, "_ng_observation_invocation_id", None) + observation_invocation_id = observation_invocation_id if isinstance(observation_invocation_id, str) else None + collect_observations = observation_invocation_id is not None + observations = None + if collect_observations: + observations = await self._collect_observations(sandbox, paths, stream, observation_invocation_id) + + result_stdout = (result.stdout if result else "") or "" + result_stderr = (result.stderr if result else "") or "" + finished = False + std_out_split = result_stdout.rsplit("Shell: ", maxsplit=1) + if len(std_out_split) > 1: + finished = FINISHED_MARKER in std_out_split[1] + return_code = getattr(result, "return_code", None) + error_type = getattr(result, "error_type", None) or run_error_type + + if collect_observations and observations is not None: + agent_sandbox_observation = self._agent_sandbox_observation( + sandbox=sandbox, + return_code=return_code, + error_type=error_type, + finished=finished, + ) + for record in observations.records: + if isinstance(record, ToolCallObservation): + record.sandbox_id = agent_sandbox_observation.sandbox_id + elif isinstance(record, AgentInvocation) and record.parent_invocation_id is None: + status = { + "completed": "completed", + "failed": "failed", + "sandbox_error": "failed", + "timeout": "incomplete", + "cancelled": "incomplete", + }.get(agent_sandbox_observation.outcome) + if status is not None: + record.status = status + observations.records.append(agent_sandbox_observation) + observations.gaps.append(ObservationGap(code="sandbox_lifecycle_timing_unavailable")) + + export_found = stream is not None + stream_result = stream.result if stream is not None else None + outcome, outcome_error = invocation_outcome(stream_result) + duration_ms = stream_result.get("duration_ms") if stream_result is not None else None + run_result: Dict[str, Any] = { + "claude_code_failed": bool(error_type) + or return_code != 0 + or not finished + or not export_found + or outcome == "failed", + "claude_code_exit_code": return_code, + "claude_code_error_type": error_type or (outcome_error if outcome == "failed" else None), + "claude_code_results_fpath": str(local_stream) if export_found else "", + "claude_code_run_stdout": result_stdout, + "claude_code_run_stderr": result_stderr, + "claude_code_export_found": export_found, + "claude_code_finished": finished, + "claude_code_result_success": outcome == "completed", + "claude_code_result_error": outcome == "failed", + "claude_code_result_missing": stream_result is None, + "claude_code_context_overflow": context_overflow(stream_result), + "claude_code_duration_s": ( + duration_ms / 1000 + if isinstance(duration_ms, (int, float)) and not isinstance(duration_ms, bool) + else None + ), + "claude_code_main_turns": stream.main_turns if stream is not None else 0, + "claude_code_subagent_turns": stream.subagent_turns if stream is not None else 0, + "claude_code_subagent_calls": stream.subagent_calls if stream is not None else 0, + "claude_code_tool_calls": stream.tool_calls if stream is not None else 0, + "claude_code_tool_errors": stream.tool_errors if stream is not None else 0, + "claude_code_unknown_tool_calls": stream.unknown_tool_calls if stream is not None else 0, + } + if collect_observations: + run_result["_ng_agent_observations"] = observations + self._sandbox_id_to_run_result[request.cookies["sandbox_id"]] = run_result + + response = NeMoGymResponse( + id=f"resp_{uuid4().hex}", + created_at=int(time()), + model=body.model or self.config.model_server.name, + object="response", + output=stream.output if stream is not None else [], + tool_choice=body.tool_choice, + tools=body.tools, + parallel_tool_calls=body.parallel_tool_calls, + usage=stream.usage if stream is not None else None, + ) + # Written before grading, so a failed or interrupted verification still leaves the + # generation and how it ended on disk. + receipt = { + "response": response.model_dump(mode="json"), + "execution": {key: value for key, value in run_result.items() if not key.startswith("_ng_")}, + } + pending = results_dir / "generation.json.partial" + pending.write_text(json.dumps(receipt)) + pending.replace(results_dir / "generation.json") + return response + + async def run( + self, request: Request, body: ClaudeCodeSandboxedAgentRunRequest + ) -> ClaudeCodeSandboxedAgentVerifyResponse: + async with self._sem: + return await self._run(request, body) + + async def _run( + self, request: Request, body: ClaudeCodeSandboxedAgentRunRequest + ) -> ClaudeCodeSandboxedAgentVerifyResponse: + cookies = request.cookies + session_key = request.session[SESSION_ID_KEY] + rollout_id = self.rollout_id_from_run(body) + + seed_session_response = await self.server_client.post( + server_name=self.config.resources_server.name, + url_path="/seed_session", + json=body.model_dump(), + cookies=cookies, + ) + await raise_for_status(seed_session_response) + cookies = cookies | seed_session_response.cookies + + # "sandbox_handle" and "workdir" come from the SWE resources servers' seed_session responses. + seed_session_result = await seed_session_response.json() + sandbox = await self._start_sandbox( + sandbox_id=seed_session_result.get("sandbox_handle"), + workdir=seed_session_result.get("workdir"), + ) + self._sandbox_id_to_sandbox[session_key] = sandbox + + # Propagating the sandbox handle + cookies["sandbox_id"] = session_key + + request._cookies = cookies + request.state._ng_observation_invocation_id = rollout_id + observations = None + try: + response = await self.responses(request, body.responses_create_params) + run_result = self._sandbox_id_to_run_result.get(session_key, {}).copy() + observations = run_result.pop("_ng_agent_observations", None) + response_dict = await verify_agent_response( + self.server_client, + self.config.resources_server, + body, + response, + cookies, + force_zero_reward=self.config.execution_failure_reward_zero + and run_result.get("claude_code_failed", False), + ) + finally: + del request.state._ng_observation_invocation_id + try: + await sandbox.stop() + except Exception: + print("Failed to stop sandbox", format_exc(), file=sys.stderr) + finally: + self._sandbox_id_to_sandbox.pop(session_key, None) + self._sandbox_id_to_run_result.pop(session_key, None) + + response_dict |= run_result + raw_verifier_sandbox_observation = response_dict.pop("verifier_sandbox_observation", None) + if rollout_id is not None: + if observations is None: + observations = AgentObservationBundle( + source="claude_code", + records=[AgentInvocation(invocation_id=rollout_id)], + gaps=[ObservationGap(code="observation_capture_failed")], + ) + if raw_verifier_sandbox_observation is not None: + try: + verifier_observation = SandboxObservation.model_validate(raw_verifier_sandbox_observation) + if verifier_observation.role != "verifier": + raise ValueError("resources server returned a non-verifier sandbox observation") + observations.records.append(verifier_observation) + except Exception: + observations.gaps.append(ObservationGap(code="verifier_sandbox_observation_invalid")) + else: + observations.gaps.append(ObservationGap(code="verifier_sandbox_observation_unavailable")) + response_dict["ng_agent_observations"] = observations.model_dump(mode="json") + return ClaudeCodeSandboxedAgentVerifyResponse.model_validate(response_dict) + + +if __name__ == "__main__": + ClaudeCodeSandboxedAgent.run_webserver() +elif is_nemo_gym_fastapi_entrypoint(__file__): + app = ClaudeCodeSandboxedAgent.run_webserver() # noqa: F401 diff --git a/responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml b/responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml new file mode 100644 index 00000000000..a1d541b6532 --- /dev/null +++ b/responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml @@ -0,0 +1,74 @@ +claude_code_sandboxed_agent: + responses_api_agents: + claude_code_sandboxed_agent: + entrypoint: app.py + num_workers: 4 + resources_server: + type: resources_servers + name: ??? + model_server: + type: responses_api_models + name: policy_model + # Claude Code drives its own model calls and returns no token ids. + token_id_capture: true + + debug: false + + claude_code_version: 2.1.287 + # Staged binaries inside the sandbox; null runs the official installer in the sandbox. + remote_claude_code_binary_path: null + remote_claude_code_musl_binary_path: null + remote_work_dir: /tmp + + model: policy + auto_compact: true + # No notes across sessions: each rollout starts from an empty config directory. + auto_memory: false + # --bare is Claude Code's minimal mode (one-line system prompt, Bash/Read/Edit only). + bare: false + # Only this rollout's empty user settings: a task repository's .claude/ cannot change the run. + setting_sources: user + allowed_tools: null + # Tools that do not fit a one-shot run in a task sandbox. No web access and no skills, as for + # the OpenCode agent. Worktrees would move the edits out of the repository the verifier diffs. + # Scheduled prompts, wakeups, workflows and agent messaging outlive or bypass a single `-p` run; + # ReportFindings is the code-review report tool. Agent (subagents), Bash, Read, Edit, Write, + # NotebookEdit and TaskStop stay. + disallowed_tools: >- + WebFetch,WebSearch,Skill,EnterWorktree,ExitWorktree,CronCreate,CronDelete,CronList,ScheduleWakeup,Workflow,SendMessage,ListAgents,ReportFindings + max_turns: null + append_system_prompt: null + # Claude Code's ceiling for models it does not recognize (128k as of 2.1.287). The model + # server still clamps each call to the context that remains. + max_output_tokens: 128000 + # A Gym model server sends the headers of a streamed reply only after generation finishes, + # so Claude Code's default 3-5 minute wait for headers would abort and retry long calls. + # 30 minutes is the most Claude Code accepts. + api_timeout_ms: 3600000 + stream_first_byte_timeout_ms: 1800000 + harness_timeout_margin_s: 120 + claude_code_settings: {} + claude_code_env: {} + + sandbox_timeout: 10800 # 3hr default for harness execution + + sandbox_provider: sandbox + 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 + # 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: {} + metadata: + benchmark: swebench-verified + harness: claude-code + + datasets: [] diff --git a/responses_api_agents/claude_code_sandboxed_agent/requirements.txt b/responses_api_agents/claude_code_sandboxed_agent/requirements.txt new file mode 100644 index 00000000000..eae22a6b0ca --- /dev/null +++ b/responses_api_agents/claude_code_sandboxed_agent/requirements.txt @@ -0,0 +1 @@ +-e nemo-gym[dev,sandbox] @ ../../ diff --git a/responses_api_agents/claude_code_sandboxed_agent/tests/__init__.py b/responses_api_agents/claude_code_sandboxed_agent/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/responses_api_agents/claude_code_sandboxed_agent/tests/test_app.py b/responses_api_agents/claude_code_sandboxed_agent/tests/test_app.py new file mode 100644 index 00000000000..224e8f99bd1 --- /dev/null +++ b/responses_api_agents/claude_code_sandboxed_agent/tests/test_app.py @@ -0,0 +1,846 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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 io +import json +import shutil +import subprocess +import tarfile +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Dict, List +from unittest.mock import AsyncMock, MagicMock + +import yaml +from pytest import MonkeyPatch, fixture, mark, raises + +from nemo_gym import PARENT_DIR +from nemo_gym.config_types import ModelServerRef, ResourcesServerRef +from nemo_gym.failure_kinds import AGENT_RUN_ERROR +from nemo_gym.openai_utils import ( + NeMoGymFunctionCallOutput, + NeMoGymResponseFunctionToolCall, + NeMoGymResponseInputTokensDetails, + NeMoGymResponseOutputMessage, + NeMoGymResponseOutputText, + NeMoGymResponseOutputTokensDetails, + NeMoGymResponseReasoningItem, + NeMoGymResponseUsage, + NeMoGymSummary, +) +from nemo_gym.rollout_observability import AgentInvocation, SandboxObservation +from nemo_gym.server_utils import SESSION_ID_KEY, ServerClient +from responses_api_agents.claude_code_sandboxed_agent import app as app_module +from responses_api_agents.claude_code_sandboxed_agent.app import ( + FINISHED_MARKER, + ClaudeCodeRunPaths, + ClaudeCodeSandboxedAgent, + ClaudeCodeSandboxedAgentConfig, + ClaudeCodeSandboxedAgentRunRequest, + context_overflow, + invocation_outcome, + parse_claude_code_stream, +) + + +AGENT_DIR = Path(app_module.__file__).parent +VERSION = "2.1.287" +QUERY = "Fix the bug.\nIt's in `parse()`: don't use $(rm -rf /) or \"quotes\" \\ wrongly." +BODY = { + "responses_create_params": {"input": [{"role": "user", "content": QUERY}]}, + "_ng_task_index": 7, + "_ng_rollout_index": 2, +} +CAPTURE_CONFIG = {"token_id_capture": {"enabled": True, "all_agents": False}} +UNKNOWN_TOOL = ( + "Error: No such tool available: read. Tool names are case-sensitive: call Read instead." + "" +) + + +def _config(**overrides: Any) -> ClaudeCodeSandboxedAgentConfig: + values: Dict[str, Any] = dict( + host="0.0.0.0", + port=8080, + entrypoint="", + name="claude_code_agent", + resources_server=ResourcesServerRef(type="resources_servers", name="task_server"), + model_server=ModelServerRef(type="responses_api_models", name="policy_model"), + claude_code_version=VERSION, + sandbox_provider="sandbox", + sandbox_config={}, + sandbox_timeout=600, + token_id_capture=True, + ) + values.update(overrides) + return ClaudeCodeSandboxedAgentConfig(**values) + + +def _agent(global_config: Dict[str, Any] | None = None, **overrides: Any) -> ClaudeCodeSandboxedAgent: + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = global_config or {} + return ClaudeCodeSandboxedAgent(config=_config(**overrides), server_client=server_client) + + +def _event(event_type: str, **fields: Any) -> Dict[str, Any]: + return {"type": event_type, "session_id": "s1", "parent_tool_use_id": None, **fields} + + +def _assistant(message_id: str, block: Dict[str, Any], usage: Dict[str, int], **fields: Any) -> Dict[str, Any]: + message = {"id": message_id, "model": "policy", "role": "assistant", "content": [block], "usage": usage} + return _event("assistant", message=message, **fields) + + +def _tool_result(call_id: str, content: Any, **fields: Any) -> Dict[str, Any]: + is_error = fields.pop("is_error", None) + block = {"type": "tool_result", "tool_use_id": call_id, "content": content} + if is_error is not None: + block["is_error"] = is_error + return _event("user", message={"role": "user", "content": [block]}, **fields) + + +MSG1_USAGE = {"input_tokens": 10, "output_tokens": 5} +STREAM_EVENTS: List[Dict[str, Any]] = [ + _event("system", subtype="init", model="policy", tools=["Bash", "Task"]), + _assistant("msg_1", {"type": "thinking", "thinking": "I should look.", "signature": ""}, MSG1_USAGE), + _assistant("msg_1", {"type": "text", "text": "\n\nLet me check. "}, MSG1_USAGE), + _assistant("msg_1", {"type": "tool_use", "id": "call_1", "name": "Bash", "input": {"command": "ls"}}, MSG1_USAGE), + _tool_result("call_1", "a.py\nb.py"), + _assistant( + "msg_2", + {"type": "tool_use", "id": "call_2", "name": "Task", "input": {"prompt": "explore", "description": "x"}}, + {"input_tokens": 20, "output_tokens": 3}, + ), + _assistant( + "msg_sub", + {"type": "text", "text": "subagent text"}, + {"input_tokens": 7, "output_tokens": 1}, + parent_tool_use_id="call_2", + ), + _tool_result("call_sub", "subagent tool output", parent_tool_use_id="call_2"), + _tool_result("call_2", [{"type": "text", "text": "sub done"}, {"type": "text", "text": "really"}]), + _event("system", subtype="status", status="compacting"), + _event("system", subtype="status", status=None, compact_result="success"), + _event( + "assistant", + message={ + "id": "msg_synthetic", + "model": "", + "role": "assistant", + "content": [{"type": "text", "text": "API Error: boom"}], + "usage": {"input_tokens": 0, "output_tokens": 0}, + }, + ), + _assistant("msg_3", {"type": "text", "text": "Fixed."}, {"input_tokens": 30, "output_tokens": 2}), +] +RESULT_EVENT = _event( + "result", + subtype="success", + is_error=False, + duration_ms=1234, + num_turns=3, + result="Fixed.", + usage={"input_tokens": 100, "cache_read_input_tokens": 50, "cache_creation_input_tokens": 10, "output_tokens": 20}, +) +OVERFLOW_EVENT = _event("result", subtype="success", is_error=True, duration_ms=99000, result="Prompt is too long") + + +def _stream_text(events: List[Dict[str, Any]], *extra_lines: str) -> str: + return "\n".join([json.dumps(event) for event in events] + list(extra_lines)) + "\n" + + +EXPECTED_OUTPUT = [ + NeMoGymResponseReasoningItem( + id="rs_x", summary=[NeMoGymSummary(text="I should look.", type="summary_text")], type="reasoning" + ), + NeMoGymResponseOutputMessage( + id="msg_x", + content=[NeMoGymResponseOutputText(text="\n\nLet me check. ", annotations=[], type="output_text")], + role="assistant", + status="completed", + type="message", + ), + NeMoGymResponseFunctionToolCall( + arguments='{"command": "ls"}', call_id="call_1", name="Bash", type="function_call", status="completed" + ), + NeMoGymFunctionCallOutput(call_id="call_1", output="a.py\nb.py", type="function_call_output"), + NeMoGymResponseFunctionToolCall( + arguments='{"prompt": "explore", "description": "x"}', + call_id="call_2", + name="Task", + type="function_call", + status="completed", + ), + NeMoGymFunctionCallOutput(call_id="call_2", output="sub done\nreally", type="function_call_output"), + NeMoGymResponseOutputMessage( + id="msg_x", + content=[NeMoGymResponseOutputText(text="Fixed.", annotations=[], type="output_text")], + role="assistant", + status="completed", + type="message", + ), +] + + +@fixture +def fixed_uuid(monkeypatch: MonkeyPatch) -> None: + monkeypatch.setattr(app_module, "uuid4", lambda: SimpleNamespace(hex="x")) + + +@fixture +def model_server_url(monkeypatch: MonkeyPatch) -> None: + monkeypatch.setattr(app_module, "sandbox_server_url", lambda _name, require_reachable=False: "http://model-server") + + +class TestParseStream: + def test_main_conversation_only_and_session_usage(self, fixed_uuid: None) -> None: + stream = parse_claude_code_stream(_stream_text([*STREAM_EVENTS, RESULT_EVENT], "not json").splitlines()) + + assert stream.output == EXPECTED_OUTPUT + assert stream.usage == NeMoGymResponseUsage( + input_tokens=160, + input_tokens_details=NeMoGymResponseInputTokensDetails(cached_tokens=50), + output_tokens=20, + output_tokens_details=NeMoGymResponseOutputTokensDetails(reasoning_tokens=0), + total_tokens=180, + ) + assert stream.result is not None and stream.result["subtype"] == "success" + assert stream.compaction_attempts == [] + assert stream.compaction_boundaries == 0 + assert stream.invalid_lines == 1 + # msg_1, msg_2 and msg_3 are the main conversation; the synthetic message is nobody's turn. + assert (stream.main_turns, stream.subagent_turns, stream.subagent_calls) == (3, 1, 1) + assert (stream.tool_calls, stream.tool_errors, stream.unknown_tool_calls) == (2, 0, 0) + + def test_without_result_counts_each_message_once(self, fixed_uuid: None) -> None: + stream = parse_claude_code_stream(_stream_text(STREAM_EVENTS).splitlines()) + + assert stream.result is None + assert stream.output == EXPECTED_OUTPUT + # msg_1 appears in three events but is counted once; the synthetic message not at all. + assert stream.usage is not None + assert (stream.usage.input_tokens, stream.usage.output_tokens) == (10 + 20 + 7 + 30, 5 + 3 + 1 + 2) + + def test_compaction_is_reported(self) -> None: + events = [ + _event("system", subtype="status", status="compacting"), + _event("system", subtype="compact_boundary"), + _event("system", subtype="status", status="compacting", session_id="s2"), + _event("system", subtype="status", status=None, compact_result="failed", session_id="s2"), + _event("system", subtype="status", status="compacting", session_id="s3"), + ] + stream = parse_claude_code_stream(_stream_text(events).splitlines()) + + assert stream.compaction_boundaries == 1 + assert sorted(stream.compaction_attempts, key=lambda a: a["invocation_id"]) == [ + {"invocation_id": "s1", "outcome": "unknown"}, + {"invocation_id": "s2", "outcome": "failed"}, + {"invocation_id": "s3", "outcome": "unknown"}, + ] + assert stream.output == [] and stream.usage is None + + def test_refused_tool_calls_are_counted_in_every_conversation(self, fixed_uuid: None) -> None: + events = [ + _assistant("msg_1", {"type": "tool_use", "id": "c1", "name": "read", "input": {"path": "a"}}, MSG1_USAGE), + _tool_result("c1", UNKNOWN_TOOL), + _assistant( + "msg_sub", + {"type": "tool_use", "id": "c2", "name": "Bash", "input": {"command": "false"}}, + MSG1_USAGE, + parent_tool_use_id="c0", + ), + _tool_result("c2", "Exit code 1", is_error=True, parent_tool_use_id="c0"), + _tool_result("c3", "fine", parent_tool_use_id="c0"), + ] + stream = parse_claude_code_stream(_stream_text(events).splitlines()) + + assert (stream.tool_calls, stream.tool_errors, stream.unknown_tool_calls) == (2, 2, 1) + assert (stream.main_turns, stream.subagent_turns, stream.subagent_calls) == (1, 1, 0) + assert [item.type for item in stream.output] == ["function_call", "function_call_output"] + + @mark.parametrize( + ("result", "expected"), + [ + (None, ("incomplete", "result_missing")), + ({"subtype": "success", "is_error": False}, ("completed", None)), + ({"subtype": "error_max_turns", "is_error": True}, ("incomplete", "error_max_turns")), + ({"subtype": "error_during_execution", "is_error": True}, ("failed", "error_during_execution")), + ({"subtype": "success", "is_error": True, "result": "API Error: 500"}, ("failed", "agent_error")), + ({"subtype": "success", "is_error": True, "result": "Prompt is too long"}, ("failed", "context_overflow")), + ], + ) + def test_invocation_outcome(self, result: Any, expected: Any) -> None: + assert invocation_outcome(result) == expected + + def test_context_overflow_needs_an_error_result(self) -> None: + assert context_overflow(None) is False + assert context_overflow({"subtype": "success", "is_error": False, "result": "Prompt is too long"}) is False + assert context_overflow({"subtype": "success", "is_error": True, "result": " Prompt is too long"}) is True + + +class TestCommand: + def test_training_settings(self) -> None: + agent = _agent( + auto_compact=False, + remote_claude_code_binary_path="/mnt/s3/claude", + max_output_tokens=128000, + api_timeout_ms=3600000, + stream_first_byte_timeout_ms=1800000, + disallowed_tools="WebFetch,WebSearch", + ) + paths = ClaudeCodeRunPaths(root="/tmp/run") + env = agent._env("http://model:5/ng-rollout/7-2/training-token-capture", paths) + + assert env["ANTHROPIC_BASE_URL"] == "http://model:5/ng-rollout/7-2/training-token-capture" + assert env["CLAUDE_CONFIG_DIR"] == "/tmp/run/config" + for name in ( + "ANTHROPIC_MODEL", + "ANTHROPIC_DEFAULT_OPUS_MODEL", + "ANTHROPIC_DEFAULT_SONNET_MODEL", + "ANTHROPIC_DEFAULT_HAIKU_MODEL", + "CLAUDE_CODE_SUBAGENT_MODEL", + ): + assert env[name] == "policy" + assert env["DISABLE_COMPACT"] == env["DISABLE_AUTO_COMPACT"] == "1" + assert env["CLAUDE_CODE_DISABLE_AUTO_MEMORY"] == "1" + assert env["CLAUDE_CODE_ATTRIBUTION_HEADER"] == "0" + assert env["CLAUDE_CODE_MAX_OUTPUT_TOKENS"] == "128000" + assert env["API_TIMEOUT_MS"] == "3600000" + assert env["CLAUDE_STREAM_FIRST_BYTE_TIMEOUT_MS"] == "1800000" + assert agent._settings() == {"autoCompactEnabled": False, "precomputeCompactionEnabled": False} + + args = agent._claude_args(paths, QUERY) + assert args[:5] == ["-p", "--output-format", "stream-json", "--verbose", "--dangerously-skip-permissions"] + assert args[args.index("--settings") + 1] == "/tmp/run/settings.json" + assert args[args.index("--disallowedTools") + 1] == "WebFetch,WebSearch" + assert args[args.index("--setting-sources") + 1] == "user" + assert "--bare" not in args and "--max-turns" not in args + assert args[-2:] == ["--", QUERY] + + def test_bare_and_setting_sources_are_optional(self) -> None: + paths = ClaudeCodeRunPaths(root="/tmp/run") + args = _agent(bare=True, setting_sources=None, max_turns=50)._claude_args(paths, QUERY) + assert "--bare" in args and "--setting-sources" not in args + assert args[args.index("--max-turns") + 1] == "50" + assert args[-2:] == ["--", QUERY] + + def test_defaults_leave_compaction_on(self) -> None: + agent = _agent() + env = agent._env("http://model:5", ClaudeCodeRunPaths(root="/tmp/run")) + + assert "DISABLE_COMPACT" not in env and "DISABLE_AUTO_COMPACT" not in env + assert "CLAUDE_CODE_MAX_OUTPUT_TOKENS" not in env and "API_TIMEOUT_MS" not in env + assert agent._settings() == {} + + def test_overrides_are_layered_last(self) -> None: + agent = _agent( + auto_compact=False, + claude_code_env={"DISABLE_COMPACT": "0", "EXTRA": "1"}, + claude_code_settings={"autoCompactEnabled": True, "includeCoAuthoredBy": False}, + ) + env = agent._env("http://model:5", ClaudeCodeRunPaths(root="/tmp/run")) + + assert (env["DISABLE_COMPACT"], env["EXTRA"]) == ("0", "1") + assert agent._settings() == { + "autoCompactEnabled": True, + "precomputeCompactionEnabled": False, + "includeCoAuthoredBy": False, + } + + def test_without_staged_binary_runs_the_official_installer(self) -> None: + command = _agent().build_command("http://model:5", QUERY, ClaudeCodeRunPaths(root="/tmp/run")) + assert "curl -fsSL https://claude.ai/install.sh -o /tmp/run/install.sh" in command + assert f"bash /tmp/run/install.sh {VERSION}" in command + + def test_musl_sandbox_without_musl_binary_fails_clearly(self) -> None: + agent = _agent(remote_claude_code_binary_path="/mnt/s3/claude") + command = agent.build_command("http://model:5", QUERY, ClaudeCodeRunPaths(root="/tmp/run")) + assert "remote_claude_code_musl_binary_path is not set" in command + agent = _agent(remote_claude_code_binary_path="/mnt/a", remote_claude_code_musl_binary_path="/mnt/b") + command = agent.build_command("http://model:5", QUERY, ClaudeCodeRunPaths(root="/tmp/run")) + assert "then src=/mnt/b; else src=/mnt/a; fi" in command + + @mark.parametrize( + ("observability_enabled", "token_capture_enabled", "expected_base_url"), + [ + (False, False, "http://model-server"), + (True, False, "http://model-server/ng-rollout/7-2"), + (False, True, "http://model-server/ng-rollout/7-2/training-token-capture"), + ], + ids=("disabled", "observability-only", "token-capture"), + ) + async def test_model_base_url_routes_model_calls( + self, + model_server_url: None, + observability_enabled: bool, + token_capture_enabled: bool, + expected_base_url: str, + ) -> None: + agent = _agent( + { + "observability_enabled": observability_enabled, + "token_id_capture": {"enabled": token_capture_enabled, "all_agents": False}, + } + ) + request = MagicMock() + request.json = AsyncMock(return_value=BODY) + + assert await agent._model_base_url(request) == expected_base_url + + +FAKE_CLAUDE = """#!/bin/sh +if [ "$1" = "--version" ]; then echo "{version} (Claude Code)"; exit 0; fi +run_dir=$(dirname "$CLAUDE_CONFIG_DIR") +printf '%s\\0' "$@" > "$run_dir/argv.bin" +env > "$run_dir/env.txt" +readlink /proc/self/fd/0 > "$run_dir/stdin.txt" +pwd > "$run_dir/cwd.txt" +echo '{{"type": "system", "subtype": "init", "session_id": "s1"}}' +echo '{{"type": "result", "subtype": "success", "is_error": false, "session_id": "s1"}}' +exit {exit_code} +""" + + +@mark.skipif(shutil.which("sh") is None, reason="needs a POSIX shell") +class TestCommandRuns: + """Run the generated command in a local shell against a staged fake binary.""" + + def _run(self, tmp_path: Path, *, reported_version: str = VERSION, exit_code: int = 0, **overrides: Any): + staged = tmp_path / "staged" / "claude" + staged.parent.mkdir() + staged.write_text(FAKE_CLAUDE.format(version=reported_version, exit_code=exit_code)) + staged.chmod(0o644) # bucket mounts keep no executable bit + repo = tmp_path / "repo" + repo.mkdir() + agent = _agent( + auto_compact=False, + remote_claude_code_binary_path=str(staged), + remote_work_dir=str(tmp_path / "work"), + **overrides, + ) + paths = agent._new_run_paths() + command = agent.build_command("http://model:5/ng-rollout/7-2/training-token-capture", QUERY, paths) + result = subprocess.run( + ["sh", "-c", command], cwd=repo, capture_output=True, text=True, timeout=60, stdin=subprocess.PIPE + ) + return agent, paths, result, repo + + def test_installs_and_runs_claude(self, tmp_path: Path) -> None: + agent, paths, result, repo = self._run(tmp_path) + + assert result.returncode == 0, result.stderr + assert result.stdout.splitlines()[-1] == FINISHED_MARKER + assert result.stdout.splitlines()[0].startswith("Shell: ") + root = Path(paths.root) + assert (root / "bin" / "claude").stat().st_mode & 0o111 + argv = (root / "argv.bin").read_bytes().split(b"\0")[:-1] + assert [arg.decode() for arg in argv] == agent._claude_args(paths, QUERY) + env = dict(line.split("=", 1) for line in (root / "env.txt").read_text().splitlines() if "=" in line) + assert env["ANTHROPIC_BASE_URL"] == "http://model:5/ng-rollout/7-2/training-token-capture" + assert env["DISABLE_COMPACT"] == "1" and env["IS_SANDBOX"] == "1" + assert (root / "stdin.txt").read_text().strip() == "/dev/null" + assert Path((root / "cwd.txt").read_text().strip()).resolve() == repo.resolve() + assert json.loads((root / "settings.json").read_text()) == agent._settings() + stream = parse_claude_code_stream((root / "stream.jsonl").read_text().splitlines()) + assert stream.result is not None and stream.result["subtype"] == "success" + assert list(repo.iterdir()) == [] # nothing lands in the repository + + def test_version_mismatch_stops_before_the_run(self, tmp_path: Path) -> None: + agent, paths, result, _ = self._run(tmp_path, reported_version="2.1.286") + + assert result.returncode != 0 + assert FINISHED_MARKER not in result.stdout + assert f"does not report version {VERSION}" in result.stderr + assert not Path(paths.stream).exists() + + def test_failed_run_prints_no_marker(self, tmp_path: Path) -> None: + _, _, result, _ = self._run(tmp_path, exit_code=1) + + assert result.returncode == 1 + assert FINISHED_MARKER not in result.stdout + + @mark.skipif(shutil.which("timeout") is None, reason="needs coreutils timeout") + def test_cli_runs_under_a_deadline_before_the_sandbox_timeout(self, tmp_path: Path) -> None: + agent, paths, result, _ = self._run(tmp_path, sandbox_timeout=130, harness_timeout_margin_s=120) + + assert "limit='timeout -k 30 10'" in agent.build_command("http://m", QUERY, paths) + assert result.returncode == 0, result.stderr + + def test_no_deadline_when_the_margin_exceeds_the_timeout(self) -> None: + agent = _agent(sandbox_timeout=60, harness_timeout_margin_s=120) + command = agent.build_command("http://m", QUERY, ClaudeCodeRunPaths(root="/tmp/run")) + assert "limit='' && " in command and "timeout -k" not in command + + +def _exec_result(*, return_code: int = 0, finished: bool = True, error_type: str | None = None) -> SimpleNamespace: + marker = f"\n{FINISHED_MARKER}" if finished else "" + return SimpleNamespace( + stdout=f"Shell: /bin/bash\nInstalled Claude Code{marker}\n", + stderr="", + return_code=return_code, + error_type=error_type, + ) + + +class _FakeSandbox: + """Runs the command, then serves the stream file and transcripts archive like AsyncSandbox.""" + + def __init__( + self, + stream: str | None, + transcripts: bytes | None, + run_result: SimpleNamespace | None = None, + tar_return_code: int = 0, + ) -> None: + self.stream = stream + self.transcripts = transcripts + self.run_result = run_result or _exec_result() + self.tar_result = SimpleNamespace(stdout="", stderr="", return_code=tar_return_code, error_type=None) + self.exec = AsyncMock(side_effect=self._exec) + self.download = AsyncMock(side_effect=self._download) + self.stop = AsyncMock() + + async def _exec(self, command: str, timeout_s: float | None = None) -> SimpleNamespace: + # The run command ends by printing the finished marker; the only other command archives transcripts. + return self.run_result if FINISHED_MARKER in command else self.tar_result + + async def _download(self, remote_path: str, local_path: Path) -> None: + if remote_path.endswith("/stream.jsonl") and self.stream is not None: + Path(local_path).write_text(self.stream) + elif remote_path.endswith("/transcripts.tar") and self.transcripts is not None: + Path(local_path).write_bytes(self.transcripts) + else: + raise FileNotFoundError(remote_path) + + +def _transcripts_tar() -> bytes: + lines = [ + { + "type": "assistant", + "sessionId": "s1", + "uuid": "u1", + "timestamp": "2026-10-01T00:00:00Z", + "message": {"id": "msg_3", "role": "assistant", "content": [{"type": "text", "text": "Fixed."}]}, + } + ] + data = ("\n".join(json.dumps(line) for line in lines) + "\n").encode() + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode="w") as archive: + info = tarfile.TarInfo("projects/-repo/s1.jsonl") + info.size = len(data) + archive.addfile(info, io.BytesIO(data)) + return buffer.getvalue() + + +def _request(observe: bool = True, cookies: Dict[str, str] | None = None) -> Any: + state = SimpleNamespace(_ng_observation_invocation_id="7-2") if observe else SimpleNamespace() + return SimpleNamespace( + cookies={"sandbox_id": "session-1"} if cookies is None else cookies, + session={SESSION_ID_KEY: "session-1"}, + state=state, + json=AsyncMock(return_value=BODY), + ) + + +def _params() -> Any: + return ClaudeCodeSandboxedAgentRunRequest.model_validate(BODY).responses_create_params + + +class TestResponses: + @fixture(autouse=True) + def _results_in_tmp(self, tmp_path: Path, monkeypatch: MonkeyPatch, model_server_url: None) -> None: + monkeypatch.setattr(app_module, "__file__", str(tmp_path / "app.py")) + + def _agent_with(self, sandbox: _FakeSandbox, **overrides: Any) -> ClaudeCodeSandboxedAgent: + agent = _agent( + CAPTURE_CONFIG, auto_compact=False, remote_claude_code_binary_path="/mnt/s3/claude", **overrides + ) + agent._sandbox_id_to_sandbox["session-1"] = sandbox + return agent + + async def test_transcript_receipt_and_observations(self, tmp_path: Path, fixed_uuid: None) -> None: + sandbox = _FakeSandbox(_stream_text([*STREAM_EVENTS, RESULT_EVENT]), _transcripts_tar()) + agent = self._agent_with(sandbox) + + response = await agent.responses(_request(), _params()) + + assert response.output == EXPECTED_OUTPUT + assert response.usage is not None and response.usage.total_tokens == 180 + command = sandbox.exec.await_args_list[0].kwargs["command"] + assert "ANTHROPIC_BASE_URL=http://model-server/ng-rollout/7-2/training-token-capture " in command + assert sandbox.exec.await_args_list[0].kwargs["timeout_s"] == 600 + assert sandbox.exec.await_args_list[1].kwargs["command"].startswith("tar -C ") + + run_result = agent._sandbox_id_to_run_result["session-1"] + observations = run_result.pop("_ng_agent_observations") + results_dir = tmp_path / "results" / "session-1" + assert run_result == { + "claude_code_failed": False, + "claude_code_exit_code": 0, + "claude_code_error_type": None, + "claude_code_results_fpath": str(results_dir / "stream.jsonl"), + "claude_code_run_stdout": sandbox.run_result.stdout, + "claude_code_run_stderr": "", + "claude_code_export_found": True, + "claude_code_finished": True, + "claude_code_result_success": True, + "claude_code_result_error": False, + "claude_code_result_missing": False, + "claude_code_context_overflow": False, + "claude_code_duration_s": 1.234, + "claude_code_main_turns": 3, + "claude_code_subagent_turns": 1, + "claude_code_subagent_calls": 1, + "claude_code_tool_calls": 2, + "claude_code_tool_errors": 0, + "claude_code_unknown_tool_calls": 0, + } + assert (results_dir / "stream.jsonl").read_text() == _stream_text([*STREAM_EVENTS, RESULT_EVENT]) + receipt = json.loads((results_dir / "generation.json").read_text()) + assert receipt["execution"]["claude_code_finished"] is True + assert [item["type"] for item in receipt["response"]["output"]] == [item.type for item in EXPECTED_OUTPUT] + [root] = [record for record in observations.records if isinstance(record, AgentInvocation)] + assert (root.invocation_id, root.status, root.duration_ms) == ("s1", "completed", 1234) + [sandbox_record] = [record for record in observations.records if isinstance(record, SandboxObservation)] + assert (sandbox_record.role, sandbox_record.outcome, sandbox_record.exit_code) == ("agent", "completed", 0) + + async def test_missing_stream_is_a_failed_run(self, tmp_path: Path) -> None: + sandbox = _FakeSandbox(None, None, run_result=_exec_result(return_code=1, finished=False), tar_return_code=2) + agent = self._agent_with(sandbox) + + response = await agent.responses(_request(), _params()) + + assert (response.output, response.usage) == ([], None) + run_result = agent._sandbox_id_to_run_result["session-1"] + observations = run_result.pop("_ng_agent_observations") + assert run_result["claude_code_failed"] is True + assert run_result["claude_code_exit_code"] == 1 + assert run_result["claude_code_export_found"] is False + assert run_result["claude_code_results_fpath"] == "" + assert run_result["claude_code_result_missing"] is True + assert run_result["claude_code_main_turns"] == 0 + assert {gap.code for gap in observations.gaps} >= {"observation_capture_failed"} + [root] = [record for record in observations.records if isinstance(record, AgentInvocation)] + assert root.invocation_id == "7-2" + assert not (tmp_path / "results" / "session-1" / "stream.jsonl").exists() + assert ( + json.loads((tmp_path / "results" / "session-1" / "generation.json").read_text())["response"]["output"] + == [] + ) + + async def test_deadline_kill_leaves_no_result(self) -> None: + # `timeout` ends the CLI with 124 before it writes its result event; the patch so far is still graded. + sandbox = _FakeSandbox( + _stream_text(STREAM_EVENTS), None, run_result=_exec_result(return_code=124, finished=False) + ) + agent = self._agent_with(sandbox) + + response = await agent.responses(_request(observe=False), _params()) + + assert len(response.output) == len(EXPECTED_OUTPUT) + run_result = agent._sandbox_id_to_run_result["session-1"] + assert "_ng_agent_observations" not in run_result + assert (run_result["claude_code_failed"], run_result["claude_code_exit_code"]) == (True, 124) + assert (run_result["claude_code_result_missing"], run_result["claude_code_result_success"]) == (True, False) + assert run_result["claude_code_error_type"] is None + assert run_result["claude_code_duration_s"] is None + sandbox.exec.assert_awaited_once() + + async def test_context_overflow_result(self) -> None: + sandbox = _FakeSandbox( + _stream_text([*STREAM_EVENTS, OVERFLOW_EVENT]), + None, + run_result=_exec_result(return_code=1, finished=False), + ) + agent = self._agent_with(sandbox) + + await agent.responses(_request(observe=False), _params()) + + run_result = agent._sandbox_id_to_run_result["session-1"] + assert (run_result["claude_code_context_overflow"], run_result["claude_code_result_error"]) == (True, True) + assert (run_result["claude_code_failed"], run_result["claude_code_error_type"]) == (True, "context_overflow") + assert run_result["claude_code_duration_s"] == 99.0 + + async def test_exec_error_is_reported(self) -> None: + sandbox = _FakeSandbox(None, None) + sandbox.exec = AsyncMock(side_effect=TimeoutError("sandbox gone")) + agent = self._agent_with(sandbox) + + response = await agent.responses(_request(observe=False), _params()) + + assert response.output == [] + run_result = agent._sandbox_id_to_run_result["session-1"] + assert (run_result["claude_code_failed"], run_result["claude_code_error_type"]) == (True, "TimeoutError") + assert run_result["claude_code_exit_code"] is None + + async def test_requires_a_seeded_sandbox(self) -> None: + with raises(ValueError, match="Use /run"): + await _agent().responses(_request(cookies={}), _params()) + + +class _Response: + ok = True + + def __init__(self, payload: Dict[str, Any]) -> None: + self.payload = payload + self.cookies: Dict[str, str] = {} + + async def json(self) -> Dict[str, Any]: + return self.payload + + async def read(self) -> bytes: + return json.dumps(self.payload).encode() + + +class _RunRequest: + def __init__(self, body: Dict[str, Any]) -> None: + self._cookies: Dict[str, str] = {} + self._body = body + self.session = {SESSION_ID_KEY: "session-1"} + self.state = SimpleNamespace() + + @property + def cookies(self) -> Dict[str, str]: + return self._cookies + + async def json(self) -> Dict[str, Any]: + return self._body + + +def _server(agent: ClaudeCodeSandboxedAgent, *, reward_for_output: Any) -> AsyncMock: + async def post(server_name: str, url_path: str, json: Any = None, cookies: Any = None) -> _Response: + assert server_name == "task_server" + if url_path == "/seed_session": + return _Response({"sandbox_handle": "task-sandbox", "workdir": "/repo"}) + assert url_path == "/verify" + return _Response(json | {"reward": reward_for_output(json["response"]["output"])}) + + agent.server_client.post = AsyncMock(side_effect=post) + return agent.server_client.post + + +class TestRun: + @fixture(autouse=True) + def _results_in_tmp(self, tmp_path: Path, monkeypatch: MonkeyPatch, model_server_url: None) -> None: + monkeypatch.setattr(app_module, "__file__", str(tmp_path / "app.py")) + + async def test_run_end_to_end(self) -> None: + agent = _agent(CAPTURE_CONFIG, auto_compact=False, remote_claude_code_binary_path="/mnt/s3/claude") + sandbox = _FakeSandbox(_stream_text([*STREAM_EVENTS, RESULT_EVENT]), _transcripts_tar()) + agent._start_sandbox = AsyncMock(return_value=sandbox) + post = _server(agent, reward_for_output=lambda output: 1.0 if output else 0.0) + + result = await agent.run(_RunRequest(BODY), ClaudeCodeSandboxedAgentRunRequest.model_validate(BODY)) + + agent._start_sandbox.assert_awaited_once_with(sandbox_id="task-sandbox", workdir="/repo") + assert [call.kwargs["url_path"] for call in post.await_args_list] == ["/seed_session", "/verify"] + assert post.await_args_list[1].kwargs["cookies"]["sandbox_id"] == "session-1" + dumped = result.model_dump(mode="json") + assert dumped["reward"] == 1.0 + assert dumped["claude_code_finished"] is True and dumped["claude_code_failed"] is False + assert dumped["claude_code_result_success"] is True + assert dumped["claude_code_export_found"] is True + assert dumped["claude_code_results_fpath"].endswith("results/session-1/stream.jsonl") + assert [item["type"] for item in dumped["response"]["output"]][-1] == "message" + assert dumped["ng_agent_observations"]["source"] == "claude_code" + assert dumped.get("failure_kind") is None + sandbox.stop.assert_awaited_once() + assert agent._sandbox_id_to_sandbox == {} and agent._sandbox_id_to_run_result == {} + + async def test_execution_failure_reward_zero_grades_an_empty_response(self) -> None: + agent = _agent( + CAPTURE_CONFIG, remote_claude_code_binary_path="/mnt/s3/claude", execution_failure_reward_zero=True + ) + sandbox = _FakeSandbox( + _stream_text(STREAM_EVENTS), None, run_result=_exec_result(return_code=124, finished=False) + ) + agent._start_sandbox = AsyncMock(return_value=sandbox) + post = _server(agent, reward_for_output=lambda output: 1.0 if output else 0.0) + + result = await agent.run(_RunRequest(BODY), ClaudeCodeSandboxedAgentRunRequest.model_validate(BODY)) + + assert post.await_args_list[1].kwargs["json"]["response"]["output"] == [] + dumped = result.model_dump(mode="json") + assert (dumped["reward"], dumped["failure_kind"]) == (0.0, AGENT_RUN_ERROR) + assert dumped["claude_code_failed"] is True and dumped["claude_code_exit_code"] == 124 + # The response keeps what the model generated, for inspection. + assert len(dumped["response"]["output"]) == len(EXPECTED_OUTPUT) + sandbox.stop.assert_awaited_once() + + async def test_sandbox_is_stopped_when_the_run_fails(self) -> None: + agent = _agent(CAPTURE_CONFIG, remote_claude_code_binary_path="/mnt/s3/claude") + sandbox = _FakeSandbox(None, None) + sandbox.exec = AsyncMock(side_effect=RuntimeError("exec failed")) + sandbox.download = AsyncMock(side_effect=RuntimeError("download failed")) + agent._start_sandbox = AsyncMock(return_value=sandbox) + + async def post(server_name: str, url_path: str, json: Any = None, cookies: Any = None) -> _Response: + if url_path == "/seed_session": + return _Response({"sandbox_handle": "task-sandbox"}) + raise RuntimeError("verifier down") + + agent.server_client.post = AsyncMock(side_effect=post) + + with raises(RuntimeError, match="verifier down"): + await agent.run(_RunRequest(BODY), ClaudeCodeSandboxedAgentRunRequest.model_validate(BODY)) + + sandbox.stop.assert_awaited_once() + assert agent._sandbox_id_to_sandbox == {} and agent._sandbox_id_to_run_result == {} + + +def test_agent_config_file_validates() -> None: + config = yaml.safe_load((AGENT_DIR / "configs" / "claude_code_sandboxed_agent.yaml").read_text()) + block = config["claude_code_sandboxed_agent"]["responses_api_agents"]["claude_code_sandboxed_agent"] + block |= { + "host": "0.0.0.0", + "port": 1, + "name": "x", + "resources_server": {"type": "resources_servers", "name": "r"}, + } + + parsed = ClaudeCodeSandboxedAgentConfig.model_validate(block) + + assert parsed.claude_code_version == VERSION + assert parsed.token_id_capture is True + assert parsed.stream_first_byte_timeout_ms == 1800000 + assert parsed.auto_compact is True # training recipes turn it off + assert (parsed.bare, parsed.auto_memory, parsed.setting_sources) == (False, False, "user") + assert (parsed.execution_failure_reward_zero, parsed.artifacts_dir) == (False, None) + + +@mark.parametrize("server", ["swe_rebench", "scale_swe", "swemer_v1", "swemer_v2", "swe_next"]) +def test_swe_wiring_configs_mirror_opencode(server: str) -> None: + configs = PARENT_DIR / "resources_servers" / server / "configs" + claude_code = yaml.safe_load((configs / f"{server}_claude_code.yaml").read_text()) + opencode = yaml.safe_load((configs / f"{server}_opencode.yaml").read_text()) + + assert claude_code["config_paths"] == [ + "responses_api_agents/claude_code_sandboxed_agent/configs/claude_code_sandboxed_agent.yaml", + f"resources_servers/{server}/configs/{server}.yaml", + ] + agent_name = f"{server}_claude_code_sandboxed_agent" + agent = claude_code[agent_name] + assert agent["_inherit_from"] == "claude_code_sandboxed_agent" + block = agent["responses_api_agents"]["claude_code_sandboxed_agent"] + opencode_block = opencode[f"{server}_opencode_sandboxed_agent"]["responses_api_agents"]["opencode_sandboxed_agent"] + assert block == opencode_block + # Rollout collection dispatches through an environment server that names the agent; a run that + # loads both wiring configs must not have the two harnesses' servers collide. + environment = claude_code[f"{server}_claude_code_environment_server"]["environment_servers"]["legacy_agent"] + assert environment["agent_server"] == {"type": "responses_api_agents", "name": agent_name} + assert set(claude_code) & set(opencode) == {"config_paths"} + # Rollout collection refuses rows whose agent type the resources server does not accept. + resources = yaml.safe_load((configs / f"{server}.yaml").read_text()) + accepted = resources[f"{server}_resources_server"]["resources_servers"][server]["allowed_agents"] + assert {"opencode_sandboxed_agent", "claude_code_sandboxed_agent"} <= set(accepted)