"""backend/api/telemetry.py — Timing + repair metrics endpoint (BG-3). Endpoint: GET /api/telemetry Ritorna avg/p50/p90 per ogni chiave in _TIMING_STORE e contatori _REPAIR_STATS. Response: { ok, timing: { : { avg, p50, p90, n } }, repair: { : int } } Nessuna autenticazione richiesta (dati aggregati, zero PII). """ import statistics from fastapi import APIRouter from fastapi.responses import JSONResponse router = APIRouter() def _percentile(data: list[float], p: int) -> float: if not data: return 0.0 sorted_data = sorted(data) k = max(0, int(len(sorted_data) * p / 100) - 1) return round(sorted_data[k], 2) @router.get("/api/telemetry") async def get_telemetry() -> JSONResponse: """Aggrega _TIMING_STORE e _REPAIR_STATS, ritorna avg/p50/p90.""" try: from api.state import _TIMING_STORE, _REPAIR_STATS except ImportError: return JSONResponse({"ok": False, "error": "state module non disponibile"}, status_code=503) timing: dict[str, dict] = {} for key, buf in _TIMING_STORE.items(): samples = list(buf) # snapshot thread-safe (list.append è GIL-safe) if samples: timing[key] = { "avg": round(sum(samples) / len(samples), 2), "p50": _percentile(samples, 50), "p90": _percentile(samples, 90), "n": len(samples), } else: timing[key] = {"avg": 0.0, "p50": 0.0, "p90": 0.0, "n": 0} repair = dict(_REPAIR_STATS) return JSONResponse({"ok": True, "timing": timing, "repair": repair}) # ── P20-F1: Fast-pass telemetry endpoint ───────────────────────────────────── @router.get("/api/telemetry/fast-pass") async def get_fast_pass_stats() -> JSONResponse: """P20-F1: Statistiche fast-pass (short-circuit goal verifier) aggregate. fast_pass_true = goal soddisfatto al primo check (nessun repair). fast_pass_false = repair necessario (coverage < soglia). fast_pass_rate = true / total (0.0–1.0, ideale > 0.7 → pochi repair). """ try: from api.state import _REPAIR_STATS except ImportError: return JSONResponse({"ok": False, "error": "state module non disponibile"}, status_code=503) stats = dict(_REPAIR_STATS) fp_true = int(stats.get("fast_pass_true", 0)) fp_false = int(stats.get("fast_pass_false", 0)) total = fp_true + fp_false rate = round(fp_true / total, 4) if total else 0.0 p36_hits = int(stats.get("p36_fast_path_hit", 0)) return JSONResponse({ "ok": True, "fast_pass_true": fp_true, "fast_pass_false": fp_false, "total_checks": total, "fast_pass_rate": rate, "p36_fast_path_hit": p36_hits, # P36: chiamate python_analyze via fast-path (<5ms) "_note": ( "fast_pass_true = goal già soddisfatto al primo check; " "fast_pass_false = repair necessario. " "Fonte: _REPAIR_STATS['fast_pass_true/false'] in api.state." ), }) # ── GAP-NEW-5: Telemetry alert loop ───────────────────────────────────────── # Campiona _REPAIR_STATS e _TIMING_STORE ogni 5 min. # Se una soglia è superata → notifica Telegram via telegram_notify.notify_task_error(). _ALERT_THRESHOLDS: dict[str, float] = { "repair_per_5min": 50.0, # > 50 riparazioni in 5 min → alert "p90_tool_call_ms": 50_000.0, # p90 latenza tool > 50s → alert } _alert_prev_repair: dict[str, int] = {} # snapshot precedente _REPAIR_STATS _alert_error_count: int = 0 # errori consecutivi rilevati async def telemetry_alert_loop() -> None: """GAP-NEW-5: Loop background — campiona metriche ogni 5 min, alert su soglie. Avviato da main.py _on_startup() con asyncio.create_task(). Fire-and-forget: non solleva mai eccezioni al caller. Usa notify_task_error() per l'invio Telegram (già rate-limited + retry). """ global _alert_prev_repair, _alert_error_count import asyncio as _aio # Prima run: attende 5 min così il server è a regime await _aio.sleep(300) while True: try: from api.state import _TIMING_STORE, _REPAIR_STATS from api.telegram_notify import notify_task_error as _tg_err alerts: list[str] = [] # ── Controllo 1: repair rate ────────────────────────────────── current_repair = dict(_REPAIR_STATS) if _alert_prev_repair: delta = { k: current_repair.get(k, 0) - _alert_prev_repair.get(k, 0) for k in current_repair } total_repairs = sum(v for v in delta.values() if v > 0) if total_repairs > _ALERT_THRESHOLDS["repair_per_5min"]: alerts.append( f"⚠️ Repair rate elevato: {total_repairs} riparazioni in 5 min " f"(soglia: {int(_ALERT_THRESHOLDS['repair_per_5min'])}). " f"Dettaglio: {dict(list(delta.items())[:5])}" ) _alert_prev_repair = current_repair # ── Controllo 2: p90 latenza tool ──────────────────────────── for key, buf in _TIMING_STORE.items(): samples = list(buf) if len(samples) < 10: continue # troppo pochi campioni — ignora sorted_s = sorted(samples) p90_ms = sorted_s[max(0, int(len(sorted_s) * 0.90) - 1)] if p90_ms > _ALERT_THRESHOLDS["p90_tool_call_ms"]: alerts.append( f"🐌 Latenza p90 elevata: {key} = {p90_ms/1000:.1f}s " f"(soglia: {_ALERT_THRESHOLDS['p90_tool_call_ms']/1000:.0f}s)" ) # ── Invia alert se necessario ───────────────────────────────── if alerts: _alert_error_count += 1 alert_body = "\n".join(alerts) # Usa task_id fittizio "telemetry-watchdog" per dedup (max 1 alert/5min) await _tg_err( task_id=f"telemetry-watchdog-{_alert_error_count}", goal="Monitoraggio telemetria automatica", error=alert_body[:600], ) else: _alert_error_count = 0 # reset streak se tutto ok except Exception as _tel_loop_err: # Silent — il loop non deve mai crashare il server import logging as _log _log.getLogger("agente_ai").debug("telemetry_alert_loop error: %s", str(_tel_loop_err)[:80]) await _aio.sleep(300) # 5 min tra ogni campionamento