"""backend/api/benchmark_handler.py — Gestore benchmark via Telegram. Versione v7 (GAP-BENCH-2): usa benchmark-extended.mjs (comprehensive, 2319 righe) con flag --json. Mantiene compatibilità backward con report v6.2-stress per /riepilogo. Sprint history: v6.2 — benchmark-ultra-v6.2-stress.mjs (208 righe, stress+Groq judge) v7 — benchmark-extended.mjs (2319 righe, 10+ categorie, HF datasets, ref vs Replit/Cursor/Devin/Manus) Fix invarianti portati da v6.2: - FIX-1: process.kill() + wait() su TimeoutError — nessun zombie su Railway. - FIX-2: _load_gap_map() usa importlib isolato — nessuna sys.path pollution. - FIX-3: I/O su file in asyncio.to_thread — non blocca l'event loop. """ from __future__ import annotations import asyncio, importlib, importlib.util, json, logging, os from typing import Any logger = logging.getLogger("agente_ai.benchmark_handler") # ── Percorsi server Railway ──────────────────────────────────────────────────── _REPO_ROOT = os.getenv("REPO_ROOT", "/home/ubuntu/Baida98_AI") # v7 (GAP-BENCH-2) _BENCH_SCRIPT = os.path.join(_REPO_ROOT, "scripts", "benchmark-extended.mjs") _REPORT_V7 = "/tmp/agente-ai/benchmark-v7-latest.json" _BENCH_TIMEOUT = float(os.getenv("BENCH_TIMEOUT_SECS", "720")) # 12 min (era 360s) # v6.2 — usato come fallback in get_smart_summary per compatibilità _REPORT_V6 = os.path.join(_REPO_ROOT, "benchmark-stress-report.json") async def run_benchmark_task(chat_id: int, send_reply_fn) -> None: """Esegue benchmark-extended.mjs v7 con --json e invia risultati via Telegram. Flag --json → scrive /tmp/agente-ai/benchmark-v7-latest.json. Variabili env richieste (Railway): GROQ_API_KEY, INTERNAL_TOKEN. """ await send_reply_fn( chat_id, "🚀 Avvio Benchmark Extended v7…\n" "10+ categorie · HF datasets · ref vs Replit/Cursor/Devin/Manus · ~10-12 min.", ) env = { **os.environ, "GROQ_API_KEY": os.getenv("GROQ_API_KEY", ""), "NVIDIA_API_KEY": os.getenv("NVIDIA_API_KEY", ""), "INTERNAL_TOKEN": os.getenv("INTERNAL_TOKEN", ""), } process: asyncio.subprocess.Process | None = None try: process = await asyncio.create_subprocess_exec( "node", _BENCH_SCRIPT, "--json", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, ) stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=_BENCH_TIMEOUT ) if process.returncode != 0: err = stderr.decode(errors="replace")[:400] logger.error("Benchmark v7 failed rc=%d: %s", process.returncode, err) await send_reply_fn(chat_id, f"❌ Errore benchmark v7:\n{err}") return except asyncio.TimeoutError: # FIX-1: kill del processo figlio prima di notificare if process is not None: try: process.kill() await process.wait() except Exception: pass logger.warning("Benchmark v7 timeout (>%.0fs) — process killed", _BENCH_TIMEOUT) await send_reply_fn( chat_id, f"⏱ Timeout benchmark v7 (>{int(_BENCH_TIMEOUT // 60)} min) — " "processo terminato, controlla log Railway.", ) return except Exception as exc: if process is not None: try: process.kill() await process.wait() except Exception: pass logger.error("run_benchmark_task v7 error: %s", exc, exc_info=True) await send_reply_fn(chat_id, f"💥 Errore critico: {exc}") return report_exists = await asyncio.to_thread(os.path.exists, _REPORT_V7) if not report_exists: await send_reply_fn(chat_id, "⚠️ Benchmark terminato ma report v7 non trovato.") return try: # FIX-3: lettura file in thread — non blocca l'event loop report: dict[str, Any] = await asyncio.to_thread(_read_json, _REPORT_V7) except Exception as exc: await send_reply_fn(chat_id, f"⚠️ Report v7 non leggibile: {exc}") return await send_reply_fn(chat_id, _format_v7_report(report)) def _format_v7_report(report: dict[str, Any]) -> str: """Formatta il report v7 per Telegram HTML.""" s = report.get("summary", {}) ts = (report.get("timestamp") or "")[:16].replace("T", " ") ver = report.get("version", "extended-v7") avg = s.get("avgScore", "N/A") repl = s.get("avgReplit", "N/A") curs = s.get("avgCursor", "N/A") devi = s.get("avgDevin", "N/A") manu = s.get("avgManus", "N/A") gaps = s.get("gapCount", 0) verd = s.get("verdict", "") canary = s.get("canaryLeaks", 0) lines: list[str] = [ f"🏆 Benchmark {ver} completato!\n\n" f"📊 Score agente: {avg}/100\n" f"📅 Run: {ts}\n\n" "📈 Confronto vs riferimenti:\n" f" • Replit: {repl}/100\n" f" • Cursor: {curs}/100\n" f" • Devin: {devi}/100\n" f" • Manus: {manu}/100\n" ] if verd: lines.append(f"\n📝 Verdetto: {verd}\n") if canary: lines.append(f"⚠️ Canary leak: {canary} task\n") # Score per categoria tasks = report.get("tasks", []) if tasks: by_cat: dict[str, list[float]] = {} for t in tasks: cat = t.get("cat", "?") sc = t.get("score") if isinstance(sc, (int, float)): by_cat.setdefault(cat, []).append(float(sc)) if by_cat: lines.append("\n📂 Per categoria:\n") for cat, scores in sorted(by_cat.items()): avg_cat = sum(scores) / len(scores) icon = "🟢" if avg_cat >= 70 else "🟡" if avg_cat >= 50 else "🔴" lines.append(f" {icon} {avg_cat:5.1f} {cat}\n") # Gap cards (prime 3) gap_cards = report.get("gapCards", []) if gap_cards: lines.append(f"\n💡 Gap ({gaps} totali):\n") for gc in gap_cards[:3]: gid = gc.get("id", "?") gtit = gc.get("title", gc.get("name", "")) gsev = gc.get("severity", "") lines.append(f" • {gid} {gtit}" + (f" [{gsev}]" if gsev else "") + "\n") if gaps > 3: lines.append(f" ...e altri {gaps - 3} gap.\n") lines.append("\n🔍 Usa /riepilogo per analisi approfondita.") return "".join(lines) # ── Helpers ─────────────────────────────────────────────────────────────────── def _read_json(path: str) -> dict[str, Any]: """Lettura JSON sincrona — da eseguire sempre in asyncio.to_thread.""" with open(path, encoding="utf-8") as f: return json.load(f) def _load_gap_map() -> tuple[list, list]: """Importa GAPS e CATS da scripts/gap_map.py senza inquinare sys.path. FIX-2: usa importlib.util.spec_from_file_location per un import isolato. """ gap_map_path = os.path.join(_REPO_ROOT, "scripts", "gap_map.py") if not os.path.exists(gap_map_path): logger.debug("gap_map.py not found at %s", gap_map_path) return [], [] try: spec = importlib.util.spec_from_file_location("_gap_map_isolated", gap_map_path) if spec is None or spec.loader is None: return [], [] mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) # type: ignore[union-attr] return getattr(mod, "GAPS", []), getattr(mod, "CATS", []) except Exception as exc: logger.warning("_load_gap_map failed: %s", exc) return [], [] async def get_smart_summary(chat_id: int) -> str: """Genera riepilogo strutturato: stato sistema + ultimo score + gap priority. GAP-BENCH-2: prova prima report v7, fallback a v6.2 per compatibilità. """ GAPS, _ = await asyncio.to_thread(_load_gap_map) lines: list[str] = ["📋 Briefing Assistente Proattivo\n\n"] lines.append("🟢 Stato Sistema: Railway operativo\n") # Prova v7 per prima v7_exists = await asyncio.to_thread(os.path.exists, _REPORT_V7) v6_exists = await asyncio.to_thread(os.path.exists, _REPORT_V6) if v7_exists: try: report: dict[str, Any] = await asyncio.to_thread(_read_json, _REPORT_V7) s = report.get("summary", {}) avg = s.get("avgScore", "N/A") ts = (report.get("timestamp") or "")[:16].replace("T", " ") ver = report.get("version", "v7") lines.append(f"📊 Ultimo Score ({ver}): {avg}/100\n") lines.append(f"📅 Run: {ts}\n\n") # Categorie con score < 50 → alimenta gap priority failed_cats: set[str] = set() for t in report.get("tasks", []): sc = t.get("score") if isinstance(sc, (int, float)) and sc < 50: failed_cats.add(t.get("cat", "").lower()) if failed_cats and GAPS: lines.append("💡 Aree prioritarie:\n") for gap in GAPS: if any(c in failed_cats for c in gap.get("categories", [])): slug = ( f"feature/{gap['id'].lower()}" f"-{gap['name'].lower().replace(' ', '-')}" ) lines.append(f" • {gap['name']}{slug}\n") else: lines.append("✨ Nessun gap critico nell'ultimo run.\n") except Exception as exc: logger.warning("get_smart_summary v7 error: %s", exc) lines.append("⚠️ Errore lettura report v7 — esegui /bench per aggiornare.\n") elif v6_exists: # Fallback v6.2 — compatibilità try: report = await asyncio.to_thread(_read_json, _REPORT_V6) score = report.get("finalScore", "N/A") ts = (report.get("timestamp") or "")[:16].replace("T", " ") judge = report.get("judge", "heuristic") lines.append(f"📊 Ultimo Score (v6.2): {score}/100 [{judge}]\n") lines.append(f"📅 Run: {ts}\n\n") failed_cats: set[str] = set() for res in report.get("results", []): if res.get("score", {}).get("total", 100) < 50: rid = res.get("id", "") if rid.startswith("STRESS_AMB"): failed_cats.add("recovery") if rid.startswith("STRESS_REC"): failed_cats.add("recovery") if rid.startswith("STRESS_MEM"): failed_cats.add("memory_context") if rid.startswith("WEB"): failed_cats.add("orchestration") if rid.startswith("CODE"): failed_cats.add("bug_fix") if rid.startswith("REASON"): failed_cats.add("reasoning") if failed_cats and GAPS: lines.append("💡 Aree prioritarie:\n") for gap in GAPS: if any(c in failed_cats for c in gap.get("categories", [])): slug = ( f"feature/{gap['id'].lower()}" f"-{gap['name'].lower().replace(' ', '-')}" ) lines.append(f" • {gap['name']}{slug}\n") else: lines.append("✨ Nessun gap critico nell'ultimo run.\n") except Exception as exc: logger.warning("get_smart_summary v6 error: %s", exc) lines.append("⚠️ Errore lettura report v6.2 — esegui /bench per aggiornare.\n") else: lines.append("📊 Nessun report — esegui /bench per generare dati.\n") lines.append("\n🚀 Task in corso: controlla /status per stato live.") return "".join(lines)