"""Benchmark koşucusu: hazır kod setinde toplu değerlendirme. Her öğe için ajan akışı çalıştırılır ve şu ölçümler toplanır: * ilk tur (geri beslemesiz) ve son tur: derlenme, geçen test oranı, satır kapsama * mutasyon skoru (otomatik enjekte edilen hatalar) * elle yerleştirilmiş gerçekçi hataların (seeded bugs) yakalanma oranı * tur sayısı, LLM çağrısı, araç çağrısı, süre """ from __future__ import annotations import json import re import threading import time import uuid from pathlib import Path from statistics import mean from typing import Any, Callable, Optional from pydantic import BaseModel from .. import store from ..config import settings from ..llm.base import LLMClient from ..models import JobRequest from ..sandbox.registry import get_runner BENCH_DIR = Path(__file__).resolve().parents[3] / "benchmarks" def load_manifest() -> list[dict[str, Any]]: items = json.loads((BENCH_DIR / "manifest.json").read_text(encoding="utf-8")) en = _english() for it in items: it["code"] = (BENCH_DIR / it["file"]).read_text(encoding="utf-8") if it.get("html_file"): it["html"] = (BENCH_DIR / it["html_file"]).read_text(encoding="utf-8") _add_english(it, en) return items def _english() -> dict[str, Any]: f = BENCH_DIR / "i18n_en.json" return json.loads(f.read_text(encoding="utf-8")).get("items", {}) if f.exists() else {} def _add_english(it: dict[str, Any], en: dict[str, Any]) -> None: """Arayüzün İngilizce görünümü için başlık, açıklama ve yerleştirilmiş hata açıklamalarını ekler.""" e = en.get(it["id"]) if it.get("source") == "QuixBugs" or it["id"].startswith("qb-"): e = {"title": it["title"], "description": "QuixBugs program; original single-line bug", "bugs": ["Original QuixBugs bug"] * len(it.get("seeded_bugs", []))} if not e: return it["title_en"], it["description_en"] = e["title"], e["description"] for b, d in zip(it.get("seeded_bugs", []), e.get("bugs", [])): b["description_en"] = d class ItemResult(BaseModel): id: str title: str title_en: str = "" language: str kind: str split: str = "dev" rep: int = 1 job_id: str = "" status: str = "pending" error: str = "" first_compiled: Optional[bool] = None first_pass_rate: Optional[float] = None first_coverage: Optional[float] = None final_compiled: Optional[bool] = None final_pass_rate: Optional[float] = None final_coverage: Optional[float] = None tests: int = 0 iterations: int = 0 mutation_score: Optional[float] = None mutation_score_pre: Optional[float] = None # hedefli mutasyon turundan önce mutfix_tests_added: int = 0 energy_wh: Optional[float] = None # koşu boyunca GPU enerji tüketimi mutants_killed: int = 0 mutants_valid: int = 0 seeded_total: int = 0 seeded_caught: int = 0 seeded_details: list[dict[str, Any]] = [] suspected_bugs: int = 0 pruned_tests: int = 0 rejected_bug_reports: int = 0 llm_calls: int = 0 tool_calls: int = 0 duration_s: float = 0.0 stop_reason: str = "" stop_reason_i18n: dict[str, str] = {} gpu_percent: Optional[int] = None # modelin GPU belleğindeki payı (Ollama /api/ps) tokens_per_s: Optional[float] = None # medyan üretim hızı tokens_in: int = 0 # tüm model çağrılarında giriş / çıkış token toplamı (maliyet tabloları) tokens_out: int = 0 class EvalState(BaseModel): id: str created_at: float status: str = "running" model: str label: str = "" config: dict[str, Any] = {} items: list[ItemResult] = [] summary: dict[str, Any] = {} _evals: dict[str, EvalState] = {} def _rate(passed: int, total: int) -> Optional[float]: return round(100.0 * passed / total, 2) if total else None def check_seeded_bugs(item: dict[str, Any], test_code: str, baseline_fail: set[str]) -> list[dict[str, Any]]: runner = get_runner(item["language"], item.get("kind", "unit")) out = [] for bug in item.get("seeded_bugs", []): src = item["code"] if bug["from"] not in src: out.append({"description": bug["description"], "description_en": bug.get("description_en", ""), "caught": None, "note": "yama uygulanamadı"}) continue buggy = src.replace(bug["from"], bug["to"], 1) r = runner.run(buggy, test_code, item.get("kind", "unit"), item.get("html"), with_coverage=False) caught = (not r.compiled) or r.timed_out or bool(r.failing_names() - baseline_fail) out.append({"description": bug["description"], "description_en": bug.get("description_en", ""), "caught": caught}) return out def _mut_rounds(cfg: dict[str, Any]) -> int: """Mutant turu sayısı. Belirtilmemişse tek atışlık taban çizgisinde (en çok 1 tur) kapalıdır: o yapılandırma hiçbir geri bildirim almamalıdır. Diğer yapılandırmalarda bir tur.""" v = cfg.get("mutation_rounds") if v is None: return 0 if (cfg.get("max_iterations") or 4) == 1 else 1 return int(v) def evaluate_item(item: dict[str, Any], model: str, cfg: dict[str, Any], llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]] = None, rep: int = 1) -> ItemResult: res = ItemResult(id=item["id"], title=item["title"], title_en=item.get("title_en", ""), language=item["language"], kind=item.get("kind", "unit"), split=item.get("split", "dev"), rep=rep) req = JobRequest(language=item["language"], code=item["code"], test_kind=item.get("kind", "unit"), html=item.get("html"), model=model, max_iterations=cfg.get("max_iterations"), coverage_target=cfg.get("coverage_target"), run_mutation=cfg.get("run_mutation", True), max_mutants=cfg.get("max_mutants"), temperature=cfg.get("temperature", 0.2), mutation_rounds=_mut_rounds(cfg), ablate=cfg.get("ablate") or [], lang=cfg.get("lang", "tr"), title=f"[bench] {item['title']}") job = store.create_job(req, llm=llm_factory(item) if llm_factory else None, background=False) res.job_id = job.id res.status = job.status if job.status != "done" or not job.result: res.error = job.error or job.status return res r = job.result if r.iterations: f = r.iterations[0] res.first_compiled = f.compiled # geçen oranı koşulan testler üzerinden: atlanan / xfail testleri paydaya girmez res.first_pass_rate = _rate(f.passed, f.passed + f.failed) if f.compiled else 0.0 res.first_coverage = f.coverage if f.compiled else 0.0 res.iterations = len([x for x in r.iterations if x.action != "mutfix"]) if r.final_run: fr = r.final_run res.final_compiled = fr.compiled res.final_pass_rate = _rate(fr.passed, fr.passed + fr.failed + fr.errors) if fr.compiled else 0.0 res.final_coverage = fr.coverage.percent if fr.compiled else 0.0 res.tests = fr.total if fr.compiled and fr.total and item.get("seeded_bugs"): res.seeded_details = check_seeded_bugs(item, r.test_code, fr.failing_names()) valid = [d for d in res.seeded_details if d["caught"] is not None] res.seeded_total = len(valid) res.seeded_caught = sum(1 for d in valid if d["caught"]) if r.mutation: res.mutation_score = r.mutation.score res.mutation_score_pre = r.mutation_before.score if r.mutation_before else r.mutation.score if r.mutation_round.get("accepted"): res.mutfix_tests_added = int(r.mutation_round.get("tests_added", 0)) res.mutants_killed = r.mutation.killed + r.mutation.timeout res.mutants_valid = r.mutation.killed + r.mutation.survived + r.mutation.timeout res.suspected_bugs = len(r.suspected_bugs) res.pruned_tests = r.pruned_tests res.rejected_bug_reports = len(r.rejected_bug_reports) res.llm_calls, res.tool_calls = r.llm_calls, r.tool_calls res.duration_s, res.stop_reason = r.duration_s, r.stop_reason res.stop_reason_i18n = r.stop_reason_i18n res.gpu_percent, res.tokens_per_s = r.compute.get("gpu_percent"), r.compute.get("tokens_per_s") res.energy_wh = r.compute.get("energy_wh") res.tokens_in = sum(int(e.data.get("tokens_in") or 0) for e in job.events if e.type == "llm") res.tokens_out = sum(int(e.data.get("tokens_out") or 0) for e in job.events if e.type == "llm") return res def _avg(vals: list[Optional[float]]) -> Optional[float]: v = [x for x in vals if x is not None] return round(mean(v), 2) if v else None def summarize(items: list[ItemResult]) -> dict[str, Any]: def agg(group: list[ItemResult]) -> dict[str, Any]: done = [i for i in group if i.status == "done"] st = sum(i.seeded_total for i in done) mv = sum(i.mutants_valid for i in done) return { "n": len(group), "completed": len(done), "first_compile_rate": _rate(sum(1 for i in done if i.first_compiled), len(done)), "final_compile_rate": _rate(sum(1 for i in done if i.final_compiled), len(done)), "first_pass_rate": _avg([i.first_pass_rate for i in done]), "final_pass_rate": _avg([i.final_pass_rate for i in done]), "first_coverage": _avg([i.first_coverage for i in done]), "final_coverage": _avg([i.final_coverage for i in done]), "mutation_score": _avg([i.mutation_score for i in done]), "mutation_score_pooled": _rate(sum(i.mutants_killed for i in done), mv), "mutation_score_pre": _avg([i.mutation_score_pre for i in done]), "avg_energy_wh": _avg([i.energy_wh for i in done]), "seeded_catch_rate": _rate(sum(i.seeded_caught for i in done), st), "seeded_total": st, "avg_iterations": _avg([float(i.iterations) for i in done]), "avg_llm_calls": _avg([float(i.llm_calls) for i in done]), "avg_duration_s": _avg([i.duration_s for i in done]), "avg_tokens_in": _avg([float(i.tokens_in) for i in done if i.tokens_in]), "avg_tokens_out": _avg([float(i.tokens_out) for i in done if i.tokens_out]), } out = {"overall": agg(items), "by_language": {}, "by_kind": {}, "by_split": {}} for lang in sorted({i.language for i in items}): out["by_language"][lang] = agg([i for i in items if i.language == lang]) for kind in sorted({i.kind for i in items}): out["by_kind"][kind] = agg([i for i in items if i.kind == kind]) for split in sorted({i.split for i in items}): out["by_split"][split] = agg([i for i in items if i.split == split]) return out def _eval_dir() -> Path: d = settings.data_dir / "evals" d.mkdir(parents=True, exist_ok=True) return d def persist(ev: EvalState) -> None: (_eval_dir() / f"{ev.id}.json").write_text(ev.model_dump_json(indent=2)) def load_persisted() -> None: for f in _eval_dir().glob("*.json"): try: ev = EvalState.model_validate_json(f.read_text()) _evals.setdefault(ev.id, ev) except Exception: # noqa: BLE001 pass def run_eval(model: str, cfg: dict[str, Any], item_ids: Optional[list[str]] = None, languages: Optional[list[str]] = None, background: bool = True, llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]] = None, split: Optional[str] = None, repeats: int = 1, label: str = "") -> EvalState: items = load_manifest() if item_ids: items = [i for i in items if i["id"] in item_ids] if languages: items = [i for i in items if i["language"] in languages] if split and split != "all": items = [i for i in items if i.get("split", "dev") == split] runs = [(r, i) for r in range(1, max(1, repeats) + 1) for i in items] ev = EvalState(id=uuid.uuid4().hex[:10], created_at=time.time(), model=model, label=label, config=cfg | {"split": split or "all", "repeats": max(1, repeats)}, items=[ItemResult(id=i["id"], title=i["title"], title_en=i.get("title_en", ""), language=i["language"], kind=i.get("kind", "unit"), split=i.get("split", "dev"), rep=r) for r, i in runs]) _evals[ev.id] = ev persist(ev) _start(ev, {i["id"]: i for i in items}, llm_factory, background) return ev _stops: dict[str, threading.Event] = {} _workers: dict[str, threading.Thread] = {} # Modelin değil altyapının hatası (Ollama/Docker kapalı, bilgisayar yeniden başladı): öğe başarısız sayılmaz, # altyapı geri gelene kadar beklenip yeniden denenir; yine olmazsa "Eksikleri tamamla" ile yeniden koşulabilir. INFRA_ERR = re.compile(r"Ollama'ya ulaşılamadı|Cannot reach Ollama|ConnectError|Connection refused|" r"Cannot connect to the Docker daemon|docker: error during connect|RemoteProtocolError", re.I) INFRA_WAIT_S = 15 * 60 # altyapının geri gelmesi için en çok bekleme INFRA_POLL_S = 15 def is_infra_error(msg: Optional[str]) -> bool: return bool(msg and INFRA_ERR.search(msg)) def _wait_for_model(model: str, stop: threading.Event) -> bool: """Ollama yeniden ulaşılabilir olana kadar bekler (en çok INFRA_WAIT_S).""" from ..llm.providers import is_local, make_llm if not is_local(model): time.sleep(INFRA_POLL_S) return True t0 = time.time() while time.time() - t0 < INFRA_WAIT_S and not stop.is_set(): try: if make_llm(model).health().get("ok"): time.sleep(10) # model servisi yeni açıldıysa biraz daha bekle return True except Exception: # noqa: BLE001 pass time.sleep(INFRA_POLL_S) return False def _start(ev: EvalState, by_id: dict[str, dict[str, Any]], llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]], background: bool) -> None: """Bitmemiş (bekleyen ya da yarıda kalmış) öğeleri sırayla koşar; her öğeden sonra diske yazar. Böylece uygulama kapanıp açıldığında değerlendirme kaldığı öğeden sürdürülebilir. """ stop = _stops.setdefault(ev.id, threading.Event()) stop.clear() alive = _workers.get(ev.id) if alive is not None and alive.is_alive(): # hâlâ çalışan işçi var: yalnızca duraklatma isteğini geri al ev.status = "running" return ev.status = "running" def work() -> None: for idx, item in enumerate(ev.items): if item.status not in ("pending", "running"): continue if stop.is_set(): if ev.id in _cancel_flags: _cancel_flags.discard(ev.id) _close_cancelled(ev) return ev.status = "paused" persist(ev) return src = by_id.get(item.id) if src is None: ev.items[idx].status, ev.items[idx].error = "failed", "öğe veri setinde yok" continue ev.items[idx].status = "running" persist(ev) for attempt in range(3): try: ev.items[idx] = evaluate_item(src, ev.model, ev.config, llm_factory, rep=item.rep) except Exception as e: # noqa: BLE001 ev.items[idx].status = "failed" ev.items[idx].error = str(e) res = ev.items[idx] if res.status == "done" or not is_infra_error(res.error) or llm_factory is not None or attempt == 2: break persist(ev) if not _wait_for_model(ev.model, stop): break ev.summary = summarize(ev.items) persist(ev) if ev.id in _cancel_flags: # iptal son öğe sırasında geldi _cancel_flags.discard(ev.id) _close_cancelled(ev) return ev.status = "done" ev.summary = summarize(ev.items) persist(ev) if background: th = threading.Thread(target=work, daemon=True) _workers[ev.id] = th th.start() else: work() def worker_alive(eval_id: str) -> bool: th = _workers.get(eval_id) return th is not None and th.is_alive() def resume_eval(eval_id: str, background: bool = True, llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]] = None) -> Optional[EvalState]: """Yarıda kalmış ya da duraklatılmış bir değerlendirmeyi kaldığı öğeden sürdürür.""" ev = _evals.get(eval_id) if ev is None or ev.status in ("done", "cancelled"): return ev if ev.status == "running" and worker_alive(eval_id): _stops.setdefault(eval_id, threading.Event()).clear() # duraklatma isteği varsa iptal return ev _start(ev, {i["id"]: i for i in load_manifest()}, llm_factory, background) return ev def reopen_infra_failures(eval_id: str) -> int: """Altyapı hatasıyla başarısız olan öğeleri yeniden koşulmak üzere bekleyen duruma alır.""" ev = _evals.get(eval_id) if ev is None or (ev.status == "running" and worker_alive(eval_id)): return 0 n = 0 for i, it in enumerate(ev.items): if it.status == "failed" and is_infra_error(it.error): ev.items[i] = ItemResult(id=it.id, title=it.title, title_en=it.title_en, language=it.language, kind=it.kind, split=it.split, rep=it.rep) n += 1 if n: ev.status = "paused" ev.summary = summarize(ev.items) persist(ev) return n def pause_eval(eval_id: str) -> bool: """Süren öğe bittikten sonra durur (öğe yarıda kesilmez).""" ev = _evals.get(eval_id) if ev is None or ev.status != "running": return False _stops.setdefault(eval_id, threading.Event()).set() return True def cancel_eval(eval_id: str) -> bool: """Değerlendirmeyi kalıcı durdurur: süren öğenin işi hemen iptal edilir, kalan öğeler koşulmaz.""" ev = _evals.get(eval_id) if ev is None or ev.status in ("done", "cancelled"): return False _stops.setdefault(eval_id, threading.Event()).set() _cancel_flags.add(eval_id) for j in store.all_jobs(): # süren değerlendirme işi (aynı anda tek değerlendirme koşar) if j.status in ("running", "queued") and (j.request.title or "").startswith("[bench]"): store.cancel(j.id) if not worker_alive(eval_id): _close_cancelled(ev) return True _cancel_flags: set[str] = set() def _close_cancelled(ev: EvalState) -> None: for it in ev.items: if it.status in ("pending", "running"): it.status = "cancelled" ev.status = "cancelled" ev.summary = summarize(ev.items) persist(ev) def interrupted() -> list[EvalState]: """Uygulama kapanırken 'running' kalmış (iş parçacığı artık olmayan) değerlendirmeler.""" return [e for e in _evals.values() if e.status == "running" and e.id not in _stops] def get_eval(eval_id: str) -> Optional[EvalState]: return _evals.get(eval_id) def all_evals() -> list[EvalState]: return sorted(_evals.values(), key=lambda e: e.created_at, reverse=True)