memory/episodic.py

"""
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()