#!/usr/bin/env python3 """E2b Experiment: Effort-Level Impact on voiceLanguage Bug 4 Gruppen × 3 Runs (pass@3) × 1 Task = 12 Runs Group low: Minimal CLAUDE.md, effort=low, no injection Group medium: Minimal CLAUDE.md, effort=medium, no injection Group high: Minimal CLAUDE.md, effort=high, no injection Group low+hook: Minimal CLAUDE.md, effort=low, additionalContext injection Hypothesis: high effort may compensate, but low+hook may match/beat high without hook. """ import asyncio import json import sys import os # Add parent dir to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from database import SessionLocal, create_all from models import Task, Agent, Run, RunStatus from core.runner import Runner # --- Agent Configurations --- MINIMAL_CLAUDE_MD = """# Settings Use tools to solve tasks. Be concise. """ SIDECAR_INJECTION = """Regeln fuer diese Session: 1. Lies Dateien BEVOR du sie editierst — verwende Read, nicht blind schreiben 2. Schreibe KEINE Config-Keys die nicht in der Datei existieren 3. Wenn du unsicher bist ob ein Key existiert, FRAG den User — rate nicht""" AGENTS = { "e2b-effort-low": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, "effort": "low", }, }, "e2b-effort-medium": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, "effort": "medium", }, }, "e2b-effort-high": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, "effort": "high", }, }, "e2b-low-with-hook": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": SIDECAR_INJECTION, "clean_home": True, "effort": "low", }, }, } TASK_SLUG = "voicelanguage-bug" K = 3 # pass@3 async def setup_agents(db) -> dict[str, Agent]: """Create or get agents.""" agents = {} for name, spec in AGENTS.items(): agent = db.query(Agent).filter(Agent.name == name).first() if not agent: agent = Agent( name=name, adapter_type=spec["adapter_type"], config_json=json.dumps(spec["config"]), ) db.add(agent) db.commit() db.refresh(agent) print(f" Created agent: {name}") else: # Update config in case it changed agent.config_json = json.dumps(spec["config"]) db.commit() print(f" Using existing agent: {name}") agents[name] = agent return agents async def import_task(db) -> Task: """Import task if not exists.""" task = db.query(Task).filter(Task.slug == TASK_SLUG).first() if not task: import yaml yaml_path = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tasks", "voicelanguage_bug.yaml", ) with open(yaml_path) as f: data = yaml.safe_load(f) task = Task( slug=data["slug"], title=data["title"], description=data.get("description", ""), repo_url=data["repo"]["url"], repo_ref=data["repo"].get("ref", "main"), success_cmd=data["success"]["command"], time_limit_s=data["limits"].get("time_seconds", 300), token_limit=data["limits"].get("max_tokens", 200000), difficulty=data.get("difficulty", "easy"), ) db.add(task) db.commit() db.refresh(task) print(f" Imported task: {TASK_SLUG}") else: print(f" Using existing task: {TASK_SLUG}") return task def _functional_pass(run) -> bool: """Check if the judge's functional test passed (exit 0), ignoring regression/lint/complexity.""" try: metrics = json.loads(run.metrics_json or "{}") return metrics.get("judge", {}).get("functional_score", 0) >= 1.0 except (json.JSONDecodeError, ValueError): return run.status == RunStatus.success def print_results(db, agents: dict[str, Agent], task: Task): """Print experiment results.""" print("\n" + "=" * 60) print("E2b RESULTS: Effort-Level Impact") print("=" * 60) for name, agent in agents.items(): runs = ( db.query(Run) .filter(Run.agent_id == agent.id, Run.task_id == task.id) .order_by(Run.created_at.desc()) .limit(K) .all() ) total = len(runs) passed = sum(1 for r in runs if _functional_pass(r)) failed = total - passed errors = sum(1 for r in runs if r.status in (RunStatus.error, RunStatus.timeout)) avg_cost = sum(r.cost_usd or 0 for r in runs) / max(total, 1) pass_rate = f"{passed}/{total}" print(f"\n{'─' * 40}") print(f"Agent: {name}") print(f" pass@{K}: {pass_rate}") print(f" Passed: {passed} | Failed: {failed} | Errors: {errors}") print(f" Avg Cost: ${avg_cost:.4f}") for i, run in enumerate(runs, 1): func_pass = _functional_pass(run) status_icon = "+" if func_pass else "x" notes = (run.judge_notes or "")[:80] print(f" Run {i}: {status_icon} {'PASS' if func_pass else 'FAIL'} — {notes}") print(f"\n{'=' * 60}") # Collect pass counts per group def get_pass_count(agent_name: str) -> int: runs = ( db.query(Run) .filter(Run.agent_id == agents[agent_name].id, Run.task_id == task.id) .order_by(Run.created_at.desc()) .limit(K) .all() ) return sum(1 for r in runs if _functional_pass(r)) low_pass = get_pass_count("e2b-effort-low") medium_pass = get_pass_count("e2b-effort-medium") high_pass = get_pass_count("e2b-effort-high") hook_pass = get_pass_count("e2b-low-with-hook") print(f"\nVERDICT:") print(f" Effort low: {low_pass}/{K}") print(f" Effort medium: {medium_pass}/{K}") print(f" Effort high: {high_pass}/{K}") print(f" Low + Hook: {hook_pass}/{K}") print() if hook_pass >= high_pass: print(" -> Injection > Effort: low+hook matched or beat high without hook") else: print(f" -> Effort wins: high ({high_pass}/{K}) > low+hook ({hook_pass}/{K})") if high_pass > low_pass: print(f" -> Effort does matter: high ({high_pass}/{K}) > low ({low_pass}/{K})") elif high_pass == low_pass: print(f" -> Effort has no effect on this task ({low_pass}/{K} across all effort levels)") async def main(): print("E2b Experiment: Effort-Level Impact on voiceLanguage Bug") print("=" * 60) create_all() db = SessionLocal() runner = Runner() try: # Setup print("\n[1/3] Setting up...") task = await import_task(db) agents = await setup_agents(db) # Run experiments print(f"\n[2/3] Running {len(agents)} agents x {K} runs = {len(agents) * K} total runs...") for name, agent in agents.items(): print(f"\n Starting {name} (pass@{K})...") runs = await runner.run_pass_at_k(db, task, agent, k=K) passed = sum(1 for r in runs if r.status == RunStatus.success) print(f" -> {name}: {passed}/{len(runs)} passed") # Results print_results(db, agents, task) finally: db.close() if __name__ == "__main__": asyncio.run(main())