import json import os import shlex import shutil import subprocess import tempfile import time from config import CLAUDE_CLI_PATH, get_pricing from models import RunStatus from adapters.base import AgentAdapter, RunContext, RunResult class ClaudeCodeAdapter(AgentAdapter): def name(self) -> str: return "Claude Code" def is_available(self) -> bool: """Check if the claude CLI is on PATH or at the configured path.""" return shutil.which(CLAUDE_CLI_PATH) is not None async def run(self, ctx: RunContext, sandbox) -> RunResult: """Run Claude Code on the host against a local clone of the task repo. Claude CLI is not available inside Docker containers, so we clone the repo to a temp directory on the host and run the CLI there. The diff is captured and returned in RunResult.diff_text so the runner can create the judge sandbox from it. """ start = time.time() tmpdir = tempfile.mkdtemp(prefix="arena-claude-") clean_home = None try: # Clone the task repo to a host temp directory. repo_url = getattr(ctx, "repo_url", None) repo_ref = getattr(ctx, "repo_ref", "main") or "main" if not repo_url: # Fallback: nothing to clone — just run in work_dir as-is. work_dir = ctx.work_dir else: clone_result = subprocess.run( ["git", "clone", "--depth=1", "-b", repo_ref, repo_url, tmpdir], capture_output=True, text=True, timeout=120, ) if clone_result.returncode != 0: return RunResult( status=RunStatus.error, duration_s=round(time.time() - start, 2), raw_output=clone_result.stderr[:5000], ) work_dir = tmpdir # Create git checkpoint so we can diff later. subprocess.run( ["git", "tag", "arena-checkpoint"], cwd=work_dir, capture_output=True, timeout=10, ) # Write CLAUDE.md if configured claude_md = self.config.get("claude_md") if claude_md: claude_md_path = os.path.join(work_dir, "CLAUDE.md") with open(claude_md_path, "w") as f: f.write(claude_md) # Build prompt based on mode prompt_mode = self.config.get("prompt_mode", "fix") if prompt_mode == "raw": prompt = ctx.task_prompt else: prompt = ( f"Fix the following issue in the codebase at {work_dir}:\n\n" f"{ctx.task_prompt}\n\n" f"After fixing, verify your changes pass: {ctx.success_cmd}" ) # Inject additional context if configured (simulates Sidecar-NG hook) additional_context = self.config.get("additional_context") if additional_context: prompt = f"[SIDECAR-NG] {additional_context}\n\n---\n\n{prompt}" # Use model from agent config if set. model = self.config.get("model", "") cmd = [ CLAUDE_CLI_PATH, "--print", "--verbose", "--dangerously-skip-permissions", "--output-format", "stream-json", ] if model: cmd.extend(["--model", model]) effort = self.config.get("effort") if effort: cmd.extend(["--effort", effort]) cmd.append(prompt) # Unset CLAUDECODE so the nested claude CLI doesn't refuse to run. env = os.environ.copy() env.pop("CLAUDECODE", None) # Isolate from host's ~/.claude/ config (hooks, MCP, global CLAUDE.md) # to prevent cross-contamination in A/B experiments. clean_home = None if self.config.get("clean_home"): clean_home = tempfile.mkdtemp(prefix="arena-home-") clean_claude_dir = os.path.join(clean_home, ".claude") os.makedirs(clean_claude_dir) # Copy only credentials (API key), not hooks/settings/CLAUDE.md real_creds = os.path.expanduser("~/.claude/.credentials.json") if os.path.exists(real_creds): shutil.copy2(real_creds, os.path.join(clean_claude_dir, ".credentials.json")) env["HOME"] = clean_home try: proc = subprocess.run( cmd, cwd=work_dir, capture_output=True, text=True, timeout=ctx.time_limit_s, env=env, ) returncode = proc.returncode stdout = proc.stdout stderr = proc.stderr except subprocess.TimeoutExpired: returncode = 124 stdout = "" stderr = f"Command timed out after {ctx.time_limit_s}s" duration = time.time() - start # Collect git diff from host clone, excluding __pycache__ and .pyc files. def _get_diff(rev): r = subprocess.run( ["git", "diff", rev, "--", ".", ":!**/__pycache__/", ":!**/*.pyc"], cwd=work_dir, capture_output=True, text=True, timeout=30, ) return r.stdout if r.returncode == 0 else "" diff_text = _get_diff("arena-checkpoint") if not diff_text.strip(): diff_text = _get_diff("arena-checkpoint..HEAD") tokens_input = 0 tokens_output = 0 cache_create = 0 cache_read = 0 trajectory = [] tool_call_count = 0 model_id = model or "claude-sonnet-4-6" cost = 0.0 # stream-json: one JSON object per line (NDJSON). for raw_line in stdout.strip().splitlines(): raw_line = raw_line.strip() if not raw_line: continue try: entry = json.loads(raw_line) except (json.JSONDecodeError, ValueError): continue entry_type = entry.get("type", "") if entry_type == "assistant": # Extract tool_use blocks from content message = entry.get("message", {}) for block in message.get("content", []): if block.get("type") == "tool_use": tool_call_count += 1 tool_name = block.get("name", "unknown") input_data = block.get("input", {}) preview = "" if isinstance(input_data, dict): preview = ( input_data.get("file_path", "") or input_data.get("command", "") or input_data.get("pattern", "") or str(input_data)[:120] ) trajectory.append({ "tool_name": tool_name, "input_preview": str(preview)[:200], }) elif entry_type == "result": usage = entry.get("usage", {}) tokens_input = usage.get("input_tokens", 0) tokens_output = usage.get("output_tokens", 0) cache_create = usage.get("cache_creation_input_tokens", 0) cache_read = usage.get("cache_read_input_tokens", 0) cost = entry.get("total_cost_usd", 0.0) or entry.get("cost_usd", 0.0) or 0.0 model_usage = entry.get("modelUsage", {}) if model_usage: model_id = next(iter(model_usage)) mu = model_usage[model_id] # Prefer modelUsage totals over per-turn usage tokens_input = mu.get("inputTokens", tokens_input) tokens_output = mu.get("outputTokens", tokens_output) cache_create = mu.get("cacheCreationInputTokens", cache_create) cache_read = mu.get("cacheReadInputTokens", cache_read) # Fallback: try parsing as single JSON (old --output-format json) if not trajectory and not tokens_input: try: entry = json.loads(stdout.strip()) usage = entry.get("usage", {}) tokens_input = usage.get("input_tokens", 0) tokens_output = usage.get("output_tokens", 0) cache_create = usage.get("cache_creation_input_tokens", 0) cache_read = usage.get("cache_read_input_tokens", 0) cost = entry.get("total_cost_usd", 0.0) or 0.0 model_usage = entry.get("modelUsage", {}) if model_usage: model_id = next(iter(model_usage)) mu = model_usage[model_id] # Prefer modelUsage totals over per-turn usage tokens_input = mu.get("inputTokens", tokens_input) tokens_output = mu.get("outputTokens", tokens_output) cache_create = mu.get("cacheCreationInputTokens", cache_create) cache_read = mu.get("cacheReadInputTokens", cache_read) except (json.JSONDecodeError, ValueError): pass # Compute cost from tokens if not provided directly if cost == 0.0 and (tokens_input or tokens_output): rate_in, rate_w, rate_r, rate_out = get_pricing(model_id) cost = (tokens_input * rate_in + cache_create * rate_w + cache_read * rate_r + tokens_output * rate_out) / 1_000_000 if duration >= ctx.time_limit_s: status = RunStatus.timeout elif returncode == 0: status = RunStatus.success else: status = RunStatus.failure return RunResult( status=status, duration_s=round(duration, 2), tokens_input=tokens_input, tokens_output=tokens_output, cost_usd=round(cost, 6), trajectory=trajectory, tool_calls=tool_call_count, diff_text=diff_text, raw_output=stdout[:50_000], ) finally: shutil.rmtree(tmpdir, ignore_errors=True) if clean_home: shutil.rmtree(clean_home, ignore_errors=True)