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