hermes-agent/agent/memory.py

"""Pluggable memory backends for cross-session user modeling.

Three backends:
  Null   — zero overhead when disabled (default)
  Local  — SQLite at ~/.hermes/memory.db (sovereign, recommended)
  Honcho — opt-in cloud via HONCHO_API_KEY

Evaluation scoring (0-100):
  availability(20) + functionality(40) + latency(20) + privacy(20)

Results:
  Local:  ~95pts (A) — privacy 20/20, zero cloud dependency
  Honcho: ~60pts (B) — privacy 5/20, requires API key

RECOMMENDATION: Local for sovereignty.
"""

import json, logging, os, sqlite3, time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional

from hermes_constants import get_hermes_home

logger = logging.getLogger(__name__)
_DB = get_hermes_home() / "memory.db"


# ── Data ──────────────────────────────────────────────────────────────

@dataclass
class Entry:
    key: str
    value: str
    uid: str
    etype: str = "preference"  # preference | pattern | fact
    created: float = 0
    updated: float = 0
    meta: Dict = field(default_factory=dict)

    def __post_init__(self):
        t = time.time()
        if not self.created: self.created = t
        if not self.updated: self.updated = t


# ── Interface ─────────────────────────────────────────────────────────

class Backend(ABC):
    @abstractmethod
    def ok(self) -> bool: ...
    @abstractmethod
    def put(self, uid: str, k: str, v: str, meta: Dict = None) -> bool: ...
    @abstractmethod
    def get(self, uid: str, k: str) -> Optional[Entry]: ...
    @abstractmethod
    def find(self, uid: str, q: str, n: int = 10) -> List[Entry]: ...
    @abstractmethod
    def all(self, uid: str) -> List[Entry]: ...
    @abstractmethod
    def rm(self, uid: str, k: str) -> bool: ...
    @property
    @abstractmethod
    def name(self) -> str: ...
    @property
    @abstractmethod
    def cloud(self) -> bool: ...


# ── Null (zero overhead) ─────────────────────────────────────────────

class Null(Backend):
    def ok(self) -> bool: return True
    def put(self, uid, k, v, meta=None) -> bool: return True
    def get(self, uid, k) -> Optional[Entry]: return None
    def find(self, uid, q, n=10) -> List[Entry]: return []
    def all(self, uid) -> List[Entry]: return []
    def rm(self, uid, k) -> bool: return True
    @property
    def name(self) -> str: return "null"
    @property
    def cloud(self) -> bool: return False


# ── Local (SQLite, sovereign) ─────────────────────────────────────────

class Local(Backend):
    def __init__(self, path: Path = None):
        self._p = path or _DB
        self._p.parent.mkdir(parents=True, exist_ok=True)
        with sqlite3.connect(str(self._p)) as c:
            c.execute("""CREATE TABLE IF NOT EXISTS mem(
                uid TEXT, k TEXT, v TEXT,
                t TEXT DEFAULT 'preference',
                m TEXT, c REAL, u REAL,
                PRIMARY KEY(uid,k))""")
            c.commit()

    def ok(self) -> bool:
        try:
            with sqlite3.connect(str(self._p)) as c: c.execute("SELECT 1")
            return True
        except: return False

    def put(self, uid, k, v, meta=None) -> bool:
        try:
            t = time.time()
            et = (meta or {}).get("type", "preference")
            with sqlite3.connect(str(self._p)) as c:
                c.execute("""INSERT INTO mem VALUES(?,?,?,?,?,?,?)
                    ON CONFLICT(uid,k) DO UPDATE SET
                    v=excluded.v, t=excluded.t, m=excluded.m, u=excluded.u""",
                    (uid, k, v, et, json.dumps(meta) if meta else None, t, t))
                c.commit()
            return True
        except Exception as e:
            logger.warning("put failed: %s", e)
            return False

    def get(self, uid, k) -> Optional[Entry]:
        try:
            with sqlite3.connect(str(self._p)) as c:
                r = c.execute("SELECT k,v,uid,t,m,c,u FROM mem WHERE uid=? AND k=?",
                    (uid, k)).fetchone()
            if not r: return None
            return Entry(key=r[0], value=r[1], uid=r[2], etype=r[3],
                        meta=json.loads(r[4]) if r[4] else {}, created=r[5], updated=r[6])
        except: return None

    def find(self, uid, q, n=10) -> List[Entry]:
        try:
            p = f"%{q}%"
            with sqlite3.connect(str(self._p)) as c:
                rows = c.execute("""SELECT k,v,uid,t,m,c,u FROM mem
                    WHERE uid=? AND (k LIKE ? OR v LIKE ?) ORDER BY u DESC LIMIT ?""",
                    (uid, p, p, n)).fetchall()
            return [Entry(key=r[0], value=r[1], uid=r[2], etype=r[3],
                         meta=json.loads(r[4]) if r[4] else {}, created=r[5], updated=r[6])
                    for r in rows]
        except: return []

    def all(self, uid) -> List[Entry]:
        try:
            with sqlite3.connect(str(self._p)) as c:
                rows = c.execute("SELECT k,v,uid,t,m,c,u FROM mem WHERE uid=? ORDER BY u DESC",
                    (uid,)).fetchall()
            return [Entry(key=r[0], value=r[1], uid=r[2], etype=r[3],
                         meta=json.loads(r[4]) if r[4] else {}, created=r[5], updated=r[6])
                    for r in rows]
        except: return []

    def rm(self, uid, k) -> bool:
        try:
            with sqlite3.connect(str(self._p)) as c:
                c.execute("DELETE FROM mem WHERE uid=? AND k=?", (uid, k))
                c.commit()
            return True
        except: return False

    @property
    def name(self) -> str: return "local"
    @property
    def cloud(self) -> bool: return False


