""" episodic.py — Episodic Memory Task completati, errori, fix — persiste su Supabase se disponibile, altrimenti SQLite locale. Backend server: HuggingFace Spaces (FastAPI). Database: Supabase. """ from __future__ import annotations import json, os, sqlite3 from datetime import datetime from pathlib import Path from dataclasses import dataclass import logging _logger = logging.getLogger("memory.episodic") # D1: usa /data/ se disponibile (HF Spaces volume persistente) — evita perdita a restart _EPIS_DATA_DIR = os.getenv('CHROMA_DATA_DIR') or ('/data' if Path('/data').exists() else '.') DB_PATH = Path(_EPIS_DATA_DIR) / 'episodic_memory.db' @dataclass class Episode: id: int type: str task: str output: str success: bool ts: str tags: list[str] def _score_episodes(episodes: list, query: str, n: int) -> list: """D2: word-overlap scoring analogo a ReflectionMemory.get_relevant_lessons(). Riordina i risultati ilike per rilevanza semantica (zero LLM, zero latenza). """ words = {w.lower() for w in query.split() if len(w) > 3} def _ep_score(ep) -> int: text = (getattr(ep, 'task', '') + ' ' + (getattr(ep, 'output', '') or '')).lower() return sum(1 for w in words if w in text) scored = sorted(episodes, key=_ep_score, reverse=True) return scored[:n] class EpisodicMemory: def __init__(self): self._db: sqlite3.Connection | None = None self._sb = None 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 as e: _logger.warning("EpisodicMemory: Supabase non disponibile: %s", e) return None def init(self): self._sb = self._try_supabase() if self._sb: try: self._sb.table("episodes").select("id").limit(1).execute() _logger.info("EpisodicMemory: Supabase ✓") return except Exception as e: _logger.warning("EpisodicMemory: Supabase table check fallita: %s", e) self._sb = None # Fallback SQLite locale (dev / HF Spaces senza Supabase configurato) self._db = sqlite3.connect(DB_PATH, check_same_thread=False) self._db.execute(""" CREATE TABLE IF NOT EXISTS episodes ( id INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT NOT NULL, task TEXT NOT NULL, output TEXT, success INTEGER DEFAULT 1, ts TEXT NOT NULL, tags TEXT DEFAULT '[]' ) """) self._db.execute("CREATE INDEX IF NOT EXISTS idx_type ON episodes(type)") self._db.execute("CREATE INDEX IF NOT EXISTS idx_ts ON episodes(ts)") self._db.commit() _logger.info("EpisodicMemory: SQLite locale (fallback) ✓") # ── Write ───────────────────────────────────────────────────────────────── def add(self, type_: str, task: str, output: str, success: bool, tags: list[str] | None = None) -> int: ts = datetime.now().isoformat() row = { "type": type_, "task": task[:500], "output": output[:2000], "success": success, "ts": ts, "tags": json.dumps(tags or []), } if self._sb: try: res = self._sb.table("episodes").insert(row).execute() return res.data[0].get("id", -1) if res.data else -1 except Exception as _exc: _logger.debug("[episodic] silenced %s", type(_exc).__name__) # noqa: BLE001 if not self._db: return -1 cur = self._db.execute( "INSERT INTO episodes (type,task,output,success,ts,tags) VALUES (?,?,?,?,?,?)", (type_, task[:500], output[:2000], int(success), ts, json.dumps(tags or [])) ) self._db.commit() return cur.lastrowid # ── Read ────────────────────────────────────────────────────────────────── def get_recent(self, n: int = 20, type_: str | None = None) -> list[Episode]: if self._sb: try: q = self._sb.table("episodes").select("*").order("id", desc=True).limit(n) if type_: q = q.eq("type", type_) rows = q.execute().data or [] return [self._from_sb(r) for r in rows] except Exception as _exc: _logger.debug("[episodic] silenced %s", type(_exc).__name__) # noqa: BLE001 if not self._db: return [] if type_: rows = self._db.execute( "SELECT * FROM episodes WHERE type=? ORDER BY id DESC LIMIT ?", (type_, n) ).fetchall() else: rows = self._db.execute( "SELECT * FROM episodes ORDER BY id DESC LIMIT ?", (n,) ).fetchall() return [Episode(r[0], r[1], r[2], r[3], bool(r[4]), r[5], json.loads(r[6])) for r in rows] def search_text(self, query: str, n: int = 5) -> list[Episode]: if self._sb: try: # S571-GAP4: due query indipendenti possono restituire lo stesso episode # (quando task e output matchano entrambi). Dedup by id prima del return. task_rows = self._sb.table("episodes").select("*").ilike("task", f"%{query}%").limit(n).execute().data or [] output_rows = self._sb.table("episodes").select("*").ilike("output", f"%{query}%").limit(n).execute().data or [] seen: set[int] = set() merged: list[dict] = [] for r in task_rows + output_rows: rid = r.get("id") if rid not in seen: seen.add(rid) merged.append(r) return _score_episodes([self._from_sb(r) for r in merged[:n]], query, n) # S571-GAP4: limit except Exception as _exc: _logger.debug("[episodic] silenced %s", type(_exc).__name__) # noqa: BLE001 if not self._db: return [] rows = self._db.execute( "SELECT * FROM episodes WHERE task LIKE ? OR output LIKE ? ORDER BY id DESC LIMIT ?", (f"%{query}%", f"%{query}%", n) ).fetchall() raw = [Episode(r[0], r[1], r[2], r[3], bool(r[4]), r[5], json.loads(r[6])) for r in rows] return _score_episodes(raw, query, n) def get_errors(self, n: int = 10) -> list[Episode]: return self.get_recent(n, type_="error") def stats(self) -> dict: if self._sb: try: res = self._sb.table("episodes").select("id", count="exact").execute() return {"total": res.count or 0, "backend": "supabase"} except Exception as _exc: _logger.debug("[episodic] silenced %s", type(_exc).__name__) # noqa: BLE001 if not self._db: return {"total": 0, "backend": "none"} total = self._db.execute("SELECT COUNT(*) FROM episodes").fetchone()[0] by_type = dict(self._db.execute("SELECT type, COUNT(*) FROM episodes GROUP BY type").fetchall()) return {"total": total, "by_type": by_type, "backend": "sqlite"} @staticmethod def _from_sb(r: dict) -> Episode: return Episode( r["id"], r["type"], r["task"], r.get("output", ""), bool(r["success"]), r["ts"], json.loads(r.get("tags", "[]")) ) def close(self): if self._db: self._db.close()