memory/semantic.py

"""
semantic.py — Semantic Memory
Preferenze, concetti, conoscenza persistente.
Database: Supabase (pgvector cosine similarity). Fallback: ilike + Jaccard.
Backend server: HuggingFace Spaces (FastAPI).

S568: similarity reale via Jaccard word-overlap (ilike path).
S569: pgvector via HF Inference API — cosine similarity reale su embeddings BAAI/bge-small-en-v1.5.
      Attivazione: eseguire migration_pgvector.sql nel SQL Editor Supabase.
      Fallback automatico a ilike+Jaccard se pgvector non ancora attivato.
S570: LRU cache degli embedding (256 entry, TTL 10 min) — evita ricalcolo HF API per query ripetute.
"""
from __future__ import annotations

import hashlib
import logging
import os
import time as _time
import uuid
from collections import OrderedDict
from pathlib import Path

# GAP-NEW-1: usa /data/chroma_db su HF Spaces (volume persistente), locale in dev
# Priorità: 1) env CHROMA_DATA_DIR  2) /data/ se esiste (HF Spaces)  3) cwd (dev)
import os as _os_gap1  # noqa: E402 — import qui per non rompere l'ordine top-level
_CHROMA_DATA_DIR = _os_gap1.getenv("CHROMA_DATA_DIR") or (
    "/data" if Path("/data").exists() else "."
)
CHROMA_PATH     = Path(_CHROMA_DATA_DIR) / "chroma_db"
COLLECTION_NAME = "agente_semantic"
EMBED_MODEL     = "BAAI/bge-small-en-v1.5"  # 384 dims
EMBED_DIMS      = 384
EMBED_MAX_CHARS = 1500  # bge-small max ~512 token ≈ 1500 chars

try:
    import chromadb
    from chromadb.utils import embedding_functions
    CHROMA_AVAILABLE = True
except ImportError:
    CHROMA_AVAILABLE = False

_logger = logging.getLogger("semantic")


# ── S570: LRU cache per embedding ─────────────────────────────────────────────

_EMBED_CACHE_MAX = 256   # entry massime in memoria
_EMBED_CACHE_TTL = 600.0 # secondi di validità (10 minuti)


class _EmbedCache:
    """LRU cache O(1) per embedding vettoriali con TTL.

    Struttura: OrderedDict[sha256_key → (embedding, monotonic_timestamp)].
    - get(): ritorna embedding se presente e non scaduto, else None.
    - set(): inserisce/aggiorna; evicta l'entry più vecchia se piena.
    - Sicuro in single-thread (asyncio è single-threaded per design).
    """

    def __init__(self, max_size: int = _EMBED_CACHE_MAX, ttl: float = _EMBED_CACHE_TTL) -> None:
        self._cache: OrderedDict[str, tuple[list[float], float]] = OrderedDict()
        self._max   = max_size
        self._ttl   = ttl
        self.hits   = 0
        self.misses = 0

    @staticmethod
    def _key(text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]

    def get(self, text: str) -> list[float] | None:
        k = self._key(text)
        if k not in self._cache:
            self.misses += 1
            return None
        emb, ts = self._cache[k]
        if _time.monotonic() - ts > self._ttl:
            del self._cache[k]
            self.misses += 1
            return None
        self._cache.move_to_end(k)  # LRU touch
        self.hits += 1
        return emb

    def set(self, text: str, emb: list[float]) -> None:
        k = self._key(text)
        if k in self._cache:
            self._cache.move_to_end(k)
        elif len(self._cache) >= self._max:
            self._cache.popitem(last=False)  # evicta LRU (il più vecchio)
        self._cache[k] = (emb, _time.monotonic())

    def __len__(self) -> int:
        return len(self._cache)

    def stats(self) -> dict:
        total = self.hits + self.misses
        return {
            "size":      len(self._cache),
            "max_size":  self._max,
            "ttl_s":     self._ttl,
            "hits":      self.hits,
            "misses":    self.misses,
            "hit_rate":  round(self.hits / total, 3) if total else 0.0,
        }


