from contextlib import asynccontextmanager from dotenv import load_dotenv import json import logging import traceback load_dotenv() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s") import yaml import asyncio from fastapi import Depends, FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from sqlalchemy.orm import Session from sqlalchemy import func from sqlalchemy.exc import IntegrityError from pydantic import BaseModel from database import create_all, get_db, SessionLocal from models import Task, Agent, Run, RunStatus from core.runner import Runner from core.metrics import compute_pass_at_k from core.statistics import wilson_interval, task_bootstrap_ci, paired_bootstrap_delta logger = logging.getLogger(__name__) runner = Runner() def _log_task_exception(task: asyncio.Task): """Callback for background asyncio tasks — logs exceptions instead of silently swallowing them.""" if task.cancelled(): return exc = task.exception() if exc: logger.error( "Background run task failed: %s\n%s", exc, "".join(traceback.format_exception(exc)), ) @asynccontextmanager async def lifespan(app: FastAPI): create_all() # Sweep zombie runs from previous server instance db = SessionLocal() try: zombies = db.query(Run).filter(Run.status.in_([RunStatus.pending, RunStatus.running])).all() for run in zombies: run.status = RunStatus.error run.judge_notes = "Marked as error: server restarted while run was in progress" if zombies: db.commit() finally: db.close() yield app = FastAPI(title="AgentArena", lifespan=lifespan) app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") # ── Helpers ────────────────────────────────────────────────────────────────── def _agent_stats(db: Session): """Return list of dicts with per-agent aggregate statistics.""" agents = db.query(Agent).all() result = [] for agent in agents: runs = agent.runs if not runs: result.append({ "agent": agent, "tasks_solved": 0, "pass_at_1": None, "pass_at_3": None, "avg_cost": None, "avg_duration": None, "avg_composite": None, "total_runs": 0, }) continue finished = [r for r in runs if r.status in (RunStatus.success, RunStatus.failure, RunStatus.error, RunStatus.timeout)] successful = [r for r in runs if (r.functional_success or 0) >= 1] # Tasks where at least one run succeeded tasks_solved = len(set(r.task_id for r in successful)) # Group finished runs per task_id for unbiased pass@k estimation runs_per_task: dict = {} for r in finished: runs_per_task.setdefault(r.task_id, []).append(r) # pass@1 / pass@3: unbiased estimators aggregated across all tasks pass_at_1_values = [] pass_at_3_values = [] for task_runs in runs_per_task.values(): n = len(task_runs) c = sum(1 for r in task_runs if (r.functional_success or 0) >= 1) pass_at_1_values.append(compute_pass_at_k(n, c, 1)) if n >= 3: pass_at_3_values.append(compute_pass_at_k(n, c, 3)) pass_at_1 = (sum(pass_at_1_values) / len(pass_at_1_values)) if pass_at_1_values else None pass_at_3 = (sum(pass_at_3_values) / len(pass_at_3_values)) if pass_at_3_values else None avg_cost = (sum(r.cost_usd for r in finished) / len(finished)) if finished else None avg_duration = (sum((r.agent_duration_s or r.duration_s or 0.0) for r in finished) / len(finished)) if finished else None functional_successes = sum(1 for r in finished if (r.functional_success or 0) >= 1) functional_ci = wilson_interval(functional_successes, len(finished)) if finished else (0.0, 0.0) task_alignment_scores = [] for task_runs in runs_per_task.values(): per_task = [r.alignment_score_llm for r in task_runs if r.alignment_score_llm is not None] if per_task: task_alignment_scores.append(sum(per_task) / len(per_task)) avg_alignment = ( sum(task_alignment_scores) / len(task_alignment_scores) if task_alignment_scores else None ) alignment_ci = ( task_bootstrap_ci(task_alignment_scores) if task_alignment_scores else (0.0, 0.0) ) composite_scores = [] for r in finished: try: m = json.loads(r.metrics_json or "{}") cs = m.get("composite_score") if cs is not None: composite_scores.append(float(cs)) except (json.JSONDecodeError, TypeError, ValueError): pass avg_composite = (sum(composite_scores) / len(composite_scores)) if composite_scores else None result.append({ "agent": agent, "tasks_solved": tasks_solved, "pass_at_1": pass_at_1, "pass_at_3": pass_at_3, "avg_cost": avg_cost, "avg_duration": avg_duration, "functional_success_rate": (functional_successes / len(finished)) if finished else None, "functional_ci_low": functional_ci[0], "functional_ci_high": functional_ci[1], "avg_alignment": avg_alignment, "alignment_ci_low": alignment_ci[0], "alignment_ci_high": alignment_ci[1], "avg_composite": avg_composite, "total_runs": len(finished), }) result.sort( key=lambda x: ( -(x["tasks_solved"]), -(x["functional_success_rate"] or 0), -(x["avg_alignment"] or 0), ) ) return result # ── Routes ──────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) def index(request: Request, db: Session = Depends(get_db)): stats = _agent_stats(db) tasks = db.query(Task).order_by(Task.created_at.desc()).all() return templates.TemplateResponse("index.html", { "request": request, "agent_stats": stats, "tasks": tasks, }) @app.get("/tasks/{slug}", response_class=HTMLResponse) def task_detail(slug: str, request: Request, db: Session = Depends(get_db)): task = db.query(Task).filter(Task.slug == slug).first() if not task: return HTMLResponse("Task not found", status_code=404) runs = ( db.query(Run) .filter(Run.task_id == task.id) .order_by(Run.created_at.desc()) .all() ) # Group by agent — best result per agent (success > others, then by score) best_by_agent: dict[int, Run] = {} for run in runs: existing = best_by_agent.get(run.agent_id) if existing is None: best_by_agent[run.agent_id] = run else: # Prefer success if (run.functional_success or 0) >= 1 and (existing.functional_success or 0) < 1: best_by_agent[run.agent_id] = run elif (run.functional_success or 0) == (existing.functional_success or 0): # Same status — prefer higher judge_score run_score = run.judge_score or 0 existing_score = existing.judge_score or 0 if run_score > existing_score: best_by_agent[run.agent_id] = run agents = {a.id: a for a in db.query(Agent).all()} return templates.TemplateResponse("task_detail.html", { "request": request, "task": task, "runs": runs, "best_by_agent": list(best_by_agent.values()), "agents": agents, }) @app.get("/runs/{run_id}", response_class=HTMLResponse) def run_detail(run_id: int, request: Request, db: Session = Depends(get_db)): run = db.query(Run).filter(Run.id == run_id).first() if not run: return HTMLResponse("Run not found", status_code=404) try: trajectory = json.loads(run.trajectory or "[]") except (json.JSONDecodeError, TypeError): trajectory = [] diff_lines = [] if run.diff_text: for line in run.diff_text.splitlines(): if line.startswith("+") and not line.startswith("+++"): diff_lines.append(("added", line)) elif line.startswith("-") and not line.startswith("---"): diff_lines.append(("removed", line)) else: diff_lines.append(("context", line)) try: metrics = json.loads(run.metrics_json or "{}") except (json.JSONDecodeError, TypeError): metrics = {} return templates.TemplateResponse("run_detail.html", { "request": request, "run": run, "trajectory": trajectory, "diff_lines": diff_lines, "metrics": metrics, }) # ── API ─────────────────────────────────────────────────────────────────────── class RunRequest(BaseModel): task_slug: str agent_name: str adapter_type: str = "claude_code" model: str = "" k: int = 1 class AgentCreate(BaseModel): name: str adapter_type: str model: str = "" config_json: str = "{}" @app.post("/api/agents") def create_agent(req: AgentCreate, db: Session = Depends(get_db)): existing = db.query(Agent).filter(Agent.name == req.name).first() if existing: return JSONResponse({"error": "Agent already exists"}, status_code=409) # Merge model into config_json if provided. config = json.loads(req.config_json) if req.config_json and req.config_json != "{}" else {} if req.model: config["model"] = req.model agent = Agent(name=req.name, adapter_type=req.adapter_type, config_json=json.dumps(config)) db.add(agent) db.commit() db.refresh(agent) return {"id": agent.id, "name": agent.name} @app.post("/api/runs") async def start_run(req: RunRequest, db: Session = Depends(get_db)): task = db.query(Task).filter(Task.slug == req.task_slug).first() if not task: return JSONResponse({"error": "Task not found"}, status_code=404) # Auto-create agent if it doesn't exist (race-safe) agent = db.query(Agent).filter(Agent.name == req.agent_name).first() if not agent: try: config = {"model": req.model} if req.model else {} agent = Agent( name=req.agent_name, adapter_type=req.adapter_type, config_json=json.dumps(config), ) db.add(agent) db.commit() db.refresh(agent) except IntegrityError: db.rollback() agent = db.query(Agent).filter(Agent.name == req.agent_name).first() if req.k > 1: task_id, agent_id = task.id, agent.id async def _run_k(): run_db = SessionLocal() try: t = run_db.query(Task).filter(Task.id == task_id).first() a = run_db.query(Agent).filter(Agent.id == agent_id).first() await runner.run_pass_at_k(run_db, t, a, k=req.k) finally: run_db.close() bg_task = asyncio.create_task(_run_k()) bg_task.add_done_callback(_log_task_exception) return {"status": "started", "k": req.k, "agent": agent.name} # Single run — fire in background (runner.run_single creates its own Run record) task_id, agent_id = task.id, agent.id async def _run_single(): run_db = SessionLocal() try: t = run_db.query(Task).filter(Task.id == task_id).first() a = run_db.query(Agent).filter(Agent.id == agent_id).first() await runner.run_single(run_db, t, a) finally: run_db.close() bg_task = asyncio.create_task(_run_single()) bg_task.add_done_callback(_log_task_exception) return {"status": "started", "agent": agent.name, "task": task.slug} @app.get("/api/runs/{run_id}/stream") async def stream_run(run_id: int): async def generate(): while True: db = SessionLocal() try: run = db.query(Run).filter(Run.id == run_id).first() if not run: yield f"event: error\ndata: {{\"error\": \"not found\"}}\n\n" return data = { "status": run.status.value, "duration_s": run.duration_s, "agent_duration_s": run.agent_duration_s, "end_to_end_duration_s": run.end_to_end_duration_s, "tokens_input": run.tokens_input, "tokens_output": run.tokens_output, "cost_usd": run.cost_usd, "tests_passed": run.tests_passed, "tests_total": run.tests_total, "functional_success": run.functional_success, "alignment_score_llm": run.alignment_score_llm, } yield f"event: update\ndata: {json.dumps(data)}\n\n" if run.status not in (RunStatus.pending, RunStatus.running): yield f"event: done\ndata: {json.dumps(data)}\n\n" return finally: db.close() await asyncio.sleep(2) return StreamingResponse( generate(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @app.post("/api/tasks/import") async def import_task(request: Request, db: Session = Depends(get_db)): body = await request.body() try: data = yaml.safe_load(body) except yaml.YAMLError as e: return JSONResponse({"error": f"Invalid YAML: {e}"}, status_code=400) existing = db.query(Task).filter(Task.slug == data.get("slug", "")).first() if existing: return JSONResponse({"error": "Task slug already exists"}, status_code=409) 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.get("limits", {}).get("time_seconds", 300), token_limit=data.get("limits", {}).get("max_tokens", 200000), difficulty=data.get("difficulty", "medium"), ) db.add(task) db.commit() db.refresh(task) return {"task_id": task.id, "slug": task.slug} @app.get("/api/leaderboard") def leaderboard_json(db: Session = Depends(get_db)): stats = _agent_stats(db) return [ { "rank": i + 1, "agent_name": s["agent"].name, "agent_id": s["agent"].id, "tasks_solved": s["tasks_solved"], "functional_success_rate": s["functional_success_rate"], "functional_ci_low": s["functional_ci_low"], "functional_ci_high": s["functional_ci_high"], "pass_at_1": s["pass_at_1"], "pass_at_3": s["pass_at_3"], "avg_alignment": s["avg_alignment"], "alignment_ci_low": s["alignment_ci_low"], "alignment_ci_high": s["alignment_ci_high"], "avg_composite_score": s["avg_composite"], "avg_cost_usd": s["avg_cost"], "avg_duration_s": s["avg_duration"], "total_runs": s["total_runs"], } for i, s in enumerate(stats) ] @app.get("/api/tasks") def list_tasks(db: Session = Depends(get_db)): tasks = db.query(Task).order_by(Task.created_at.desc()).all() return [ { "id": t.id, "slug": t.slug, "title": t.title, "difficulty": t.difficulty, "time_limit_s": t.time_limit_s, } for t in tasks ] @app.get("/api/agents") def list_agents(db: Session = Depends(get_db)): agents = db.query(Agent).all() return [{"id": a.id, "name": a.name, "adapter_type": a.adapter_type} for a in agents] @app.get("/api/experiments/{experiment_id}/compare") def compare_experiment(experiment_id: str, left_arm: str, right_arm: str, db: Session = Depends(get_db)): runs = ( db.query(Run) .filter( Run.experiment_id == experiment_id, Run.arm_id.in_([left_arm, right_arm]), Run.status.in_([RunStatus.success, RunStatus.failure, RunStatus.error, RunStatus.timeout]), ) .all() ) if not runs: return JSONResponse({"error": "No runs found for experiment"}, status_code=404) by_arm_task: dict[tuple[str, int], list[Run]] = {} for run in runs: by_arm_task.setdefault((run.arm_id, run.task_id), []).append(run) shared_tasks = sorted({ task_id for arm_id, task_id in by_arm_task if (left_arm, task_id) in by_arm_task and (right_arm, task_id) in by_arm_task }) if not shared_tasks: return JSONResponse({"error": "No shared tasks between arms"}, status_code=400) left_func = [] right_func = [] left_align = [] right_align = [] for task_id in shared_tasks: left_runs = by_arm_task[(left_arm, task_id)] right_runs = by_arm_task[(right_arm, task_id)] left_func.append(sum((r.functional_success or 0) for r in left_runs) / len(left_runs)) right_func.append(sum((r.functional_success or 0) for r in right_runs) / len(right_runs)) left_align_runs = [r.alignment_score_llm for r in left_runs if r.alignment_score_llm is not None] right_align_runs = [r.alignment_score_llm for r in right_runs if r.alignment_score_llm is not None] if left_align_runs and right_align_runs: left_align.append(sum(left_align_runs) / len(left_align_runs)) right_align.append(sum(right_align_runs) / len(right_align_runs)) response = { "experiment_id": experiment_id, "left_arm": left_arm, "right_arm": right_arm, "shared_tasks": len(shared_tasks), "functional_success_delta": paired_bootstrap_delta(left_func, right_func), } if left_align and right_align and len(left_align) == len(right_align): response["alignment_delta"] = paired_bootstrap_delta(left_align, right_align) return response # ── Matrix View ────────────────────────────────────────────────────────────── def _matrix_data(db: Session) -> dict: """Build harness x model matrix from all runs.""" agents = db.query(Agent).all() cells: dict[tuple, list] = {} # (adapter_type, model) -> list of runs all_models: set[str] = set() all_harnesses: set[str] = set() for agent in agents: config = json.loads(agent.config_json or "{}") model = config.get("model", "unknown") harness = agent.adapter_type all_models.add(model) all_harnesses.add(harness) key = (harness, model) if key not in cells: cells[key] = [] finished = [ r for r in agent.runs if r.status in (RunStatus.success, RunStatus.failure, RunStatus.error, RunStatus.timeout) ] cells[key].extend(finished) matrix = {} for key, runs in cells.items(): if not runs: continue n = len(runs) c = sum(1 for r in runs if (r.functional_success or 0) >= 1) pass_1 = compute_pass_at_k(n, c, 1) avg_cost = sum(r.cost_usd or 0 for r in runs) / n avg_duration = sum((r.agent_duration_s or r.duration_s or 0) for r in runs) / n avg_alignment = ( sum(r.alignment_score_llm for r in runs if r.alignment_score_llm is not None) / max(1, sum(1 for r in runs if r.alignment_score_llm is not None)) ) if any(r.alignment_score_llm is not None for r in runs) else None composite_scores = [] for r in runs: try: m = json.loads(r.metrics_json or "{}") cs = m.get("composite_score") if cs is not None: composite_scores.append(float(cs)) except (json.JSONDecodeError, TypeError, ValueError): pass avg_composite = (sum(composite_scores) / len(composite_scores)) if composite_scores else None matrix[key] = { "pass_at_1": round(pass_1, 3), "avg_composite": round(avg_composite, 3) if avg_composite else None, "avg_cost": round(avg_cost, 6), "avg_duration": round(avg_duration, 1), "avg_alignment": round(avg_alignment, 3) if avg_alignment is not None else None, "total_runs": n, "tasks_solved": len(set(r.task_id for r in runs if (r.functional_success or 0) >= 1)), } return { "models": sorted(all_models), "harnesses": sorted(all_harnesses), "cells": {f"{h}|{m}": v for (h, m), v in matrix.items()}, } @app.get("/matrix", response_class=HTMLResponse) def matrix_view(request: Request, db: Session = Depends(get_db)): data = _matrix_data(db) return templates.TemplateResponse("matrix.html", { "request": request, "models": data["models"], "harnesses": data["harnesses"], "cells": data["cells"], }) @app.get("/api/matrix") def matrix_json(db: Session = Depends(get_db)): return _matrix_data(db) # ── Harness Report ─────────────────────────────────────────────────────────── def _harness_report(db: Session, adapter_type: str) -> dict | None: """Generate comprehensive report for a harness type.""" agents = db.query(Agent).filter(Agent.adapter_type == adapter_type).all() if not agents: return None all_runs = [] for agent in agents: all_runs.extend(agent.runs) finished = [ r for r in all_runs if r.status in (RunStatus.success, RunStatus.failure, RunStatus.error, RunStatus.timeout) ] successful = [r for r in all_runs if (r.functional_success or 0) >= 1] if not finished: return {"adapter_type": adapter_type, "total_runs": 0} n = len(finished) c = len(successful) pass_at_1 = compute_pass_at_k(n, c, 1) avg_cost = sum(r.cost_usd or 0 for r in finished) / n cost_per_solved = (sum(r.cost_usd or 0 for r in successful) / c) if c > 0 else None avg_duration = sum((r.agent_duration_s or r.duration_s or 0) for r in finished) / n functional_ci = wilson_interval(c, n) diff_minimalities = [] lint_deltas = [] complexity_deltas = [] directness_scores = [] exploration_ratios = [] phase_totals = {"exploration": 0, "implementation": 0, "testing": 0, "debugging": 0} composite_scores = [] alignment_task_scores = [] runs_per_task: dict[int, list] = {} for r in finished: runs_per_task.setdefault(r.task_id, []).append(r) try: m = json.loads(r.metrics_json or "{}") except (json.JSONDecodeError, TypeError, ValueError): continue diff = m.get("diff", {}) churn = diff.get("churn") if churn is not None: diff_minimalities.append(1.0 / (1.0 + churn / 50.0)) judge = m.get("judge", {}) if "lint_delta" in judge: lint_deltas.append(judge["lint_delta"]) if "complexity_delta" in judge: complexity_deltas.append(judge["complexity_delta"]) traj = m.get("trajectory", {}) if "directness_score" in traj: directness_scores.append(traj["directness_score"]) if "exploration_ratio" in traj: exploration_ratios.append(traj["exploration_ratio"]) phases = traj.get("phase_breakdown", {}) for pk, pv in phases.items(): phase_totals[pk] = phase_totals.get(pk, 0) + pv cs = m.get("composite_score") if cs is not None: composite_scores.append(float(cs)) for task_runs in runs_per_task.values(): per_task_alignment = [r.alignment_score_llm for r in task_runs if r.alignment_score_llm is not None] if per_task_alignment: alignment_task_scores.append(sum(per_task_alignment) / len(per_task_alignment)) weak_tasks = [] for task_id, task_runs in runs_per_task.items(): tn = len(task_runs) tc = sum(1 for r in task_runs if (r.functional_success or 0) >= 1) task_pass = compute_pass_at_k(tn, tc, 1) if task_pass < 0.5: task_obj = task_runs[0].task weak_tasks.append({ "task_slug": task_obj.slug if task_obj else f"task_{task_id}", "task_title": task_obj.title if task_obj else f"Task {task_id}", "pass_at_1": round(task_pass, 3), "runs": tn, }) avg_composite = (sum(composite_scores) / len(composite_scores)) if composite_scores else None cost_benefit = (avg_composite / avg_cost) if avg_composite and avg_cost > 0 else None def _safe_avg(lst): return round(sum(lst) / len(lst), 3) if lst else None return { "adapter_type": adapter_type, "total_runs": n, "total_agents": len(agents), "tasks_attempted": len(runs_per_task), "tasks_solved": len(set(r.task_id for r in successful)), "reliability": { "pass_at_1": round(pass_at_1, 3), "success_rate": round(c / n, 3) if n > 0 else 0, "success_ci_low": round(functional_ci[0], 3), "success_ci_high": round(functional_ci[1], 3), }, "efficiency": { "avg_cost_usd": round(avg_cost, 6), "cost_per_solved_task": round(cost_per_solved, 6) if cost_per_solved else None, "avg_duration_s": round(avg_duration, 1), }, "cleanliness": { "avg_diff_minimality": _safe_avg(diff_minimalities), "avg_lint_delta": _safe_avg(lint_deltas), "avg_complexity_delta": _safe_avg(complexity_deltas), }, "work_style": { "avg_directness": _safe_avg(directness_scores), "avg_exploration_ratio": _safe_avg(exploration_ratios), "phase_breakdown": phase_totals, }, "alignment": { "avg_alignment": round(sum(alignment_task_scores) / len(alignment_task_scores), 3) if alignment_task_scores else None, "ci": ( [round(v, 3) for v in task_bootstrap_ci(alignment_task_scores)] if alignment_task_scores else None ), }, "weaknesses": sorted(weak_tasks, key=lambda x: x["pass_at_1"]), "cost_benefit": { "avg_composite": round(avg_composite, 3) if avg_composite else None, "composite_per_dollar": round(cost_benefit, 2) if cost_benefit else None, }, } @app.get("/harness/{adapter_type}", response_class=HTMLResponse) def harness_report_view(adapter_type: str, request: Request, db: Session = Depends(get_db)): report = _harness_report(db, adapter_type) if report is None: return HTMLResponse("Harness not found", status_code=404) return templates.TemplateResponse("harness_report.html", { "request": request, "report": report, }) @app.get("/api/harness/{adapter_type}/report") def harness_report_json(adapter_type: str, db: Session = Depends(get_db)): report = _harness_report(db, adapter_type) if report is None: return JSONResponse({"error": "No agents found for this adapter type"}, status_code=404) return report