AgentArena

1. Pipeline-Flow

Jeder Benchmark-Run durchlaeuft diese 8 Phasen sequenziell. Der Agent-Container und der Judge-Container existieren nie gleichzeitig — der Agent-Container wird zerstoert, bevor der Judge-Container erzeugt wird.

Was passiert in jeder Phase?

Aus dem Code: Runner-Kernlogik

# core/runner.py (vereinfacht)
async def run_single(self, db, task, agent) -> Run:
    # 1. Sandbox erstellen (git clone + checkpoint)
    sandbox = await asyncio.to_thread(Sandbox.create, task)

    # 2. Adapter holen und ausfuehren (HOST subprocess, nicht im Container)
    adapter = get_adapter(agent.adapter_type, agent_config)
    adapter_result = await adapter.run(ctx, sandbox)

    # 3. Diff aus Host-Tempdir oder Container holen
    diff_text = adapter_result.diff_text or await asyncio.to_thread(sandbox.get_diff)

    # 4. AGENT CONTAINER ZERSTOEREN (Security-Grenze!)
    await asyncio.to_thread(sandbox.destroy)
    sandbox = None

    # 5. FRISCHEN Judge-Container erstellen + Diff einspielen
    judge_sandbox = await asyncio.to_thread(
        Sandbox.create_judge_sandbox, task, diff_text
    )

    # 6. Judge laeuft NUR im frischen Container
    judge_result = await asyncio.to_thread(
        self.judge.judge_run, judge_sandbox, task.success_cmd
    )

2. Sicherheitsmodell: 2-Container-Isolation

Das kritischste Design-Entscheidung: Agent und Judge laufen in separaten, kurzlebigen Docker-Containern. Der Agent hat keine Moeglichkeit, die Testumgebung zu manipulieren, weil der Judge-Container erst nach Zerstoerung des Agent-Containers erstellt wird.

Sandbox.create_judge_sandbox() im Detail

# core/sandbox.py
@classmethod
def create_judge_sandbox(cls, task, diff_text: str) -> "Sandbox":
    # Frischer Container — kein Bezug zum Agent-Container
    container_id = docker_run(image="python:3.12-slim")
    sandbox = cls(container_id)

    # Frischer Clone direkt aus Remote
    sandbox.exec(f"git clone --depth=1 -b {repo_ref} {repo_url} .")

    # Diff als Datei in Container kopieren
    subprocess.run(["docker", "cp", tmp_path, f"{container_id}:/tmp/agent.patch"])

    # Patch anwenden (git apply, Fallback: patch -p1)
    rc, out, err = sandbox.exec(
        "git apply --whitespace=fix /tmp/agent.patch || patch -p1 < /tmp/agent.patch"
    )
    sandbox.patch_applied = rc == 0
    return sandbox

3. Adapter-Architektur

Alle Adapter implementieren das AgentAdapter ABC und laufen auf dem HOST-System, nicht im Container. Der Grund: Claude Code und Aider koennen nicht sinnvoll innerhalb von Docker-Containern ausgefuehrt werden (OAuth-Zustand, API-Keys, Dateisystem-Zugriff).

Gemeinsames Pattern: Host-Clone-Strategy

Alle Adapter klonen das Repo in ein Host-Tempverzeichnis, setzen einen Git-Checkpoint-Tag, fuehren dann den CLI-Prozess aus, und sammeln den Diff vor dem Cleanup.

# Muster fuer alle Adapter (adapters/claude_code.py, adapters/aider.py)
tmpdir = tempfile.mkdtemp(prefix="arena-claude-")  # oder arena-aider-

# 1. Repo klonen
subprocess.run(["git", "clone", "--depth=1", "-b", repo_ref, repo_url, tmpdir])

# 2. Checkpoint vor Agent-Arbeit
subprocess.run(["git", "tag", "arena-checkpoint"], cwd=tmpdir)

# 3. CLI ausfuehren (timeout = task.time_limit_s)
proc = subprocess.run(cmd, cwd=tmpdir, capture_output=True, timeout=ctx.time_limit_s)

# 4. Diff sammeln (excludiert __pycache__ und .pyc)
diff = subprocess.run(
    ["git", "diff", "arena-checkpoint", "--", ".",
     ":!**/__pycache__/", ":!**/*.pyc"],
    cwd=tmpdir
)

# 5. Cleanup (always, via try/finally)
shutil.rmtree(tmpdir, ignore_errors=True)

Preistabellen

4. Judge-System: 4 Dimensionen