class SemanticMemory:
    def __init__(self, sb_client=None, chroma_client=None):
        self._client     = chroma_client   # chromadb fallback
        self._collection = None
        self._embed_fn   = None
        self._sb         = sb_client     # Supabase client (injected when available)
        self._hf_client  = None          # HuggingFace InferenceClient (lazy)
        self._pgvector   = False         # S569: True quando match_semantic_memory RPC disponibile
        self._embed_cache = _EmbedCache() # S570: LRU 256 entry, TTL 10 min
        self.available   = False

    def _try_supabase(self):
        url = os.getenv("SUPABASE_URL", "")
        key = os.getenv("SUPABASE_ANON_KEY") or os.getenv("SUPABASE_KEY", "")
        if not url or not key:
            return None
        try:
            from supabase import create_client
            return create_client(url, key)
        except Exception:
            return None

    async def init(self):
        if self._sb is None:
            self._sb = self._try_supabase()
        if self._sb:
            try:
                self._sb.table("semantic_memory").select("id").limit(1).execute()
                self.available = True
                # S569: proba se match_semantic_memory RPC esiste (pgvector attivato)
                _zero_emb = "[" + ",".join(["0.0"] * EMBED_DIMS) + "]"
                try:
                    self._sb.rpc("match_semantic_memory", {
                        "query_embedding": _zero_emb,
                        "match_count": 1,
                        "match_threshold": 0.99,
                    }).execute()
                    self._pgvector = True
                    _logger.info("SemanticMemory: Supabase ✓  pgvector ✓ (cosine similarity attiva)")
                except Exception:
                    self._pgvector = False
                    _logger.warning(
                        "SemanticMemory: Supabase pgvector non attivo — ilike+Jaccard. Esegui migration_pgvector.sql per cosine similarity reale."
                    )
                return
            except Exception as e:
                if "PGRST205" in str(e) or "not found" in str(e).lower():
                    _logger.warning(
                        "SemanticMemory: tabella semantic_memory mancante su Supabase. Crea la tabella base e poi esegui migration_pgvector.sql."
                    )
                else:
                    _logger.warning("SemanticMemory: Supabase error: %s", e)
                self._sb = None

        # Fallback ChromaDB locale (HF Spaces senza Supabase)
        if not CHROMA_AVAILABLE:
            _logger.warning("SemanticMemory: disabilitata (chromadb non installato, tabella Supabase mancante).")
            return
        try:
            self._embed_fn   = embedding_functions.SentenceTransformerEmbeddingFunction(
                model_name=EMBED_MODEL
            )
            self._client     = chromadb.PersistentClient(path=str(CHROMA_PATH))
            self._collection = self._client.get_or_create_collection(
                name=COLLECTION_NAME,
                embedding_function=self._embed_fn,
                metadata={"hnsw:space": "cosine"},
            )
            self.available = True
            _logger.info("SemanticMemory: ChromaDB locale (fallback) ✓")
        except Exception as e:
            _logger.warning("SemanticMemory non disponibile: %s", e)

    # ── Embedding ─────────────────────────────────────────────────────────────

    def _embed(self, text: str) -> list[float] | None:
        """S569/S570: Calcola embedding via HF Inference API con LRU cache.

        - S569: usa BAAI/bge-small-en-v1.5 (384 dims) via HuggingFace InferenceClient.
        - S570: controlla _embed_cache prima della chiamata HTTP — cache hit ≈ 0ms vs ~200ms API.
          Cache: 256 entry, TTL 10 min, eviction LRU. Key: sha256(text)[:16].

        Ritorna None se HF_TOKEN mancante o API non raggiungibile → fallback trasparente.
        """
        _txt_key = text[:EMBED_MAX_CHARS]

        # S570: cache hit → ritorna senza chiamata HF (≈0ms)
        cached = self._embed_cache.get(_txt_key)
        if cached is not None:
            return cached

        # Cache miss → chiama HF Inference API
        try:
            from huggingface_hub import InferenceClient
            if self._hf_client is None:
                token = os.getenv("HF_TOKEN", "")
                if not token:
                    if not getattr(self, '_hf_token_warned', False):
                        _logger.warning("SemanticMemory: HF_TOKEN mancante — embedding HF disabilitato, fallback a ilike search")
                        self._hf_token_warned = True
                    return None
                self._hf_client = InferenceClient(token=token)
            # S750-GAP-C: timeout 10s su HF Inference API (sync bloccante).
            # ThreadPoolExecutor.submit().result(timeout) solleva concurrent.futures.TimeoutError
            # se la chiamata supera 10s — evita hang infinito del worker asyncio.
            import concurrent.futures as _cf
            with _cf.ThreadPoolExecutor(max_workers=1) as _pool:
                _fut = _pool.submit(
                    self._hf_client.feature_extraction,
                    text=_txt_key,
                    model=EMBED_MODEL,
                )
                try:
                    result = _fut.result(timeout=10.0)
                except _cf.TimeoutError:
                    return None
            # huggingface_hub ritorna np.ndarray shape (384,) o (1, 384)
            if hasattr(result, "tolist"):
                flat = result.tolist()
            else:
                flat = list(result)
            # Normalizza shape (1, 384) → (384,)
            if flat and isinstance(flat[0], list):
                flat = flat[0]
            if len(flat) != EMBED_DIMS:
                return None
            # S570: salva in cache prima di ritornare
            self._embed_cache.set(_txt_key, flat)
            return flat
        except Exception:
            return None

    @staticmethod
    def _emb_to_pg(emb: list[float]) -> str:
        """Serializza embedding a stringa vettoriale per PostgREST vector type: '[x,y,...]'."""
        return "[" + ",".join(f"{x:.8f}" for x in emb) + "]"

    # ── Write ─────────────────────────────────────────────────────────────────

    def add(self, text: str, metadata: dict | None = None, doc_id: str | None = None):
        _id  = doc_id or str(uuid.uuid4())
        _txt = text[:4000]  # S568: era 2000, aumentato per output codice
        if self._sb:
            try:
                row: dict = {"id": _id, "content": _txt, "metadata": metadata or {}}
                if self._pgvector:
                    # S569: calcola embedding e salva nel campo vector — abilita cosine search
                    emb = self._embed(_txt)
                    if emb is not None:
                        row["embedding"] = self._emb_to_pg(emb)
                self._sb.table("semantic_memory").upsert(row).execute()
                return
            except Exception as _e:
                _logger.warning("semantic.add: Supabase upsert failed: %s", _e)
        if self._collection:
            self._collection.upsert(
                documents=[text],
                metadatas=[metadata or {}],
                ids=[_id],
            )

    # ── Read ──────────────────────────────────────────────────────────────────

    @staticmethod
    def _word_overlap_similarity(query: str, content: str) -> float:
        """S568: stima similarity via word-overlap (Jaccard) tra query e content.
        Usato come fallback quando pgvector non è attivo.
        Range reale: 0.05-0.95 (mai 0.0 — ilike garantisce match parziale).
        """
        q_words = set(query.lower().split())
        c_words = set(content.lower().split())
        if not q_words or not c_words:
            return 0.5
        intersection = len(q_words & c_words)
        union        = len(q_words | c_words)
        return round(intersection / union, 4) if union else 0.5

    def search(self, query: str, n_results: int = 5) -> list[dict]:
        if self._sb:
            # S569: path pgvector — cosine similarity reale via RPC
            if self._pgvector:
                emb = self._embed(query)
                if emb is not None:
                    try:
                        rows = (
                            self._sb.rpc("match_semantic_memory", {
                                "query_embedding": self._emb_to_pg(emb),
                                "match_count":     n_results,
                                "match_threshold": 0.2,  # soglia coseno: 0.2 = rilevante
                            }).execute().data or []
                        )
                        return [
                            {
                                "content":    r["content"],
                                "metadata":   r.get("metadata", {}),
                                "similarity": round(float(r.get("similarity", 0)), 4),
                            }
                            for r in rows
                        ]
                    except Exception:
                        pass  # RPC fallita → fallback ilike sotto
            # Fallback ilike + Jaccard (pgvector non attivo o embedding non disponibile)
            try:
                rows = (
                    self._sb.table("semantic_memory")
                    .select("*").limit(1000)  # BUGFIX: LIMIT 1000 — senza limit OOM su memoria semantica grande
                    .ilike("content", f"%{query}%")
                    .limit(n_results)
                    .execute().data or []
                )
                return [
                    {
                        "content":    r["content"],
                        "metadata":   r.get("metadata", {}),
                        "similarity": self._word_overlap_similarity(query, r["content"]),
                    }
                    for r in rows
                ]
            except Exception as _exc:
                _logger.debug("[semantic] silenced %s", type(_exc).__name__)  # noqa: BLE001
        if not self._collection:
            return []
        try:
            results = self._collection.query(
                query_texts=[query],
                n_results=min(n_results, self._collection.count() or 1),
            )
            docs  = results.get("documents", [[]])[0]
            metas = results.get("metadatas",  [[]])[0]
            dists = results.get("distances",  [[]])[0]
            return [
                {"content": doc, "metadata": meta, "similarity": round(1 - dist, 4)}
                for doc, meta, dist in zip(docs, metas, dists)
            ]
        except Exception:
            return []

    def count(self) -> int:
        if self._sb:
            try:
                return self._sb.table("semantic_memory").select("id", count="exact").execute().count or 0
            except Exception as _exc:
                _logger.debug("[semantic] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return self._collection.count() if self._collection else 0

    def stats(self) -> dict:
        backend = "supabase+pgvector" if (self._sb and self._pgvector) else \
                  ("supabase"         if self._sb  else
                  ("chroma"           if self._collection else "none"))
        return {
            "available":    self.available,
            "documents":    self.count(),
            "model":        EMBED_MODEL,
            "backend":      backend,
            "pgvector":     self._pgvector,
            "embed_cache":  self._embed_cache.stats(),  # S570: hit_rate, size, TTL
        }

    # ── Export / Import ───────────────────────────────────────────────────────

    def export_all(self, limit: int = 2000) -> list[dict]:
        """Esporta tutti i record della semantic memory come lista di dict portabile.

        Funziona su entrambi i backend (Supabase e ChromaDB).
        Usato da GET /api/memory/sync/export per snapshot cross-sessione.
        """
        if self._sb:
            try:
                rows = (
                    self._sb.table("semantic_memory")
                    .select("id, content, metadata")
                    .limit(limit)
                    .execute().data or []
                )
                return [
                    {"id": r["id"], "content": r["content"], "metadata": r.get("metadata") or {}}
                    for r in rows
                ]
            except Exception as exc:
                _logger.warning("SemanticMemory.export_all (supabase) error: %s", exc)
                return []
        if self._collection:
            try:
                n   = self._collection.count()
                res = self._collection.get(limit=min(n, limit)) if n else {"ids": [], "documents": [], "metadatas": []}
                ids   = res.get("ids", [])
                docs  = res.get("documents", [])
                metas = res.get("metadatas", [])
                return [
                    {"id": i, "content": d, "metadata": m or {}}
                    for i, d, m in zip(ids, docs, metas)
                ]
            except Exception as exc:
                _logger.warning("SemanticMemory.export_all (chroma) error: %s", exc)
                return []
        return []

    def import_all(self, records: list[dict], overwrite: bool = False) -> dict:
        """Importa una lista di record nella semantic memory (upsert idempotente).

        Ogni record: {"id": str, "content": str, "metadata": dict}.
        Se overwrite=False (default) usa upsert — record esistenti vengono aggiornati.
        Usato da POST /api/memory/sync/import per restore cross-sessione.
        """
        import json as _json
        imported = 0
        skipped  = 0
        for r in records:
            content = str(r.get("content", "")).strip()
            if not content:
                skipped += 1
                continue
            doc_id   = r.get("id") or None
            metadata = r.get("metadata", {})
            if isinstance(metadata, str):
                try:
                    metadata = _json.loads(metadata)
                except Exception:
                    metadata = {}
            try:
                self.add(content, metadata or {}, doc_id=doc_id)
                imported += 1
            except Exception:
                skipped += 1
        return {"imported": imported, "skipped": skipped, "total": len(records)}