Initial public upload from Projekte/AgentArena

3b0d6b3 verified about 1 month ago

11.1 kB

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)

	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)