splitbit_llm/memory/fast_cache.py

"""Fast Reply Cache — near-instant responses using skill and memory cache.

When the LLM has accumulated enough skills and conversation patterns,
responses can be served from cache instead of running full inference.

Three cache layers:
1. Exact match cache — same question asked before → return cached answer
2. Skill-based cache — high-confidence skill match → return skill content
3. Semantic cache — similar question asked before → return best cached answer

Cache hits skip LLM inference entirely → near-instant response (< 1ms).
Cache misses fall through to normal LLM inference.

The cache "warms up" over time — as more conversations happen and more
skills are created, the cache hit rate increases. Eventually most common
questions get instant responses.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import sqlite3
import time
from dataclasses import dataclass, field
from typing import Any

logger = logging.getLogger(__name__)


@dataclass
class CacheEntry:
    """A cached response."""
    id: str
    query_hash: str
    query_text: str
    response_text: str
    channel: str = "cli"
    skill_id: str = ""
    confidence: float = 0.0
    hit_count: int = 0
    created_at: float = field(default_factory=time.time)
    last_accessed: float = field(default_factory=time.time)
    response_time_s: float = 0.0  # original response time (for stats)


class FastReplyCache:
    """Multi-layer cache for near-instant responses.

    Layers:
    1. Exact match — hash of normalized query → cached response
    2. Skill match — high-confidence skill → skill content as response
    3. Semantic match — keyword overlap with past queries → best response

    SQLite-backed — survives restarts. Gets faster over time.
    """

    EXACT_CONFIDENCE_THRESHOLD = 0.95
    SKILL_CONFIDENCE_THRESHOLD = 0.85
    SEMANTIC_CONFIDENCE_THRESHOLD = 0.75
    MAX_CACHE_SIZE = 10000
    STALE_ENTRY_DAYS = 30

    def __init__(self, db_path: str) -> None:
        self.db_path = db_path
        os.makedirs(os.path.dirname(db_path) or ".", exist_ok=True)
        self._init_db()
        self._stats = {
            "exact_hits": 0,
            "skill_hits": 0,
            "semantic_hits": 0,
            "misses": 0,
            "total_lookups": 0,
            "time_saved_s": 0.0,
            "entries_stored": 0,
        }
        self._load_stats()

    def _init_db(self) -> None:
        with sqlite3.connect(self.db_path) as conn:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS reply_cache (
                    id TEXT PRIMARY KEY,
                    query_hash TEXT UNIQUE,
                    query_text TEXT,
                    response_text TEXT,
                    channel TEXT DEFAULT 'cli',
                    skill_id TEXT DEFAULT '',
                    confidence REAL DEFAULT 0,
                    hit_count INTEGER DEFAULT 0,
                    created_at REAL,
                    last_accessed REAL,
                    response_time_s REAL DEFAULT 0
                );
                CREATE INDEX IF NOT EXISTS idx_cache_hash ON reply_cache(query_hash);
                CREATE INDEX IF NOT EXISTS idx_cache_confidence ON reply_cache(confidence);
                CREATE INDEX IF NOT EXISTS idx_cache_hits ON reply_cache(hit_count);
            """)

    def _load_stats(self) -> None:
        with sqlite3.connect(self.db_path) as conn:
            self._stats["entries_stored"] = conn.execute("SELECT COUNT(*) FROM reply_cache").fetchone()[0]

    @staticmethod
    def _normalize_query(query: str) -> str:
        """Normalize a query for hashing — lowercase, strip whitespace, remove punctuation."""
        import re
        normalized = re.sub(r'[^\w\s]', '', query.lower().strip())
        normalized = ' '.join(normalized.split())
        return normalized

    @staticmethod
    def _hash_query(query: str) -> str:
        """Hash a normalized query."""
        normalized = FastReplyCache._normalize_query(query)
        return hashlib.sha256(normalized.encode()).hexdigest()[:16]

    def lookup(self, query: str, channel: str = "cli",
               skills: list = None) -> dict[str, Any] | None:
        """Look up a query in the cache.

        Returns cached response dict if hit, None if miss.
        Sets "cache_hit": True, "cache_type": "exact"|"skill"|"semantic"
        """
        self._stats["total_lookups"] += 1
        query_hash = self._hash_query(query)

        # Layer 1: Exact match
        result = self._lookup_exact(query_hash, channel)
        if result:
            self._stats["exact_hits"] += 1
            self._stats["time_saved_s"] += result.get("response_time_s", 0.1)
            logger.debug("Cache EXACT hit: %s", query[:40])
            return {**result, "cache_hit": True, "cache_type": "exact"}

        # Layer 2: Skill-based match
        if skills:
            result = self._lookup_skill(query, skills, channel)
            if result:
                self._stats["skill_hits"] += 1
                logger.debug("Cache SKILL hit: %s", query[:40])
                return {**result, "cache_hit": True, "cache_type": "skill"}

        # Layer 3: Semantic match (keyword overlap)
        result = self._lookup_semantic(query, channel)
        if result:
            self._stats["semantic_hits"] += 1
            self._stats["time_saved_s"] += result.get("response_time_s", 0.1) * 0.5
            logger.debug("Cache SEMANTIC hit: %s", query[:40])
            return {**result, "cache_hit": True, "cache_type": "semantic"}

        self._stats["misses"] += 1
        return None

    def _lookup_exact(self, query_hash: str, channel: str) -> dict[str, Any] | None:
        """Look up by exact query hash."""
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT * FROM reply_cache WHERE query_hash = ? AND confidence >= ?",
                (query_hash, self.EXACT_CONFIDENCE_THRESHOLD)
            ).fetchone()

        if row:
            self._increment_hits(row[0])
            return {
                "response": row[3],  # response_text
                "confidence": row[6],
                "hit_count": row[7] + 1,
                "response_time_s": row[10],
            }
        return None

    def _lookup_skill(self, query: str, skills: list, channel: str) -> dict[str, Any] | None:
        """Look up using skill matches."""
        for skill in skills:
            if hasattr(skill, 'confidence') and skill.confidence >= self.SKILL_CONFIDENCE_THRESHOLD:
                if hasattr(skill, 'effectiveness_score') and skill.effectiveness_score >= 0.8:
                    # Check if skill triggers match the query
                    query_lower = query.lower()
                    matched_triggers = sum(1 for t in skill.trigger_conditions if t.lower() in query_lower)
                    if matched_triggers >= 2 or (matched_triggers >= 1 and skill.confidence >= 0.9):
                        return {
                            "response": skill.content[:500],
                            "confidence": skill.confidence * skill.effectiveness_score,
                            "skill_id": skill.id,
                            "response_time_s": 0.001,
                        }
        return None

    def _lookup_semantic(self, query: str, channel: str) -> dict[str, Any] | None:
        """Look up by semantic similarity (keyword overlap)."""
        query_words = set(self._normalize_query(query).split())
        if not query_words:
            return None

        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                "SELECT * FROM reply_cache WHERE confidence >= ? AND hit_count > 0 "
                "ORDER BY hit_count DESC, confidence DESC LIMIT 20",
                (self.SEMANTIC_CONFIDENCE_THRESHOLD,)
            ).fetchall()

        best_match = None
        best_score = 0.0

        for row in rows:
            cached_words = set(self._normalize_query(row[2]).split())  # row[2] = query_text
            if not cached_words:
                continue
            overlap = len(query_words & cached_words)
            score = overlap / max(len(query_words), len(cached_words))
            # Boost score by confidence and hit count
            score = score * 0.5 + row[6] * 0.3 + min(row[7] / 10.0, 1.0) * 0.2  # confidence, hits

            if score > best_score and score >= self.SEMANTIC_CONFIDENCE_THRESHOLD:
                best_score = score
                best_match = row

        if best_match:
            self._increment_hits(best_match[0])
            return {
                "response": best_match[3],
                "confidence": best_score,
                "hit_count": best_match[7] + 1,
                "response_time_s": best_match[10],
            }
        return None

    def store(self, query: str, response: str, channel: str = "cli",
              skill_id: str = "", confidence: float = 0.8,
              response_time_s: float = 0.1) -> str:
        """Store a query-response pair in the cache."""
        query_hash = self._hash_query(query)
        entry_id = hashlib.sha256(f"{query_hash}:{time.time()}".encode()).hexdigest()[:16]

        with sqlite3.connect(self.db_path) as conn:
            # Check if exact hash already exists — update it
            existing = conn.execute(
                "SELECT id, hit_count FROM reply_cache WHERE query_hash = ?", (query_hash,)
            ).fetchone()

            if existing:
                # Update existing entry with better response if confidence is higher
                conn.execute(
                    "UPDATE reply_cache SET response_text = ?, confidence = ?, "
                    "response_time_s = ?, last_accessed = ? WHERE query_hash = ?",
                    (response, confidence, response_time_s, time.time(), query_hash)
                )
                return existing[0]

            conn.execute(
                """INSERT OR REPLACE INTO reply_cache
                   (id, query_hash, query_text, response_text, channel, skill_id,
                    confidence, hit_count, created_at, last_accessed, response_time_s)
                   VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
                (entry_id, query_hash, query, response, channel, skill_id,
                 confidence, 0, time.time(), time.time(), response_time_s)
            )

        self._stats["entries_stored"] += 1

        # Prune if cache is too large
        if self._stats["entries_stored"] > self.MAX_CACHE_SIZE:
            self._prune()

        return entry_id

    def _increment_hits(self, entry_id: str) -> None:
        """Increment hit count for a cache entry."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE reply_cache SET hit_count = hit_count + 1, last_accessed = ? WHERE id = ?",
                (time.time(), entry_id)
            )

    def _prune(self) -> None:
        """Remove stale and low-value entries."""
        cutoff = time.time() - (self.STALE_ENTRY_DAYS * 86400)
        with sqlite3.connect(self.db_path) as conn:
            # Remove old, unused entries
            conn.execute(
                "DELETE FROM reply_cache WHERE last_accessed < ? AND hit_count < 2",
                (cutoff,)
            )
            # Remove lowest-confidence entries if still too many
            count = conn.execute("SELECT COUNT(*) FROM reply_cache").fetchone()[0]
            if count > self.MAX_CACHE_SIZE:
                conn.execute(
                    "DELETE FROM reply_cache WHERE id IN "
                    "(SELECT id FROM reply_cache ORDER BY confidence ASC, hit_count ASC LIMIT ?)",
                    (count - self.MAX_CACHE_SIZE,)
                )
            self._stats["entries_stored"] = conn.execute("SELECT COUNT(*) FROM reply_cache").fetchone()[0]

    def get_hit_rate(self) -> float:
        """Get cache hit rate (0-1)."""
        total = self._stats["total_lookups"]
        if total == 0:
            return 0.0
        hits = self._stats["exact_hits"] + self._stats["skill_hits"] + self._stats["semantic_hits"]
        return hits / total

    def is_fast_ready(self) -> bool:
        """Check if the cache is warm enough for near-instant responses."""
        return self._stats["entries_stored"] >= 50 and self.get_hit_rate() >= 0.3

    def get_stats(self) -> dict[str, Any]:
        return {
            **self._stats,
            "hit_rate": round(self.get_hit_rate(), 3),
            "fast_ready": self.is_fast_ready(),
            "avg_time_saved_s": round(
                self._stats["time_saved_s"] / max(self._stats["total_lookups"], 1), 4
            ),
        }