""" skill_tracker.py — GAP-SKILL-SYNC: Session-scoped adaptive tool success/failure tracker. GAP-SUP1: Supabase persistence — skill stats sopravvivono ai riavvii del backend. Design originale: - In-memory Dict[session_id, Dict[tool_name, SkillStats]] — zero DB dep, zero latency - SkillStats: success_count, fail_count, last_used, total_latency_ms - record() sincrono — GIL-safe in CPython, asyncio single-threaded - get_sorted_fallbacks() — Wilson score lower bound (95% CI) per robustezza su n piccoli - get_stats() — JSON-serializable per /api/agent/skill-stats endpoint - clear_session() — cleanup opzionale fine task (evita memory leak su run lunghissimi) GAP-SUP1 (Supabase persistence): - _supabase_upsert(): httpx POST → PostgREST /rest/v1/skill_stats (upsert conflict) - _supabase_load_session(): httpx GET → ripristina sessione precedente al boot - record() fire-and-forget ogni _SYNC_EVERY_N chiamate per tool/sessione - load_session_from_cloud(): chiamato da unified_loop al boot sessione - Fallback silente se SUPABASE_URL/SUPABASE_ANON_KEY assenti → comportamento invariato - Timeout conservativo 8s — mai blocca il loop principale - Schema SQL: backend/migrations/gap1_skill_stats.sql Integrazione: - unified_loop.py: record() dopo ogni executor.run_tool() — registra successo/fallimento - unified_loop.py: load_session_from_cloud() al boot sessione (se Supabase abilitato) - api/agent.py: GET /api/agent/skill-stats/{session_id} per merge Dexie frontend Singleton: get_skill_tracker() restituisce sempre lo stesso SkillTracker globale. """ from __future__ import annotations import asyncio import math import os import time import logging from collections import defaultdict from dataclasses import dataclass from typing import Any import httpx _logger = logging.getLogger("agente_ai.skill_tracker") # ─── Supabase config (GAP-SUP1) ─────────────────────────────────────────────── _SUPA_URL = os.getenv("SUPABASE_URL", "").rstrip("/") _SUPA_KEY = os.getenv("SUPABASE_ANON_KEY", "") _SUPA_ENABLED = bool(_SUPA_URL and _SUPA_KEY) _SUPA_TABLE = "skill_stats" # tabella PostgREST — vedi gap1_skill_stats.sql _SYNC_EVERY_N = 5 # upsert ogni N record() per tool/sessione (throttle) if _SUPA_ENABLED: _logger.info("[skill_tracker] Supabase persistence ABILITATA → %s/rest/v1/%s", _SUPA_URL, _SUPA_TABLE) else: _logger.debug("[skill_tracker] Supabase non configurato — solo in-memory") # ─── SkillStats ─────────────────────────────────────────────────────────────── @dataclass class SkillStats: success_count: int = 0 fail_count: int = 0 last_used: float = 0.0 total_latency_ms: float = 0.0 @property def total_count(self) -> int: return self.success_count + self.fail_count @property def success_rate(self) -> float: if not self.total_count: return 1.0 # ottimismo iniziale — tool mai usato return self.success_count / self.total_count @property def avg_latency_ms(self) -> float: if not self.total_count: return 0.0 return self.total_latency_ms / self.total_count def wilson_score(self) -> float: """Wilson score lower bound (95% CI). Bilanciamento statisticamente robusto tra success rate e confidenza. Esempio: tool 1/1 (score ≈ 0.21) vs tool 10/11 (score ≈ 0.68) — il secondo viene preferito anche se il primo ha 100% raw rate. Usato da get_sorted_fallbacks() per ordinamento adattivo. """ n = self.total_count if n == 0: return 0.5 # prior neutro su tool mai usati in questa sessione p = self.success_count / n z = 1.96 # 95% confidence interval num = p + z * z / (2 * n) - z * math.sqrt((p * (1 - p) + z * z / (4 * n)) / n) den = 1 + z * z / n return num / den def to_supabase_row(self, session_id: str, tool_name: str) -> dict: """Serializza per upsert PostgREST.""" return { "session_id": session_id, "tool_name": tool_name, "success_count": self.success_count, "fail_count": self.fail_count, "last_used": self.last_used, "total_latency_ms": self.total_latency_ms, } @classmethod def from_supabase_row(cls, row: dict) -> "SkillStats": """Deserializza da riga PostgREST.""" return cls( success_count = int(row.get("success_count", 0)), fail_count = int(row.get("fail_count", 0)), last_used = float(row.get("last_used", 0.0)), total_latency_ms = float(row.get("total_latency_ms", 0.0)), ) # ─── Supabase helpers (GAP-SUP1) ────────────────────────────────────────────── async def _supabase_upsert(session_id: str, tool_name: str, stats: SkillStats) -> None: """Fire-and-forget: upsert riga skill_stats su Supabase (PostgREST). Fallback silente su qualsiasi errore — mai blocca il loop principale. Timeout 8s conservativo. """ if not _SUPA_ENABLED: return row = stats.to_supabase_row(session_id, tool_name) try: async with httpx.AsyncClient(timeout=8.0) as client: resp = await client.post( f"{_SUPA_URL}/rest/v1/{_SUPA_TABLE}", json=row, headers={ "apikey": _SUPA_KEY, "Authorization": f"Bearer {_SUPA_KEY}", "Content-Type": "application/json", "Prefer": "resolution=merge-duplicates,return=minimal", }, ) if resp.status_code not in (200, 201, 204): _logger.debug( "[skill_tracker] supabase upsert %s: HTTP %d %s", tool_name[:20], resp.status_code, resp.text[:120], ) except Exception as exc: # noqa: BLE001 _logger.debug("[skill_tracker] supabase upsert silenced: %s", type(exc).__name__) async def _supabase_load_session(session_id: str) -> dict[str, SkillStats]: """Carica tutti i tool stats di una sessione da Supabase. Ritorna dict vuoto su qualsiasi errore (fallback silente). Chiamato da load_session_from_cloud() al boot sessione. """ if not _SUPA_ENABLED: return {} try: async with httpx.AsyncClient(timeout=8.0) as client: resp = await client.get( f"{_SUPA_URL}/rest/v1/{_SUPA_TABLE}", params={"session_id": f"eq.{session_id}", "select": "*"}, headers={ "apikey": _SUPA_KEY, "Authorization": f"Bearer {_SUPA_KEY}", }, ) if resp.status_code != 200: _logger.debug( "[skill_tracker] supabase load %s: HTTP %d", session_id[:12], resp.status_code, ) return {} rows: list[dict] = resp.json() loaded = { row["tool_name"]: SkillStats.from_supabase_row(row) for row in rows if "tool_name" in row } if loaded: _logger.info( "[skill_tracker] GAP-SUP1: ripristinati %d tool stats per sessione %s", len(loaded), session_id[:12], ) return loaded except Exception as exc: # noqa: BLE001 _logger.debug("[skill_tracker] supabase load silenced: %s", type(exc).__name__) return {} # ─── SkillTracker ───────────────────────────────────────────────────────────── class SkillTracker: """Singleton session-scoped tracker: impara quali tool funzionano per ogni sessione. GAP-SUP1: i dati persistono su Supabase e vengono ripristinati al boot sessione. """ def __init__(self) -> None: self._sessions: dict[str, dict[str, SkillStats]] = defaultdict( lambda: defaultdict(SkillStats) ) # Contatori per throttle upsert (session_id → tool_name → count_since_last_sync) self._sync_counters: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int)) # ── Write ───────────────────────────────────────────────────────────────── def record( self, session_id: str, tool_name: str, success: bool, latency_ms: float = 0.0, ) -> None: """Registra il risultato di una chiamata tool (sincrono, GIL-safe). Chiamato da unified_loop.py dopo ogni executor.run_tool(). GAP-SUP1: fire-and-forget upsert Supabase ogni _SYNC_EVERY_N chiamate. """ s = self._sessions[session_id][tool_name] if success: s.success_count += 1 else: s.fail_count += 1 s.last_used = time.monotonic() s.total_latency_ms += latency_ms # GAP-SUP1: sync throttled — ogni _SYNC_EVERY_N record per questo tool/sessione if _SUPA_ENABLED: self._sync_counters[session_id][tool_name] += 1 if self._sync_counters[session_id][tool_name] >= _SYNC_EVERY_N: self._sync_counters[session_id][tool_name] = 0 try: loop = asyncio.get_running_loop() loop.create_task( _supabase_upsert(session_id, tool_name, s), name=f"skill_sync_{tool_name[:20]}", ) except RuntimeError: pass # no running loop (test context) — silente # ── Cloud bootstrap (GAP-SUP1) ──────────────────────────────────────────── async def load_session_from_cloud(self, session_id: str) -> int: """Ripristina stats precedenti da Supabase per la sessione (chiamare al boot task). Merge con in-memory: somma i contatori (in-memory è vuoto al boot, ma sicuro). Ritorna numero di tool ripristinati (0 se Supabase non configurato). Idempotente: chiamate multiple sommano i dati — chiamare una sola volta per sessione. """ loaded = await _supabase_load_session(session_id) if not loaded: return 0 session = self._sessions[session_id] for tool_name, cloud_stats in loaded.items(): mem = session[tool_name] # Merge additivo — in-memory è tipicamente vuoto al boot mem.success_count += cloud_stats.success_count mem.fail_count += cloud_stats.fail_count mem.total_latency_ms += cloud_stats.total_latency_ms # last_used: prendi il più recente if cloud_stats.last_used > mem.last_used: mem.last_used = cloud_stats.last_used return len(loaded) async def flush_session_to_cloud(self, session_id: str) -> int: """Forza upsert di tutti i tool di una sessione su Supabase (chiamare a fine task). Ritorna numero di tool sincronizzati. Fallback silente su errori. """ if not _SUPA_ENABLED: return 0 session = self._sessions.get(session_id, {}) tasks = [ _supabase_upsert(session_id, tool_name, stats) for tool_name, stats in session.items() ] if tasks: await asyncio.gather(*tasks, return_exceptions=True) _logger.info( "[skill_tracker] GAP-SUP1: flush %d tool stats per sessione %s", len(tasks), session_id[:12], ) return len(tasks) # ── Read / routing ──────────────────────────────────────────────────────── def get_sorted_fallbacks( self, session_id: str, candidates: list[str], ) -> list[str]: """Riordina i candidati tool per Wilson score decrescente. Tool mai usati in questa sessione → prior neutro 0.5, non penalizzati. Uso tipico: reordina la lista fallback prima di provarli. """ session = self._sessions.get(session_id, {}) return sorted( candidates, key=lambda t: session[t].wilson_score() if t in session else 0.5, reverse=True, ) def get_stats(self, session_id: str) -> dict[str, dict]: """Statistiche JSON-serializable per una sessione (ordinato per Wilson score desc).""" session = self._sessions.get(session_id, {}) return { tool: { "success_count": s.success_count, "fail_count": s.fail_count, "total_count": s.total_count, "success_rate": round(s.success_rate, 3), "wilson_score": round(s.wilson_score(), 3), "avg_latency_ms": round(s.avg_latency_ms, 1), "last_used": round(s.last_used, 3), } for tool, s in sorted( session.items(), key=lambda kv: kv[1].wilson_score(), reverse=True, ) } def get_all_sessions(self) -> dict[str, Any]: """Debug: panoramica tutte le sessioni attive.""" return { sid: { "tool_count": len(tools), "total_calls": sum(s.total_count for s in tools.values()), "tools": list(tools.keys()), } for sid, tools in self._sessions.items() } def clear_session(self, session_id: str) -> None: """Libera memoria per una sessione terminata.""" removed = self._sessions.pop(session_id, None) self._sync_counters.pop(session_id, None) if removed is not None: _logger.debug( "[skill_tracker] cleared session %s (%d tools tracked)", session_id[:12], len(removed), ) # ─── Singleton globale ──────────────────────────────────────────────────────── _skill_tracker = SkillTracker() def get_skill_tracker() -> SkillTracker: """Restituisce il singleton SkillTracker. Thread-safe in CPython (GIL).""" return _skill_tracker # ─── P17-B2: FastAPI router per sync frontend ───────────────────────────────── # Endpoint REST che permette al frontend (Dexie) di leggere e scrivere skill stats. # Montato in backend/main.py tramite _on_startup() se importato. try: from fastapi import APIRouter as _APIRouter from pydantic import BaseModel as _BM skill_router = _APIRouter(prefix="/api/agent", tags=["skill-tracker"]) class _SkillRecordBody(_BM): success: bool latency_ms: float = 0.0 error_msg: str = "" @skill_router.get("/skill-stats/{session_id}") async def api_get_skill_stats(session_id: str): """Restituisce stats tool per sessione — per merge con Dexie frontend.""" return get_skill_tracker().get_stats(session_id) @skill_router.post("/skill-record/{session_id}/{tool_name}") async def api_record_skill(session_id: str, tool_name: str, body: _SkillRecordBody): """Registra un risultato tool dal frontend (es. tool chiamato via browser).""" get_skill_tracker().record( session_id, tool_name, success=body.success, latency_ms=body.latency_ms, ) return {"ok": True, "session_id": session_id, "tool": tool_name} @skill_router.delete("/skill-stats/{session_id}") async def api_clear_skill_session(session_id: str): """Pulisce la sessione skill tracker alla fine del task.""" get_skill_tracker().clear_session(session_id) return {"ok": True} except ImportError: skill_router = None # type: ignore[assignment] # FastAPI non disponibile (unit test env)