sync: 2 changed, 0 deleted — from push d9b29fe0 (2026-06-18T20:42 UTC)

agents/skill_tracker.py

@@ -1,10 +1,8 @@
 """
 skill_tracker.py — GAP-SKILL-SYNC: Session-scoped adaptive tool success/failure tracker.
+GAP-SUP1: Supabase persistence — skill stats sopravvivono ai riavvii del backend.
 
-Chiude il ciclo di feedback mancante: il backend impara quali tool funzionano meglio
-per una data sessione e li preferisce nelle iterazioni successive via Wilson score.
-
-Design:
+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
@@ -12,24 +10,52 @@ Design:
   - 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
@@ -72,16 +98,115 @@ class SkillStats:
         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."""
+    """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 ─────────────────────────────────────────────────────────────────
 
@@ -95,7 +220,7 @@ class SkillTracker:
         """Registra il risultato di una chiamata tool (sincrono, GIL-safe).
 
         Chiamato da unified_loop.py dopo ogni executor.run_tool().
-        Zero overhead: solo incremento contatori in-memory.
+        GAP-SUP1: fire-and-forget upsert Supabase ogni _SYNC_EVERY_N chiamate.
         """
         s = self._sessions[session_id][tool_name]
         if success:
@@ -105,6 +230,64 @@ class SkillTracker:
         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(
@@ -158,6 +341,7 @@ class SkillTracker:
     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)",

migrations/gap1_skill_stats.sql

@@ -0,0 +1,85 @@
+-- GAP-SUP1: Supabase Skill Tracker persistence
+-- Eseguire nel SQL Editor di Supabase:
+--   https://app.supabase.com/project/_/sql/new
+--
+-- Tabella skill_stats: statistiche tool per sessione (backend skill_tracker.py)
+-- Tabella skill_patterns: pattern tool sequenze (frontend skillPersistence.ts)
+
+-- ─── skill_stats ──────────────────────────────────────────────────────────────
+
+CREATE TABLE IF NOT EXISTS skill_stats (
+  session_id       text             NOT NULL,
+  tool_name        text             NOT NULL,
+  success_count    integer          NOT NULL DEFAULT 0,
+  fail_count       integer          NOT NULL DEFAULT 0,
+  last_used        double precision NOT NULL DEFAULT 0,
+  total_latency_ms double precision NOT NULL DEFAULT 0,
+  updated_at       timestamptz      NOT NULL DEFAULT now(),
+  PRIMARY KEY (session_id, tool_name)
+);
+
+-- Index per query per sessione (GET ?session_id=eq.xxx)
+CREATE INDEX IF NOT EXISTS idx_skill_stats_session ON skill_stats (session_id);
+
+-- Row Level Security: anon può leggere e scrivere (chiave anon pubblica)
+ALTER TABLE skill_stats ENABLE ROW LEVEL SECURITY;
+
+DO $$ BEGIN
+  IF NOT EXISTS (
+    SELECT 1 FROM pg_policies
+    WHERE tablename = 'skill_stats' AND policyname = 'anon full access'
+  ) THEN
+    CREATE POLICY "anon full access" ON skill_stats
+      FOR ALL TO anon
+      USING (true)
+      WITH CHECK (true);
+  END IF;
+END $$;
+
+-- Trigger: aggiorna updated_at ad ogni upsert
+CREATE OR REPLACE FUNCTION _touch_updated_at()
+RETURNS TRIGGER LANGUAGE plpgsql AS $$
+BEGIN NEW.updated_at = now(); RETURN NEW; END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_skill_stats_updated_at ON skill_stats;
+CREATE TRIGGER trg_skill_stats_updated_at
+  BEFORE UPDATE ON skill_stats
+  FOR EACH ROW EXECUTE FUNCTION _touch_updated_at();
+
+
+-- ─── skill_patterns ────────────────────────────────────────────────────────────
+-- Persistenza cross-device per skillPersistence.ts (frontend Dexie → Supabase cloud)
+
+CREATE TABLE IF NOT EXISTS skill_patterns (
+  id             text             PRIMARY KEY,   -- djb2(taskSignature)
+  task_signature text             NOT NULL,
+  tool_sequence  text[]           NOT NULL,
+  success_count  integer          NOT NULL DEFAULT 0,
+  total_count    integer          NOT NULL DEFAULT 0,
+  last_used      bigint           NOT NULL DEFAULT 0,
+  confidence     double precision NOT NULL DEFAULT 0,
+  updated_at     timestamptz      NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_skill_patterns_confidence ON skill_patterns (confidence DESC);
+CREATE INDEX IF NOT EXISTS idx_skill_patterns_last_used  ON skill_patterns (last_used  DESC);
+
+ALTER TABLE skill_patterns ENABLE ROW LEVEL SECURITY;
+
+DO $$ BEGIN
+  IF NOT EXISTS (
+    SELECT 1 FROM pg_policies
+    WHERE tablename = 'skill_patterns' AND policyname = 'anon full access'
+  ) THEN
+    CREATE POLICY "anon full access" ON skill_patterns
+      FOR ALL TO anon
+      USING (true)
+      WITH CHECK (true);
+  END IF;
+END $$;
+
+DROP TRIGGER IF EXISTS trg_skill_patterns_updated_at ON skill_patterns;
+CREATE TRIGGER trg_skill_patterns_updated_at
+  BEFORE UPDATE ON skill_patterns
+  FOR EACH ROW EXECUTE FUNCTION _touch_updated_at();