Der Judge bewertet den Code in vier Dimensionen. Funktionalitaet wirkt als Gatekeeper: besteht die Aufgabe nicht, ist der Overall-Score 0 unabhaengig von allen anderen Dimensionen.

if rc == 0:
    result.functional_score = 1.0
else:
    result.functional_score = 0.0

# Crash detection
is_crash = rc != 0 and passed == 0 \
           and crash_match and not has_real_tests
if total > 0 and not is_crash:
    result.regression_score = passed / total

# git stash trick fuer Before-State
sandbox.exec("git stash")  # temporaer zuruecksetzen
before = count_ruff_warnings(sandbox, files)
sandbox.exec("git stash pop")  # Agent-Aenderungen wiederherstellen
after = count_ruff_warnings(sandbox, files)
# score: max(0, 1.0 - delta * 0.1)

# Fallback-Proxy (simple_complexity)
max_indent = max(len(l) - len(l.lstrip()) for l in lines)
func_count = sum(1 for l in lines
             if l.strip().startswith('def ')
             or l.strip().startswith('class '))

pytest Output Parser

Der Parser erkennt verschiedene pytest-Ausgabeformate robuster als ein einfaches Regex-Match:

# core/metrics.py — parse_pytest_output()
# Erkennt: "5 passed", "3 failed, 2 passed", "1 error, 2 failed, 3 passed"
passed_match = re.search(r"(\d+)\s+passed", output)
failed_match = re.search(r"(\d+)\s+failed", output)
error_match  = re.search(r"(\d+)\s+error",  output)
# Fallback: "X/Y tests passed" (unittest-Stil)
fraction = re.search(r"(\d+)/(\d+)\s+(?:tests?\s+)?passed", output)

5. Metriken-System

# Directness Score — kombiniert 5 Sub-Metriken
score = (
    0.35 * trajectory_efficiency   # Edit/Write-Ratio
  + 0.25 * early_edit_score         # wie frueh erster Edit
  - 0.20 * repeat_penalty           # gleiche Datei wiederholt gelesen
  - 0.10 * loop_penalty             # zyklische Tool-Patterns
  - 0.10 * stall_rate               # lange Read-Phasen ohne Edit
)

# --output-format json liefert:
entry["usage"]["input_tokens"]
entry["usage"]["output_tokens"]
entry["usage"]["cache_creation_input_tokens"]
entry["usage"]["cache_read_input_tokens"]
entry["total_cost_usd"]  # direkt verfuegbar!

# Regex aus Aider-Output:
_TOKEN_RE = re.compile(
  r"Tokens:\s*(?P<sent>\d+)\s+sent,"
  r"\s*(?P<recv>\d+)\s+received"
)
_COST_RE = re.compile(
  r"Cost:\s*\$(?P<cost>[0-9]+\.[0-9]+)"
)

# Kosten-Kalkulation wenn kein direkter Wert (Fallback)
cost = (tokens_input * rate_in + tokens_output * rate_out) / 1_000_000

# Effizienz-Score fuer Composite (normalisiert auf 0-1)
efficiency = 1.0 / (1.0 + cost_usd / 0.50)  # $0 = 1.0, $0.50 = 0.5

# core/metrics.py
def compute_pass_at_k(n: int, c: int, k: int) -> float:
    if n < k: return 0.0
    if c == 0: return 0.0
    if n - c < k: return 1.0
    return 1.0 - comb(n - c, k) / comb(n, k)

Judge Score Visualisierung (Beispiel: Run #10, Aider + MiniMax)

6. Live-Updates via Server-Sent Events

Die Webapp streamt Fortschritts-Updates in Echtzeit ohne Polling. Kein WebSocket-Overhead, kein periodisches Refetching — SSE ist ideal fuer unidirektionale Streams.

<div
  hx-ext="sse"
  sse-connect="/runs/{{id}}/stream"
  sse-swap="message"
  hx-target="#status-box">
  <div id="status-box">Warte...</div>
</div>

# main.py — SSE Generator
async def run_status_stream(run_id: int):
    while True:
        progress = runner.get_progress(run_id)
        if progress:
            status = progress.get("status", "unknown")
            elapsed = time.time() - progress.get("started_at", time.time())
            yield f"data: {json.dumps({'status': status, 'elapsed': elapsed})}\n\n"
        else:
            yield "data: {\"status\": \"done\"}\n\n"
            break
        await asyncio.sleep(0.5)

return StreamingResponse(run_status_stream(run_id), media_type="text/event-stream")

7. Dateistruktur

Datenmodelle

8. Bisherige Ergebnisse

Was kommt als naechstes?