[examples]feat: add Claude Code blackbox training recipe#78
[examples]feat: add Claude Code blackbox training recipe#78zhaizhiqiangA wants to merge 11 commits into
Conversation
blackbox cc anthropic adapter
fix(framework): tag trajectory materialization reason
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a self-contained Claude Code in-sandbox execution recipe, featuring a dedicated Dockerfile, build scripts, runner, parallel inference, and configurations. It also adds a standalone gateway debug launcher and extends the gateway with an Anthropic Messages adapter to support both OpenAI and Anthropic protocols. The review feedback highlights several critical improvements: fixing a missing urlparse import in anthropic.py, gracefully handling null values for tool_choice in both adapters, resolving potential issues with missing ports and lost query/fragment parameters in sandbox_client.py, wrapping synchronous I/O-bound calls in asyncio.to_thread to prevent blocking the event loop, and avoiding unbound variable crashes in build_tool.sh.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| import secrets | ||
| from collections.abc import AsyncIterator |
There was a problem hiding this comment.
The urlparse function is used on line 343 but is not imported in this file. This will cause a NameError at runtime. Please import urlparse from urllib.parse.
| import secrets | |
| from collections.abc import AsyncIterator | |
| import secrets | |
| from urllib.parse import urlparse | |
| from collections.abc import AsyncIterator |
| tool_choice_payload = payload.get("tool_choice", "auto") | ||
| if isinstance(tool_choice_payload, str): | ||
| tool_choice = tool_choice_payload.lower() | ||
| if tool_choice not in {"auto", "none"}: | ||
| raise MalformedRequestError( | ||
| f'tool_choice="{tool_choice_payload}" is not supported (only "auto" / "none" are supported)' | ||
| ) | ||
| elif isinstance(tool_choice_payload, dict): | ||
| raise MalformedRequestError( | ||
| 'tool_choice with a specific function is not supported (only "auto" / "none" are supported)' | ||
| ) | ||
| else: | ||
| raise MalformedRequestError("tool_choice must be a string or object") |
There was a problem hiding this comment.
If tool_choice is explicitly set to null in the JSON payload, tool_choice_payload will be None. This will trigger the else block and raise a MalformedRequestError. We should handle None gracefully by defaulting it to 'auto'.
| tool_choice_payload = payload.get("tool_choice", "auto") | |
| if isinstance(tool_choice_payload, str): | |
| tool_choice = tool_choice_payload.lower() | |
| if tool_choice not in {"auto", "none"}: | |
| raise MalformedRequestError( | |
| f'tool_choice="{tool_choice_payload}" is not supported (only "auto" / "none" are supported)' | |
| ) | |
| elif isinstance(tool_choice_payload, dict): | |
| raise MalformedRequestError( | |
| 'tool_choice with a specific function is not supported (only "auto" / "none" are supported)' | |
| ) | |
| else: | |
| raise MalformedRequestError("tool_choice must be a string or object") | |
| tool_choice_payload = payload.get("tool_choice", "auto") | |
| if tool_choice_payload is None: | |
| tool_choice = "auto" | |
| elif isinstance(tool_choice_payload, str): | |
| tool_choice = tool_choice_payload.lower() | |
| if tool_choice not in {"auto", "none"}: | |
| raise MalformedRequestError( | |
| f'tool_choice="{tool_choice_payload}" is not supported (only "auto" / "none" are supported)' | |
| ) | |
| elif isinstance(tool_choice_payload, dict): | |
| raise MalformedRequestError( | |
| 'tool_choice with a specific function is not supported (only "auto" / "none" are supported)' | |
| ) | |
| else: | |
| raise MalformedRequestError("tool_choice must be a string or object") |
| def _apply_tool_choice(payload: dict[str, Any], tools: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None: | ||
| tool_choice = payload.get("tool_choice", "auto") | ||
| if isinstance(tool_choice, str): | ||
| if tool_choice == "auto": | ||
| return tools | ||
| if tool_choice == "none": | ||
| return None | ||
| raise MalformedRequestError(f"Unsupported tool_choice: {tool_choice}") | ||
| if isinstance(tool_choice, dict): | ||
| choice_type = tool_choice.get("type") | ||
| if choice_type == "auto": | ||
| return tools | ||
| if choice_type == "none": | ||
| return None | ||
| raise MalformedRequestError(f"Unsupported tool_choice type: {choice_type}") | ||
| raise MalformedRequestError("tool_choice must be a string or object") |
There was a problem hiding this comment.
If tool_choice is explicitly set to null in the JSON payload, tool_choice will be None. This will trigger the else block and raise a MalformedRequestError. We should handle None gracefully by returning tools.
| def _apply_tool_choice(payload: dict[str, Any], tools: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None: | |
| tool_choice = payload.get("tool_choice", "auto") | |
| if isinstance(tool_choice, str): | |
| if tool_choice == "auto": | |
| return tools | |
| if tool_choice == "none": | |
| return None | |
| raise MalformedRequestError(f"Unsupported tool_choice: {tool_choice}") | |
| if isinstance(tool_choice, dict): | |
| choice_type = tool_choice.get("type") | |
| if choice_type == "auto": | |
| return tools | |
| if choice_type == "none": | |
| return None | |
| raise MalformedRequestError(f"Unsupported tool_choice type: {choice_type}") | |
| raise MalformedRequestError("tool_choice must be a string or object") | |
| def _apply_tool_choice(payload: dict[str, Any], tools: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None: | |
| tool_choice = payload.get("tool_choice", "auto") | |
| if tool_choice is None: | |
| return tools | |
| if isinstance(tool_choice, str): | |
| if tool_choice == "auto": | |
| return tools | |
| if tool_choice == "none": | |
| return None | |
| raise MalformedRequestError(f"Unsupported tool_choice: {tool_choice}") | |
| if isinstance(tool_choice, dict): | |
| choice_type = tool_choice.get("type") | |
| if choice_type == "auto": | |
| return tools | |
| if choice_type == "none": | |
| return None | |
| raise MalformedRequestError(f"Unsupported tool_choice type: {choice_type}") | |
| raise MalformedRequestError("tool_choice must be a string or object") |
| def extract_upstream(gateway_url: str) -> str: | ||
| """Extract host:port from a gateway URL for upstream tunnel config. | ||
|
|
||
| Example: "http://8.92.9.155:40169/sessions/abc/v1" -> "8.92.9.155:40169" | ||
| """ | ||
| parsed = urlparse(gateway_url) | ||
| return f"{parsed.hostname}:{parsed.port}" |
There was a problem hiding this comment.
If the gateway_url does not explicitly specify a port (e.g., standard port 80/443 or a domain name behind a reverse proxy), parsed.port will be None, resulting in an invalid upstream address like example.com:None. We should handle None ports gracefully by defaulting to standard ports based on the scheme.
def extract_upstream(gateway_url: str) -> str:
"""Extract host:port from a gateway URL for upstream tunnel config.
Example: "http://8.92.9.155:40169/sessions/abc/v1" -> "8.92.9.155:40169"
"""
parsed = urlparse(gateway_url)
if parsed.port is None:
port = 80 if parsed.scheme == "http" else 443
return f"{parsed.hostname}:{port}"
return f"{parsed.hostname}:{parsed.port}"| def rewrite_gateway_url( | ||
| gateway_url: str, | ||
| proxy_port: int = DEFAULT_PROXY_PORT, | ||
| *, | ||
| strip_v1: bool = False, | ||
| ) -> str: | ||
| """Rewrite gateway URL to use the sandbox-internal tunnel. | ||
|
|
||
| Replaces host:port with 127.0.0.1:<proxy_port>, keeps path intact. | ||
|
|
||
| Example: | ||
| "http://8.92.9.155:40169/sessions/abc/v1" | ||
| -> "http://127.0.0.1:8766/sessions/abc/v1" | ||
| """ | ||
| parsed = urlparse(gateway_url) | ||
| path = parsed.path.removesuffix("/v1") if strip_v1 else parsed.path | ||
| return f"http://127.0.0.1:{proxy_port}{path}" |
There was a problem hiding this comment.
If the gateway_url contains query parameters or fragments (e.g., for authentication or session metadata), they are completely lost during the rewrite. We should preserve the query and fragment parts of the URL.
| def rewrite_gateway_url( | |
| gateway_url: str, | |
| proxy_port: int = DEFAULT_PROXY_PORT, | |
| *, | |
| strip_v1: bool = False, | |
| ) -> str: | |
| """Rewrite gateway URL to use the sandbox-internal tunnel. | |
| Replaces host:port with 127.0.0.1:<proxy_port>, keeps path intact. | |
| Example: | |
| "http://8.92.9.155:40169/sessions/abc/v1" | |
| -> "http://127.0.0.1:8766/sessions/abc/v1" | |
| """ | |
| parsed = urlparse(gateway_url) | |
| path = parsed.path.removesuffix("/v1") if strip_v1 else parsed.path | |
| return f"http://127.0.0.1:{proxy_port}{path}" | |
| def rewrite_gateway_url( | |
| gateway_url: str, | |
| proxy_port: int = DEFAULT_PROXY_PORT, | |
| *, | |
| strip_v1: bool = False, | |
| ) -> str: | |
| """Rewrite gateway URL to use the sandbox-internal tunnel. | |
| Replaces host:port with 127.0.0.1:<proxy_port>, keeps path intact. | |
| Example: | |
| "http://8.92.9.155:40169/sessions/abc/v1" | |
| -> "http://127.0.0.1:8766/sessions/abc/v1" | |
| """ | |
| parsed = urlparse(gateway_url) | |
| path = parsed.path.removesuffix("/v1") if strip_v1 else parsed.path | |
| query = f"?{parsed.query}" if parsed.query else "" | |
| fragment = f"#{parsed.fragment}" if parsed.fragment else "" | |
| return f"http://127.0.0.1:{proxy_port}{path}{query}{fragment}" |
| async def cleanup(self) -> None: | ||
| """Kill the sandbox if still running.""" | ||
| if self._sandbox is not None: | ||
| sandbox_id = getattr(self._sandbox, "sandbox_id", "?") | ||
| try: | ||
| if self._sandbox.is_running(): | ||
| await asyncio.to_thread(self._sandbox.kill) | ||
| logger.info("sandbox %s killed", sandbox_id) | ||
| else: | ||
| logger.info("sandbox %s already stopped", sandbox_id) | ||
| except Exception as e: | ||
| logger.warning("Failed to kill sandbox %s: %s", sandbox_id, e) | ||
| self._sandbox = None |
There was a problem hiding this comment.
The self._sandbox.is_running() call is synchronous and likely performs network I/O to check the sandbox status. Calling it directly inside an async def blocks the asyncio event loop. We should wrap it in asyncio.to_thread to keep the event loop non-blocking.
| async def cleanup(self) -> None: | |
| """Kill the sandbox if still running.""" | |
| if self._sandbox is not None: | |
| sandbox_id = getattr(self._sandbox, "sandbox_id", "?") | |
| try: | |
| if self._sandbox.is_running(): | |
| await asyncio.to_thread(self._sandbox.kill) | |
| logger.info("sandbox %s killed", sandbox_id) | |
| else: | |
| logger.info("sandbox %s already stopped", sandbox_id) | |
| except Exception as e: | |
| logger.warning("Failed to kill sandbox %s: %s", sandbox_id, e) | |
| self._sandbox = None | |
| async def cleanup(self) -> None: | |
| """Kill the sandbox if still running.""" | |
| if self._sandbox is not None: | |
| sandbox_id = getattr(self._sandbox, "sandbox_id", "?") | |
| try: | |
| is_running = await asyncio.to_thread(self._sandbox.is_running) | |
| if is_running: | |
| await asyncio.to_thread(self._sandbox.kill) | |
| logger.info("sandbox %s killed", sandbox_id) | |
| else: | |
| logger.info("sandbox %s already stopped", sandbox_id) | |
| except Exception as e: | |
| logger.warning("Failed to kill sandbox %s: %s", sandbox_id, e) | |
| self._sandbox = None |
| async def _default_vision_info_extractor( | ||
| self, | ||
| messages: list[dict[str, Any]], | ||
| *, | ||
| image_patch_size: int, | ||
| **_extra: Any, | ||
| ) -> tuple[list[Any] | None, list[Any] | None]: | ||
| # Keep the dataset dependency lazy so custom extractors do not pay for | ||
| # RLHFDataset imports unless they actually use the default path. | ||
| from verl.utils.dataset.rl_dataset import RLHFDataset | ||
| # Lazy import so callers without multi-modal needs do not load | ||
| # qwen_vl_utils. ``_extra`` absorbs ``vision_info_extractor_kwargs`` that | ||
| # ``extract_multi_modal_data`` forwards for custom extractors; the | ||
| # default path needs nothing beyond ``messages`` and patch size. | ||
| from qwen_vl_utils import process_vision_info | ||
|
|
||
| return await RLHFDataset.process_vision_info( | ||
| return process_vision_info( | ||
| messages, | ||
| image_patch_size=image_patch_size, | ||
| config=self._vision_info_extractor_kwargs.get("config"), | ||
| return_video_metadata=True, | ||
| ) |
There was a problem hiding this comment.
process_vision_info is a synchronous function that performs network/disk I/O (e.g., downloading images from URLs). Calling it directly inside an async def blocks the asyncio event loop. We should wrap it in asyncio.to_thread to keep the event loop non-blocking.
async def _default_vision_info_extractor(
self,
messages: list[dict[str, Any]],
*,
image_patch_size: int,
**_extra: Any,
) -> tuple[list[Any] | None, list[Any] | None]:
# Lazy import so callers without multi-modal needs do not load
# qwen_vl_utils. ``_extra`` absorbs ``vision_info_extractor_kwargs`` that
# ``extract_multi_modal_data`` forwards for custom extractors; the
# default path needs nothing beyond ``messages`` and patch size.
import asyncio
from qwen_vl_utils import process_vision_info
return await asyncio.to_thread(
process_vision_info,
messages,
image_patch_size=image_patch_size,
return_video_metadata=True,
)| while [[ $# -gt 0 ]]; do | ||
| case "$1" in | ||
| --registry) REGISTRY="$2"; shift 2 ;; | ||
| --npm-registry) NPM_REGISTRY="$2"; shift 2 ;; | ||
| --tool-version) TOOL_VERSION="$2"; shift 2 ;; | ||
| *) echo "Unknown arg: $1"; exit 1 ;; | ||
| esac | ||
| done |
There was a problem hiding this comment.
Since set -u is active, referencing $2 when parsing the last argument without a value (e.g., --registry with no value) will raise an unbound variable error and crash the script. We should use ${2:-} to safely default to an empty string.
| while [[ $# -gt 0 ]]; do | |
| case "$1" in | |
| --registry) REGISTRY="$2"; shift 2 ;; | |
| --npm-registry) NPM_REGISTRY="$2"; shift 2 ;; | |
| --tool-version) TOOL_VERSION="$2"; shift 2 ;; | |
| *) echo "Unknown arg: $1"; exit 1 ;; | |
| esac | |
| done | |
| while [[ $# -gt 0 ]]; do | |
| case "$1" in | |
| --registry) REGISTRY="${2:-}"; shift 2 ;; | |
| --npm-registry) NPM_REGISTRY="${2:-}"; shift 2 ;; | |
| --tool-version) TOOL_VERSION="${2:-}"; shift 2 ;; | |
| *) echo "Unknown arg: $1"; exit 1 ;; | |
| esac | |
| done |
2 participants
What does this PR do?
This PR adds a Claude Code blackbox SWE-bench recipe under
examples/blackbox_recipes/claude_code/.Claude Code runs inside the remote SWE-bench sandbox through a sidecar tool image, talks to the rollout model through Uni-Agent's gateway Anthropic Messages endpoint, and is evaluated in the same sandbox through the existing reward-spec path. The PR also adds standalone inference/debug utilities and gateway adapter coverage needed to support Claude Code style Anthropic requests.
Related work:
Checklist Before Starting
[{modules}] {type}: {description}(checked by CI)[examples, tools, train] feat: add Claude Code blackbox training recipeTest
Result:
84 passed, 4 warnings.generate_sequences summary: num_input_prompts=1 num_success_sessions=1 num_failed_sessions=0 num_success_outputs=1INFERENCE_RESULT_JSON={"resolved": 0, "total": 1, "mean_score": 0.0, "per_sample_scores": [0.0]}23prepare/generation rounds with stable Claude Code tool schema.Full RL training was not run in CI because it requires multi-GPU Ray resources, AKernel credentials, remote sandbox images, and a pushed Claude Code tool image.
API and Usage Example
No breaking public API changes. This adds a new example recipe and gateway/debug entrypoints.
Design & Code Changes
claude_code_runner.py, which creates a remote sandbox, mounts the Claude Code sidecar image at/opt/claude-code, rewrites the gateway URL to the sandbox tunnel, runsclaude -p, evaluates reward in the same sandbox, and postsreward_info.run_train.sh, and standaloneparallel_infer.py/run_infer.sh.sandbox_client.pyfor OpenYuanRong sandbox creation, sidecar mounting, command execution, cleanup, and gateway tunnel URL handling.Checklist Before Submitting
pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always