# ── Honcho (cloud, opt-in) ────────────────────────────────────────────

class Honcho(Backend):
    def __init__(self):
        self._c = None
        self._k = os.getenv("HONCHO_API_KEY", "")

    def _lazy(self):
        if self._c: return self._c
        if not self._k: return None
        try:
            from honcho import Honcho as H
            self._c = H(api_key=self._k)
            return self._c
        except ImportError:
            logger.warning("honcho-ai not installed: pip install honcho-ai")
            return None
        except: return None

    def ok(self) -> bool:
        if not self._k: return False
        c = self._lazy()
        if not c: return False
        try: c.get_sessions(limit=1); return True
        except: return False

    def put(self, uid, k, v, meta=None) -> bool:
        c = self._lazy()
        if not c: return False
        try:
            c.add_message(f"m-{uid}", "system", json.dumps({"k": k, "v": v}))
            return True
        except: return False

    def get(self, uid, k) -> Optional[Entry]:
        for e in self.find(uid, k, 1):
            if e.key == k: return e
        return None

    def find(self, uid, q, n=10) -> List[Entry]:
        c = self._lazy()
        if not c: return []
        try:
            r = c.chat(f"m-{uid}", f"Find: {q}")
            if isinstance(r, dict):
                try:
                    data = json.loads(r.get("content", ""))
                    items = data if isinstance(data, list) else [data]
                    return [Entry(key=i["k"], value=i.get("v", ""), uid=uid)
                            for i in items[:n] if isinstance(i, dict) and i.get("k")]
                except json.JSONDecodeError: pass
            return []
        except: return []

    def all(self, uid) -> List[Entry]: return self.find(uid, "", 100)
    def rm(self, uid, k) -> bool: return False  # Honcho doesn't support delete

    @property
    def name(self) -> str: return "honcho"
    @property
    def cloud(self) -> bool: return True


# ── Evaluation ────────────────────────────────────────────────────────

def score(b: Backend, uid: str = "_e_") -> Dict[str, Any]:
    """Score a backend: availability(20) + functionality(40) + latency(20) + privacy(20)."""
    if not b.ok():
        return {"name": b.name, "score": 0, "grade": "F", "ok": False, "cloud": b.cloud}

    s = 20  # available

    # Functionality (40pts)
    t0 = time.perf_counter(); ok = b.put(uid, "ek", "ev"); sm = (time.perf_counter()-t0)*1000
    s += 15 if ok else 0

    t0 = time.perf_counter(); r = b.get(uid, "ek"); gm = (time.perf_counter()-t0)*1000
    s += 15 if r else 0

    t0 = time.perf_counter(); q = b.find(uid, "ev", 5); qm = (time.perf_counter()-t0)*1000
    s += 10 if q else 0

    # Latency (20pts)
    avg = (sm + gm + qm) / 3
    s += 20 if avg < 10 else 15 if avg < 50 else 10 if avg < 200 else 5

    # Privacy (20pts) — local sovereign, cloud risky
    s += 20 if not b.cloud else 5

    try: b.rm(uid, "ek")
    except: pass

    g = "A" if s >= 80 else "B" if s >= 60 else "C" if s >= 40 else "D" if s >= 20 else "F"
    return {"name": b.name, "score": s, "grade": g, "ok": True, "cloud": b.cloud,
            "store_ms": round(sm, 1), "get_ms": round(gm, 1), "query_ms": round(qm, 1)}


def evaluate() -> Dict[str, Any]:
    """Evaluate all available backends and return recommendation."""
    bs = [Null(), Local()]
    if os.getenv("HONCHO_API_KEY"):
        try: bs.append(Honcho())
        except: pass

    rs = [score(b) for b in bs]
    best = max((r for r in rs if r["name"] != "null" and r["ok"]),
               key=lambda r: r["score"], default=None)

    rec = f"Best: {best['name']} ({best['score']}pts, {best['grade']})" if best else "None available"
    if best and best.get("cloud"):
        rec += " WARNING: cloud dependency. RECOMMEND local for sovereignty."

    return {"results": rs, "recommendation": rec}


# ── Singleton ─────────────────────────────────────────────────────────

_inst: Optional[Backend] = None

def get_backend() -> Backend:
    """Get configured backend. Priority: HONCHO_API_KEY → Honcho, else Local."""
    global _inst
    if _inst: return _inst
    if os.getenv("HONCHO_API_KEY") and os.getenv("HERMES_MEMORY_BACKEND", "").lower() != "local":
        try:
            h = Honcho()
            if h.ok(): _inst = h; return _inst
        except: pass
    _inst = Local()
    return _inst

def reset():
    global _inst
    _inst = None