#!/usr/bin/env python3 """E7 Experiment: Injection-Strategie (Granularitat) 5 Gruppen x 3 Runs (pass@3) x 1 Task = 15 Runs e7-baseline: Keine Injection (Kontrollgruppe) e7-broad-rule: EINE breite Regel (generisch) e7-specific-3: DREI spezifische Regeln (wie E2 hook-only) e7-minimal-rule: NUR die kritische Einzelregel e7-english-3: Gleiche 3 Regeln auf Englisch Hypothese: 1. Reicht eine einzelne Regel? (minimal vs specific-3rules) 2. Breit vs spezifisch — welche Formulierung wirkt besser? 3. Spracheinfluss: deutsche vs englische Injection """ 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. """ # Keine Injection — Kontrollgruppe BASELINE_INJECTION = None # EINE breite Regel BROAD_RULE_INJECTION = ( "Verandere niemals Konfigurationsdateien ohne vorher den aktuellen Inhalt zu lesen " "und sicherzustellen dass deine Anderungen zu den existierenden Keys passen." ) # DREI spezifische Regeln (identisch zu E2 hook-only) SPECIFIC_3_INJECTION = ( "Regeln fuer diese Session:\n" "1. Lies Dateien BEVOR du sie editierst\n" "2. Schreibe KEINE Config-Keys die nicht in der Datei existieren\n" "3. Wenn du unsicher bist ob ein Key existiert, FRAG den User" ) # NUR die kritische Einzelregel MINIMAL_RULE_INJECTION = ( "Schreibe KEINE Config-Keys die nicht in der Datei existieren." ) # Gleiche 3 Regeln auf Englisch ENGLISH_3_INJECTION = ( "Rules for this session:\n" "1. Read files BEFORE editing them\n" "2. Do NOT write config keys that don't exist in the file\n" "3. If unsure whether a key exists, ASK the user" ) AGENTS = { "e7-baseline": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, }, }, "e7-broad-rule": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": BROAD_RULE_INJECTION, "clean_home": True, }, }, "e7-specific-3rules": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": SPECIFIC_3_INJECTION, "clean_home": True, }, }, "e7-minimal-rule": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": MINIMAL_RULE_INJECTION, "clean_home": True, }, }, "e7-english-rules": { "adapter_type": "claude_code", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": ENGLISH_3_INJECTION, "clean_home": True, }, }, } 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("E7 RESULTS: Injection-Strategie (Granularitat)") print("=" * 60) pass_counts = {} 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_counts[name] = passed 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}") # Verdicts baseline = pass_counts.get("e7-baseline", 0) broad = pass_counts.get("e7-broad-rule", 0) specific3 = pass_counts.get("e7-specific-3rules", 0) minimal = pass_counts.get("e7-minimal-rule", 0) english3 = pass_counts.get("e7-english-rules", 0) print(f"\nVERDICT:") print(f" Baseline: {baseline}/{K}") print(f" Broad rule: {broad}/{K}") print(f" Specific-3: {specific3}/{K}") print(f" Minimal: {minimal}/{K}") print(f" English-3: {english3}/{K}") print(f"\n [1] Minimal vs Specific-3 (reicht eine Regel?):") if minimal >= specific3 and minimal > baseline: print(f" -> JA: Einzelregel ausreichend (minimal={minimal}, specific={specific3})") elif specific3 > minimal: print(f" -> NEIN: Drei Regeln besser als eine (specific={specific3}, minimal={minimal})") else: print(f" -> Kein klarer Unterschied (minimal={minimal}, specific={specific3})") print(f"\n [2] Broad vs Specific-3 (Formulierung):") if specific3 > broad: print(f" -> Spezifische Regeln wirken besser (specific={specific3}, broad={broad})") elif broad > specific3: print(f" -> Breite Regel wirkt besser (broad={broad}, specific={specific3})") else: print(f" -> Gleichstand (broad={broad}, specific={specific3})") print(f"\n [3] Specific-3 vs English-3 (Sprache):") if specific3 > english3: print(f" -> Deutsch wirkt besser (de={specific3}, en={english3})") elif english3 > specific3: print(f" -> Englisch wirkt besser (en={english3}, de={specific3})") else: print(f" -> Kein Sprachunterschied (de={specific3}, en={english3})") best = max(pass_counts, key=pass_counts.get) print(f"\n Beste Strategie: {best} ({pass_counts[best]}/{K})") async def main(): print("E7 Experiment: Injection-Strategie (Granularitat)") 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())