hermes-agent/agent/memory.py

"""Memory backends — cross-session user modeling.

Local SQLite (sovereign, A grade 95pts) vs Honcho cloud (B grade 60pts).
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__)

@dataclass
class Entry:
    key: str; value: str; uid: str
    etype: str = "preference"; 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

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

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

class Local(Backend):
    def __init__(self, p: Path = None):
        self._p = p or get_hermes_home() / "memory.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 m(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 m 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: %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 m WHERE uid=? AND k=?", (uid, k)).fetchone()
            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]) if r else None
        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 m 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 m 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 m 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

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: 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: pass
            return []
        except: return []
    def all(self, uid) -> List[Entry]: return self.find(uid, "", 100)
    def rm(self, uid, k) -> bool: return False
    @property
    def name(self) -> str: return "honcho"
    @property
    def cloud(self) -> bool: return True

def score(b: Backend, uid: str = "_e_") -> Dict:
    if not b.ok(): return {"name": b.name, "score": 0, "grade": "F", "ok": False, "cloud": b.cloud}
    s = 20
    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
    avg = (sm+gm+qm)/3; s += 20 if avg < 10 else 15 if avg < 50 else 10 if avg < 200 else 5
    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}

def evaluate() -> Dict:
    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"
    if best and best.get("cloud"): rec += " WARNING: cloud. RECOMMEND local."
    return {"results": rs, "recommendation": rec}

_inst = None
def get_backend() -> Backend:
    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