core/continuous_benchmark.py

"""
benchmark/continuous_benchmark.py — VORTEX GOD v1.2
Benchmark continu avec détection de régression.

Tourne en thread daemon et :
  - Évalue le système toutes les N heures sur un suite de tâches fixes
  - Détecte les régressions vs le meilleur score historique
  - Enregistre les résultats dans benchmark_history.jsonl
  - Expose une API simple pour l'onglet Gradio
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional

log = logging.getLogger("vortex.benchmark")

DATA_DIR       = Path(os.environ.get("DATA_DIR",   "/app/data"))
BENCH_LOG      = DATA_DIR / "benchmark_history.jsonl"
BENCH_INTERVAL = float(os.environ.get("BENCH_INTERVAL_HOURS", "6.0"))
REGRESSION_THR = float(os.environ.get("REGRESSION_THRESHOLD", "0.05"))   # -5% = régression


@dataclass
class BenchmarkRun:
    run_id:     str
    ts:         float = field(default_factory=time.time)
    scores:     Dict[str, float] = field(default_factory=dict)
    global_score: float = 0.0
    duration_s: float = 0.0
    regression: bool  = False
    regression_details: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict:
        return {
            "run_id":      self.run_id,
            "ts":          self.ts,
            "scores":      self.scores,
            "global_score": round(self.global_score, 4),
            "duration_s":  round(self.duration_s, 1),
            "regression":  self.regression,
            "regression_details": self.regression_details,
        }


# ─────────────────────────────────────────────
# Suite de tâches benchmark (déterministes)
# ─────────────────────────────────────────────

BENCHMARK_SUITE = {
    "engineer": [
        {
            "user": "Écris une fonction Python `fibonacci(n)` qui retourne le n-ième nombre de Fibonacci.",
            "checks": [
                lambda r: "def fibonacci" in r,
                lambda r: "return" in r,
                lambda r: any(w in r for w in ["n-1", "n - 1", "n-2", "n - 2", "memo", "cache"]),
            ],
        },
        {
            "user": "Écris une fonction `is_palindrome(s)` qui vérifie si une chaîne est un palindrome.",
            "checks": [
                lambda r: "def is_palindrome" in r,
                lambda r: "return" in r,
                lambda r: any(w in r for w in ["reverse", "[::-1]", "lower", "==", "!="]),
            ],
        },
        {
            "user": "Écris un générateur Python qui yield les nombres premiers jusqu'à N.",
            "checks": [
                lambda r: "def " in r,
                lambda r: "yield" in r,
                lambda r: any(w in r for w in ["prime", "premier", "divisible", "sqrt", "%"]),
            ],
        },
    ],
    "planner": [
        {
            "user": "Planifie le déploiement d'une application Flask sur un serveur Ubuntu.",
            "checks": [
                lambda r: any(w in r.lower() for w in ["étape", "step", "install", "nginx", "gunicorn", "systemd"]),
                lambda r: len(r) > 100,
            ],
        },
        {
            "user": "Planifie la mise en place d'un pipeline de données ETL.",
            "checks": [
                lambda r: any(w in r.lower() for w in ["extract", "transform", "load", "source", "étape"]),
                lambda r: len(r) > 80,
            ],
        },
    ],
    "critic": [
        {
            "user": "Audite : import pickle; data = pickle.loads(user_input)",
            "checks": [
                lambda r: any(w in r.lower() for w in ["pickle", "dangereux", "dangerous", "injection", "unsafe"]),
            ],
        },
        {
            "user": "Audite : cursor.execute('DELETE FROM users WHERE id=' + user_id)",
            "checks": [
                lambda r: any(w in r.lower() for w in ["injection", "sql", "paramètre", "parameter", "dangereux"]),
            ],
        },
    ],
    "researcher": [
        {
            "user": "Qu'est-ce que le RAG (Retrieval-Augmented Generation) ?",
            "checks": [
                lambda r: any(w in r.lower() for w in ["retrieval", "récupération", "vectoriel", "contexte", "llm"]),
                lambda r: len(r) > 100,
            ],
        },
    ],
    "optimizer": [
        {
            "user": "Optimise : for i in range(len(lst)): if lst[i] in target: result.append(lst[i])",
            "checks": [
                lambda r: any(w in r.lower() for w in ["set", "intersection", "comprehension", "filter", "o(1)"]),
            ],
        },
    ],
}

AGENT_SYSTEMS = {
    "engineer":   "Tu es un ingénieur Python. Génère uniquement du code Python valide.",
    "planner":    "Tu es un planificateur. Décris les étapes principales en 5 lignes max.",
    "critic":     "Tu es un expert sécurité. Identifie les problèmes critiques.",
    "researcher": "Tu es un chercheur. Explique en 3-5 phrases.",
    "optimizer":  "Tu es un optimiseur Python. Propose une version améliorée.",
}


class ContinuousBenchmark:
    """
    Benchmark continu qui évalue le système en appelant directement le LLM engine.
    Ne dépend pas des agents spécialisés (pour isoler les régressions du LLM seul).
    """

    def __init__(self, llm_engine):
        self._engine   = llm_engine
        self.history:  List[BenchmarkRun] = self._load_history()
        self._best:    float = max((r.global_score for r in self.history), default=0.0)
        self._running: bool  = False

    def _load_history(self) -> List[BenchmarkRun]:
        runs = []
        if BENCH_LOG.exists():
            with open(BENCH_LOG) as f:
                for line in f:
                    try:
                        d = json.loads(line)
                        runs.append(BenchmarkRun(
                            run_id       = d["run_id"],
                            ts           = d["ts"],
                            scores       = d["scores"],
                            global_score = d["global_score"],
                            duration_s   = d["duration_s"],
                            regression   = d.get("regression", False),
                        ))
                    except Exception:
                        pass
        return runs

    # ── Évaluation ───────────────────────────────────────────────────────────

    async def _eval_task(self, task_type: str, task: Dict) -> float:
        """Évalue une tâche individuelle. Retourne 0.0–1.0."""
        system = AGENT_SYSTEMS.get(task_type, "Tu es un assistant.")
        try:
            resp = await self._engine.call(
                agent       = None,
                system      = system,
                user        = task["user"],
                max_tokens  = 400,
                temperature = 0.1,
                use_cache   = False,
            )
            content = resp.content
            checks  = task.get("checks", [])
            if not checks:
                return 0.5
            passed = sum(1 for c in checks if c(content))
            return round(passed / len(checks), 3)
        except Exception as exc:
            log.debug(f"[Bench] Tâche {task_type} échouée : {exc}")
            return 0.0

    async def run_once(self) -> BenchmarkRun:
        """Lance un passage complet du benchmark. Retourne le BenchmarkRun."""
        import uuid
        run     = BenchmarkRun(run_id=str(uuid.uuid4())[:8])
        t0      = time.time()
        scores  = {}
        regs    = []

        for task_type, tasks in BENCHMARK_SUITE.items():
            task_scores = await asyncio.gather(
                *[self._eval_task(task_type, t) for t in tasks],
                return_exceptions=True,
            )
            valid  = [s for s in task_scores if isinstance(s, float)]
            avg    = round(sum(valid) / len(valid), 4) if valid else 0.0
            scores[task_type] = avg

            # Détecter régression par rapport au meilleur historique
            best_for_type = max(
                (r.scores.get(task_type, 0) for r in self.history),
                default=0.0,
            )
            if best_for_type > 0.1 and avg < best_for_type - REGRESSION_THR:
                regs.append(f"{task_type}: {avg:.3f} vs best {best_for_type:.3f} (Δ={avg-best_for_type:+.3f})")

        run.scores      = scores
        run.global_score = round(sum(scores.values()) / len(scores), 4) if scores else 0.0
        run.duration_s  = time.time() - t0
        run.regression  = bool(regs)
        run.regression_details = regs

        if run.regression:
            log.warning(f"[Bench] ⚠️ RÉGRESSION détectée : {regs}")
        else:
            log.info(f"[Bench] Score global : {run.global_score:.4f} (best={self._best:.4f})")

        if run.global_score > self._best:
            self._best = run.global_score

        self.history.append(run)
        self._save_run(run)
        return run

    def _save_run(self, run: BenchmarkRun):
        DATA_DIR.mkdir(parents=True, exist_ok=True)
        with open(BENCH_LOG, "a") as f:
            f.write(json.dumps(run.to_dict(), ensure_ascii=False) + "\n")

    # ── Boucle daemon ────────────────────────────────────────────────────────

    async def run_forever(self, interval_hours: float = BENCH_INTERVAL):
        self._running = True
        log.info(f"[Bench] Benchmark continu démarré (intervalle={interval_hours}h)")
        # Premier run immédiat au démarrage
        await asyncio.sleep(60)
        while self._running:
            try:
                await self.run_once()
            except Exception as exc:
                log.error(f"[Bench] Erreur : {exc}")
            await asyncio.sleep(interval_hours * 3600)

    def stop(self):
        self._running = False

    # ── API pour Gradio ───────────────────────────────────────────────────────

    def get_dashboard(self) -> Dict[str, Any]:
        """Données pour l'onglet Gradio."""
        if not self.history:
            return {"status": "Aucun benchmark exécuté", "history": []}

        last = self.history[-1]
        trend = []
        for r in self.history[-20:]:
            trend.append({"ts": r.ts, "score": r.global_score, "regression": r.regression})

        return {
            "last_run": {
                "run_id":      last.run_id,
                "global_score": last.global_score,
                "scores":      last.scores,
                "duration_s":  last.duration_s,
                "regression":  last.regression,
                "regression_details": last.regression_details,
            },
            "best_score": self._best,
            "total_runs": len(self.history),
            "regressions": sum(1 for r in self.history if r.regression),
            "trend":       trend,
        }