| |
| """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 |
|
|
| |
| 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 |
|
|
| |
|
|
| 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 |
|
|
|
|
| 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: |
| |
| 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}") |
|
|
| |
| 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: |
| |
| print("\n[1/3] Setting up...") |
| task = await import_task(db) |
| agents = await setup_agents(db) |
|
|
| |
| 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") |
|
|
| |
| print_results(db, agents, task) |
|
|
| finally: |
| db.close() |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
|
|
