Blame for experiments/e2b_effort_levels.py

#!/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())