#!/usr/bin/env python3 """E8 Experiment: Domain Knowledge — Baseline vs Injection across 4 Scenarios 4 Szenarien × 2 Agents (baseline/injected) × 3 Runs (pass@3) = 24 Runs E8a enum-convention: Does project knowledge about UPPERCASE enums fix the bug? E8b timeout-unit: Does knowledge about millisecond timeouts fix the bug? E8c breaking-change: Does knowledge about the V2 API migration fix the bug? E8d implicit-null: Does knowledge about NOT NULL constraints fix the bug? Hypothesis: Domain-knowledge injection (Sidecar-NG) enables agents to solve project-specific bugs that baseline agents cannot solve without context. """ 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 # --- Constants --- MINIMAL_CLAUDE_MD = """# Settings Use tools to solve tasks. Be concise. """ K = 3 # pass@3 # --- Scenarios --- SCENARIOS = { "e8a": { "slug": "e8a", "task_slug": "enum-convention-bug", "task_yaml": "enum_convention_bug.yaml", "injection": ( "In diesem Projekt verwenden alle Enums UPPERCASE Werte (z.B. 'ACTIVE', 'INACTIVE'). " "Die Migrations zeigen das Pattern." ), }, "e8b": { "slug": "e8b", "task_slug": "timeout-unit-bug", "task_yaml": "timeout_unit_bug.yaml", "injection": ( "Die API-Client-Klasse verwendet Millisekunden für alle Timeout-Werte (timeout_ms), " "nicht Sekunden." ), }, "e8c": { "slug": "e8c", "task_slug": "breaking-change-bug", "task_yaml": "breaking_change_bug.yaml", "injection": ( "database.py wurde auf V2 migriert: find_one() heißt jetzt get_by_id(). " "Siehe database.py für die aktuelle API." ), }, "e8d": { "slug": "e8d", "task_slug": "implicit-null-bug", "task_yaml": "implicit_null_bug.yaml", "injection": ( "Das DB-Schema (schema.sql) hat NOT NULL Constraints auf email. " "None/NULL ist nicht erlaubt." ), }, } def _build_agents() -> dict[str, dict]: """Generate agent specs dynamically from SCENARIOS.""" agents = {} for scenario_key, scenario in SCENARIOS.items(): slug = scenario["slug"] # Baseline: no injection agents[f"{slug}-baseline"] = { "adapter_type": "claude_code", "scenario": scenario_key, "variant": "baseline", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, }, } # Injected: with additional_context agents[f"{slug}-injected"] = { "adapter_type": "claude_code", "scenario": scenario_key, "variant": "injected", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": scenario["injection"], "clean_home": True, }, } return agents AGENTS = _build_agents() async def setup_agents(db) -> dict[str, Agent]: """Create or update agents in the database.""" 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_tasks(db) -> dict[str, Task]: """Import all scenario tasks from YAML if not already in DB.""" import yaml tasks_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tasks", ) tasks = {} for scenario_key, scenario in SCENARIOS.items(): task_slug = scenario["task_slug"] task = db.query(Task).filter(Task.slug == task_slug).first() if not task: yaml_path = os.path.join(tasks_dir, scenario["task_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}") tasks[scenario_key] = task return tasks 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], tasks: dict[str, Task]): """Print per-scenario baseline vs injected results, then overall verdict.""" print("\n" + "=" * 60) print("E8 RESULTS: Domain Knowledge — Baseline vs Injection") print("=" * 60) overall_baseline_pass = 0 overall_injected_pass = 0 overall_scenarios = 0 for scenario_key, scenario in SCENARIOS.items(): task = tasks[scenario_key] slug = scenario["slug"] baseline_name = f"{slug}-baseline" injected_name = f"{slug}-injected" print(f"\n{'=' * 60}") print(f"Scenario {slug.upper()}: {scenario['task_slug']}") print(f"{'=' * 60}") scenario_results = {} for variant_name in (baseline_name, injected_name): agent = agents[variant_name] 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) scenario_results[variant_name] = passed print(f"\n{'─' * 40}") print(f"Agent: {variant_name}") print(f" pass@{K}: {passed}/{total}") 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}") # Per-scenario mini-verdict b_pass = scenario_results[baseline_name] i_pass = scenario_results[injected_name] overall_baseline_pass += b_pass overall_injected_pass += i_pass overall_scenarios += 1 print(f"\n Scenario verdict: baseline={b_pass}/{K}, injected={i_pass}/{K}") if i_pass > b_pass: print(f" -> Injection WIRKT fuer {slug.upper()}") elif i_pass == b_pass: if i_pass == K: print(f" -> Beide bestanden — Task evtl. zu einfach fuer Differenzierung") else: print(f" -> Kein Unterschied — Injection hat nicht geholfen") else: print(f" -> Baseline besser?! Unerwartetes Ergebnis — genauer untersuchen") # Overall verdict print(f"\n{'=' * 60}") print("GESAMTVERDIKT") print(f"{'=' * 60}") print(f" Scenarios: {overall_scenarios}") print(f" Baseline total: {overall_baseline_pass}/{overall_scenarios * K}") print(f" Injected total: {overall_injected_pass}/{overall_scenarios * K}") print() if overall_injected_pass > overall_baseline_pass: delta = overall_injected_pass - overall_baseline_pass print(f" -> Domain-Knowledge-Injection WIRKT: +{delta} Pass(es) ueber alle Szenarien") elif overall_injected_pass == overall_baseline_pass: print(" -> Kein Gesamtunterschied — Injection neutral") else: delta = overall_baseline_pass - overall_injected_pass print(f" -> Baseline besser um {delta} — Injection hat geschadet?!") print() async def main(): print("E8 Experiment: Domain Knowledge — Baseline vs Injection") print("=" * 60) print(f"Scenarios: {len(SCENARIOS)} | Agents per scenario: 2 | K={K}") print(f"Total runs: {len(SCENARIOS) * 2 * K}") create_all() db = SessionLocal() runner = Runner() try: # Setup print("\n[1/3] Setting up...") tasks = await import_tasks(db) agents = await setup_agents(db) # Run experiments sequentially total_agents = len(agents) total_runs = total_agents * K print(f"\n[2/3] Running {total_agents} agents x {K} runs = {total_runs} total runs...") for name, agent in agents.items(): spec = AGENTS[name] scenario_key = spec["scenario"] task = tasks[scenario_key] 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 _functional_pass(r)) print(f" -> {name}: {passed}/{len(runs)} passed") # Results print_results(db, agents, tasks) finally: db.close() if __name__ == "__main__": asyncio.run(main())