diff --git a/plugins/memory/mempalace/__init__.py b/plugins/memory/mempalace/__init__.py new file mode 100644 index 000000000000..dc08822247ae --- /dev/null +++ b/plugins/memory/mempalace/__init__.py @@ -0,0 +1,1098 @@ +"""MemPalace memory provider plugin for Hermes. + +Local, API-key-free memory using ChromaDB + AAAK layered recall. +Wings auto-route exchanges to project-scoped memory stores. +L0+L1 wake-up context injects identity + essential story every session. + +Config: + Palace path: ~/.mempalace/palace (or override via $HERMES_HOME/mempalace.json) + Wing config: ~/.mempalace/wing_config.json (generated by `mempalace init`) + Identity: ~/.mempalace/identity.txt (generated by `mempalace init`) + +Run `mempalace init ` to set up wings and identity. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import queue +import re +import tempfile +import threading +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List + +from agent.memory_provider import MemoryProvider + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Named constants (FIX 12) +# --------------------------------------------------------------------------- + +_MAX_COLLECTION_SCAN: int = 10_000 +_MAX_TOP_K: int = 20 +_MAX_DIARY_LIMIT: int = 50 +_QUEUE_MAXSIZE: int = 500 +_CIRCUIT_BREAKER_THRESHOLD: int = 5 +_CIRCUIT_BREAKER_COOLDOWN: float = 120.0 +_WORKER_SHUTDOWN_TIMEOUT: float = 10.0 + + +# --------------------------------------------------------------------------- +# Hall patterns (generic — topic routing, not user-specific) +# --------------------------------------------------------------------------- + +HALL_PATTERNS: Dict[str, List[str]] = { + "hall_facts": ["decided", "agreed", "fixed", "the issue was", "root cause"], + "hall_events": ["deployed", "tested", "booked", "called", "ran"], + "hall_discoveries": ["found", "turns out", "realized", "the problem is"], + "hall_preferences": ["i prefer", "i like", "always", "never", "going forward"], + "hall_advice": ["you should", "recommend", "try", "use x instead"], +} + +_STOP_WORDS = frozenset({ + "a", "an", "the", "and", "or", "but", "in", "on", "at", "to", "for", + "of", "with", "by", "from", "is", "are", "was", "were", "be", "been", + "being", "have", "has", "had", "do", "does", "did", "will", "would", + "could", "should", "may", "might", "shall", "can", "need", "dare", + "ought", "used", "i", "you", "he", "she", "it", "we", "they", + "this", "that", "these", "those", "what", "which", "who", "whom", + "my", "your", "his", "her", "its", "our", "their", "just", "so", + "if", "then", "when", "how", "why", "up", "out", "no", "not", +}) + + +# --------------------------------------------------------------------------- +# Classification helpers +# --------------------------------------------------------------------------- + +def _classify_wing_from_config(user_msg: str, asst_msg: str, wing_config: Dict[str, Any]) -> str: + """Keyword-score both messages against loaded wing_config and return best-matching wing.""" + if not wing_config: + return "wing_general" + combined = (user_msg + " " + asst_msg).lower() + scores: Dict[str, int] = {} + for wing_name, wing_cfg in wing_config.items(): + keywords = wing_cfg.get("keywords", []) + score = sum(1 for kw in keywords if kw in combined) + if score > 0: + scores[wing_name] = score + return max(scores, key=scores.get) if scores else "wing_general" + + +def _classify_hall(user_msg: str, asst_msg: str) -> str: + """Return hall name based on dominant content pattern.""" + combined = (user_msg + " " + asst_msg).lower() + scores: Dict[str, int] = {} + for hall, patterns in HALL_PATTERNS.items(): + score = sum(1 for p in patterns if p in combined) + if score > 0: + scores[hall] = score + return max(scores, key=scores.get) if scores else "hall_facts" + + +def _make_room_name(user_msg: str, asst_msg: str) -> str: + """Slugify the first 3 meaningful words as a room name.""" + combined = (user_msg + " " + asst_msg)[:500].lower() + words = re.findall(r"\b[a-z][a-z0-9]*\b", combined) + words = [w for w in words if w not in _STOP_WORDS and len(w) > 2] + slug = "-".join(words[:3]) if words else "general" + return slug[:50] + + +def _extract_text_content(content: Any) -> str: + """Flatten message content (string or content-block list) to plain text.""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for block in content: + if isinstance(block, dict): + if block.get("type") == "text": + parts.append(block.get("text", "")) + elif isinstance(block, str): + parts.append(block) + return " ".join(parts) + return str(content) if content else "" + + +# --------------------------------------------------------------------------- +# Tool schemas +# --------------------------------------------------------------------------- + +_TOOL_SCHEMAS: List[Dict[str, Any]] = [ + { + "name": "mempalace_search", + "description": ( + "Semantic search across the palace memory store. " + "Returns verbatim drawer content ranked by similarity. " + "Optionally filter by wing (project) or room (topic)." + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What to search for."}, + "wing": { + "type": "string", + "description": "Wing to search within (e.g. wing_work, wing_personal).", + }, + "room": { + "type": "string", + "description": "Room to filter by (e.g. technical, decisions).", + }, + "top_k": { + "type": "integer", + "description": "Max results (default: 5).", + }, + }, + "required": ["query"], + }, + }, + { + "name": "mempalace_status", + "description": ( + "Show palace overview: drawer counts per wing/room, " + "identity layer status, total memory size." + ), + "parameters": {"type": "object", "properties": {}, "required": []}, + }, + { + "name": "mempalace_list_wings", + "description": "List all wings in the palace with drawer counts.", + "parameters": {"type": "object", "properties": {}, "required": []}, + }, + { + "name": "mempalace_list_rooms", + "description": "List rooms within a wing with drawer counts.", + "parameters": { + "type": "object", + "properties": { + "wing": { + "type": "string", + "description": "Wing to inspect (leave empty for all wings).", + }, + }, + "required": [], + }, + }, + { + "name": "mempalace_kg_query", + "description": ( + "Query the knowledge graph for entity relationships. " + "Supports time-filtered queries (as_of date)." + ), + "parameters": { + "type": "object", + "properties": { + "entity": {"type": "string", "description": "Entity name to look up."}, + "as_of": { + "type": "string", + "description": "ISO date string — only return facts valid at this date.", + }, + "direction": { + "type": "string", + "enum": ["outgoing", "incoming", "both"], + "description": "Relationship direction (default: outgoing).", + }, + }, + "required": ["entity"], + }, + }, + { + "name": "mempalace_kg_add", + "description": "Add a fact to the knowledge graph as a subject-predicate-object triple.", + "parameters": { + "type": "object", + "properties": { + "subject": {"type": "string", "description": "Subject entity."}, + "predicate": {"type": "string", "description": "Relationship type (e.g. works_on, loves)."}, + "object": {"type": "string", "description": "Object entity or value."}, + "valid_from": { + "type": "string", + "description": "ISO date when this fact became true.", + }, + }, + "required": ["subject", "predicate", "object"], + }, + }, + { + "name": "mempalace_diary_write", + "description": "Write an AAAK diary entry — a timestamped note for future sessions.", + "parameters": { + "type": "object", + "properties": { + "entry": {"type": "string", "description": "The diary entry to store."}, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "Optional tags (e.g. ['goal', 'insight']).", + }, + }, + "required": ["entry"], + }, + }, + { + "name": "mempalace_diary_read", + "description": "Read recent diary entries, optionally filtered by tag.", + "parameters": { + "type": "object", + "properties": { + "limit": { + "type": "integer", + "description": "Number of recent entries to return (default: 10).", + }, + "tag": { + "type": "string", + "description": "Filter entries by tag.", + }, + }, + "required": [], + }, + }, +] + + +# --------------------------------------------------------------------------- +# MemoryProvider implementation +# --------------------------------------------------------------------------- + +class MemPalaceMemoryProvider(MemoryProvider): + """Local MemPalace memory — AAAK layered recall, no API key required. + + Wing config and identity are loaded from ~/.mempalace/ at session start. + Run `mempalace init ` to generate your configuration. + """ + + def __init__(self): + self._palace_path: Path = Path(os.path.expanduser("~/.mempalace/palace")) + self._mempalace_dir: Path = Path(os.path.expanduser("~/.mempalace")) + self._diary_path: Path = self._mempalace_dir / "diary.jsonl" + + # Loaded from ~/.mempalace/wing_config.json (empty = no wing routing) + self._wing_config: Dict[str, Any] = {} + # Loaded from ~/.mempalace/identity.txt (empty = skip identity block) + self._identity: str = "" + # Config from hermes memory setup + self._config: Dict[str, Any] = {} + + # Wake-up cache (set during initialize) + self._wakeup_cache: str = "" + self._wakeup_lock = threading.Lock() + + # Prefetch cache + self._prefetch_result: str = "" + self._prefetch_lock = threading.Lock() + + # Background worker + self._work_queue: queue.Queue = queue.Queue(maxsize=_QUEUE_MAXSIZE) + self._worker_thread: threading.Thread | None = None + + # Circuit breaker + self._consecutive_failures: int = 0 + self._breaker_open_until: float = 0.0 + self._breaker_lock = threading.Lock() + + # ChromaDB client cache (FIX 3: avoid recreating PersistentClient on every call) + self._chroma_client = None + self._chroma_collection = None + self._chroma_lock = threading.Lock() + + # Async context tracking + self._agent_context: str = "primary" + + # -- Identity ------------------------------------------------------------ + + @property + def name(self) -> str: + return "mempalace" + + # -- Availability -------------------------------------------------------- + + def is_available(self) -> bool: + import importlib.util + return ( + importlib.util.find_spec("mempalace") is not None + and importlib.util.find_spec("chromadb") is not None + ) + + # -- Config schema ------------------------------------------------------- + + def get_config_schema(self) -> List[Dict[str, Any]]: + return [ + { + "key": "palace_path", + "description": "Directory for the MemPalace data store", + "required": False, + "default": "~/.mempalace/palace", + }, + { + "key": "identity_path", + "description": ( + "Path to identity.txt (L0 layer, loaded every session). " + "Create with `mempalace init`." + ), + "required": False, + "default": "~/.mempalace/identity.txt", + }, + { + "key": "top_k", + "description": "Number of results to return from semantic search", + "required": False, + "default": "5", + }, + { + "key": "min_score", + "description": "Minimum similarity score for prefetch results (0-1)", + "required": False, + "default": "0.5", + }, + ] + + def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: + """Write config to $HERMES_HOME/mempalace.json.""" + config_path = Path(hermes_home) / "mempalace.json" + existing: Dict[str, Any] = {} + if config_path.exists(): + try: + existing = json.loads(config_path.read_text(encoding="utf-8")) + except Exception as e: + logger.warning("MemPalace: could not parse existing config at %s (%s) — overwriting", config_path, e) + existing.update(values) + config_path.write_text(json.dumps(existing, indent=2), encoding="utf-8") + + # -- Circuit breaker helpers --------------------------------------------- + + def _is_breaker_open(self) -> bool: + with self._breaker_lock: + if self._consecutive_failures < _CIRCUIT_BREAKER_THRESHOLD: + return False + if time.monotonic() >= self._breaker_open_until: + self._consecutive_failures = 0 + return False + return True + + def _record_success(self) -> None: + with self._breaker_lock: + self._consecutive_failures = 0 + + def _record_failure(self) -> None: + with self._breaker_lock: + self._consecutive_failures += 1 + if self._consecutive_failures >= _CIRCUIT_BREAKER_THRESHOLD: + self._breaker_open_until = time.monotonic() + _CIRCUIT_BREAKER_COOLDOWN + logger.warning( + "MemPalace circuit breaker tripped after %d consecutive failures. " + "Pausing palace ops for %.0fs.", + self._consecutive_failures, + _CIRCUIT_BREAKER_COOLDOWN, + ) + + # -- Background worker --------------------------------------------------- + + def _start_worker(self) -> None: + """Start the persistent background worker thread.""" + self._worker_thread = threading.Thread( + target=self._worker_loop, + daemon=True, + name="mempalace-worker", + ) + self._worker_thread.start() + + def _worker_loop(self) -> None: + while True: + item = self._work_queue.get() + if item is None: + self._work_queue.task_done() + break + try: + fn, args = item + fn(*args) + except Exception as exc: + logger.warning("MemPalace worker error in %s: %s", getattr(fn, "__name__", repr(fn)), exc) + finally: + self._work_queue.task_done() + + def _enqueue(self, fn, *args) -> None: + """Queue a function for background execution (skips if breaker open).""" + if self._is_breaker_open(): + logger.debug("MemPalace: circuit breaker open, skipping %s", fn.__name__) + return + try: + self._work_queue.put((fn, args), block=True, timeout=0.5) + except queue.Full: + logger.warning("MemPalace: work queue full — dropping %s", fn.__name__) + + # -- Palace helpers ------------------------------------------------------ + + def _get_collection(self): + """Return the cached mempalace_drawers ChromaDB collection. + + No lock needed: _chroma_collection is written exactly once in initialize() + before the worker thread starts. All subsequent reads are safe without locking. + _chroma_lock is reserved for any future mutable writes to this field. + """ + return self._chroma_collection + + def _refresh_wakeup(self) -> None: + """Regenerate and cache the L0+L1 wake-up text.""" + try: + from mempalace.layers import MemoryStack + stack = MemoryStack( + palace_path=str(self._palace_path), + identity_path=str(self._mempalace_dir / "identity.txt"), + ) + text = stack.wake_up() or "" + with self._wakeup_lock: + self._wakeup_cache = text + logger.info("MemPalace: wake-up cache refreshed (%d chars)", len(text)) + except Exception as exc: + logger.warning("MemPalace: wake-up refresh failed: %s", exc) + + def _classify_wing(self, user_msg: str, asst_msg: str) -> str: + """Classify a turn into a wing using loaded wing_config.""" + return _classify_wing_from_config(user_msg, asst_msg, self._wing_config) + + # -- Lifecycle ----------------------------------------------------------- + + def initialize(self, session_id: str, **kwargs) -> None: + """Load palace config, wing routing, and identity; start worker.""" + self._agent_context = kwargs.get("agent_context", "primary") + + # FIX 5+6: Resolve hermes_home robustly with get_hermes_home() fallback + _raw_hermes_home = kwargs.get("hermes_home", "") + hermes_home = Path(_raw_hermes_home or str(Path.home() / ".hermes")) + try: + from hermes_constants import get_hermes_home + _hermes_home = get_hermes_home() + except ImportError: + _hermes_home = hermes_home + + cfg_path = _hermes_home / "mempalace.json" + if cfg_path.exists(): + try: + self._config = json.loads(cfg_path.read_text(encoding="utf-8")) + if self._config.get("palace_path"): + self._palace_path = Path( + os.path.expanduser(self._config["palace_path"]) + ) + except Exception as e: + logger.warning("MemPalace: could not read config at %s (%s) — using defaults", cfg_path, e) + + # FIX 5: Update mempalace_dir based on config, not hardcoded in __init__ + if self._config.get("mempalace_dir"): + self._mempalace_dir = Path( + os.path.expanduser(self._config["mempalace_dir"]) + ) + elif (_hermes_home / "mempalace").exists(): + self._mempalace_dir = _hermes_home / "mempalace" + # else keep default ~/.mempalace (backward compat) + self._diary_path = self._mempalace_dir / "diary.jsonl" + + # Ensure palace directory exists + self._palace_path.mkdir(parents=True, exist_ok=True) + self._mempalace_dir.mkdir(parents=True, exist_ok=True) + + # FIX 1: ChromaDB init wrapped in try/except — palace writes disabled gracefully on failure + try: + import chromadb + self._chroma_client = chromadb.PersistentClient(path=str(self._palace_path)) + self._chroma_collection = self._chroma_client.get_or_create_collection( + "mempalace_drawers" + ) + except Exception as e: + logger.error( + "MemPalace: failed to initialize ChromaDB at %s: %s — palace writes disabled.", + self._palace_path, e, + ) + self._chroma_client = None + self._chroma_collection = None + + # Load wing config if it exists — do not create it + wing_config_path = self._mempalace_dir / "wing_config.json" + if wing_config_path.exists(): + try: + with open(wing_config_path) as f: + self._wing_config = json.load(f).get("wings", {}) + if not isinstance(self._wing_config, dict): + logger.warning( + "MemPalace: wing_config 'wings' is not a dict — ignoring, " + "run `mempalace init` to fix" + ) + self._wing_config = {} + logger.info( + "MemPalace: loaded %d wings from wing_config.json", + len(self._wing_config), + ) + except Exception as exc: + logger.warning("MemPalace: failed to load wing_config.json: %s", exc) + self._wing_config = {} + else: + self._wing_config = {} + logger.info( + "MemPalace: no wing_config.json found — run `mempalace init` to configure wings" + ) + + # Load identity if it exists + identity_path = Path( + self._config.get("identity_path", "~/.mempalace/identity.txt") + ).expanduser() + self._identity = ( + identity_path.read_text(encoding="utf-8").strip() + if identity_path.exists() + else "" + ) + + # FIX 4: Start worker FIRST, then queue wake-up in background + # This avoids blocking initialize() with a potentially slow mempalace call. + # system_prompt_block() already returns "" when cache is empty — safe. + self._start_worker() + self._enqueue(self._refresh_wakeup) + + logger.info( + "MemPalace: initialized (session=%s, palace=%s)", + session_id, + self._palace_path, + ) + + def system_prompt_block(self) -> str: + """Return cached L0+L1 AAAK wake-up for injection into the system prompt.""" + with self._wakeup_lock: + cached = self._wakeup_cache + if not cached: + return "" + block = "# MemPalace Context\n" + if self._identity: + block += self._identity + "\n\n" + block += cached + return block + + # -- Prefetch ------------------------------------------------------------ + + def prefetch(self, query: str, *, session_id: str = "") -> str: + """Return cached prefetch result (populated by queue_prefetch).""" + with self._prefetch_lock: + result = self._prefetch_result + self._prefetch_result = "" + if not result: + return "" + return f"## MemPalace Recall\n{result}" + + def queue_prefetch(self, query: str, *, session_id: str = "") -> None: + """Kick off background semantic search; result is ready for next prefetch().""" + if self._is_breaker_open(): + return + self._enqueue(self._do_prefetch, query) + + def _do_prefetch(self, query: str) -> None: + """Run in background: search palace, store result in _prefetch_result.""" + try: + from mempalace.searcher import search_memories + + # Check if query keywords hint at a specific wing + wing_hint = self._classify_wing(query, "") + if wing_hint == "wing_general": + wing_hint = None # No strong signal — search globally + + result = search_memories( + query=query, + palace_path=str(self._palace_path), + wing=wing_hint, + n_results=5, + ) + + if "error" in result: + # Fall back to global search if wing-scoped search failed + if wing_hint: + result = search_memories( + query=query, + palace_path=str(self._palace_path), + n_results=5, + ) + + hits = result.get("results", []) + min_score = float(self._config.get("min_score", 0.5)) + relevant = [h for h in hits if h.get("similarity", 0) >= min_score] + + if relevant: + lines = [] + for h in relevant: + lines.append( + f"[{h.get('wing','?')}/{h.get('room','?')}] " + f"{h.get('text', h.get('content',''))[:200].strip()}" + ) + with self._prefetch_lock: + self._prefetch_result = "\n".join(lines) + + self._record_success() + except Exception as exc: + self._record_failure() + logger.debug("MemPalace prefetch failed: %s", exc) + + # -- Sync turn ----------------------------------------------------------- + + def sync_turn( + self, + user_content: str, + assistant_content: str, + *, + session_id: str = "", + ) -> None: + """Non-blocking — queue the turn for background palace filing.""" + if self._agent_context not in ("primary", ""): + return # Skip cron / subagent contexts + if self._is_breaker_open(): + return + self._enqueue(self._do_sync_turn, user_content, assistant_content) + + def _do_sync_turn(self, user_content: str, assistant_content: str) -> None: + """File a turn into the palace (runs in background worker).""" + try: + wing = self._classify_wing(user_content, assistant_content) + hall = _classify_hall(user_content, assistant_content) + room = _make_room_name(user_content, assistant_content) + + content = f"Human: {user_content}\nAssistant: {assistant_content}" + + # Stable ID for deduplication (SHA-256 over full content avoids false dedup) + drawer_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()[:24] + drawer_id = f"drawer_{wing}_{room}_{drawer_hash}" + + col = self._get_collection() + if col is None: + logger.debug("MemPalace: collection not ready, skipping %s", "_do_sync_turn") + return + + # Check for existing drawer + existing = col.get(ids=[drawer_id]) + if existing.get("ids"): + logger.debug("MemPalace: skipping duplicate drawer %s", drawer_id) + self._record_success() + return + + col.add( + documents=[content], + ids=[drawer_id], + metadatas=[ + { + "wing": wing, + "room": room, + "hall": hall, + "source_file": f"hermes_sync_{drawer_hash}", + "added_by": "hermes", + "filed_at": datetime.now().isoformat(), + "ingest_mode": "sync_turn", + } + ], + ) + self._record_success() + logger.info( + "MemPalace: filed drawer %s/%s/%s (%s)", + wing, room, hall, drawer_id, + ) + except Exception as exc: + self._record_failure() + logger.warning("MemPalace sync_turn failed: %s", exc) + + # -- Memory write hook --------------------------------------------------- + + def on_memory_write(self, action: str, target: str, content: str) -> None: + """Mirror built-in memory writes to the palace.""" + # FIX 7: capture explicit memory writes from the core memory system + if action not in ("add", "replace") or not content.strip(): + return + hall = "hall_preferences" if target == "user" else "hall_facts" + self._enqueue(self._do_file_single, content, "wing_general", hall, "memory-writes") + + def _do_file_single(self, content: str, wing: str, hall: str, room: str) -> None: + """File a single text entry to the palace collection (background helper).""" + try: + drawer_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()[:24] + drawer_id = f"drawer_{wing}_{room}_{drawer_hash}" + + col = self._get_collection() + if col is None: + logger.debug("MemPalace: collection not ready, skipping %s", "_do_file_single") + return + + existing = col.get(ids=[drawer_id]) + if existing.get("ids"): + logger.debug("MemPalace: skipping duplicate drawer %s", drawer_id) + self._record_success() + return + + col.add( + documents=[content], + ids=[drawer_id], + metadatas=[ + { + "wing": wing, + "room": room, + "hall": hall, + "source_file": f"hermes_memory_write_{drawer_hash}", + "added_by": "hermes", + "filed_at": datetime.now().isoformat(), + "ingest_mode": "memory_write", + } + ], + ) + self._record_success() + logger.info( + "MemPalace: filed memory-write drawer %s/%s/%s", wing, room, hall + ) + except Exception as exc: + self._record_failure() + logger.warning("MemPalace _do_file_single failed: %s", exc) + + # -- Session end --------------------------------------------------------- + + def on_session_end(self, messages: List[Dict[str, Any]]) -> None: + """Mine full session into palace + regenerate L1 AAAK cache.""" + self._enqueue(self._do_session_end, messages) + + def _do_session_end(self, messages: List[Dict[str, Any]]) -> None: + """Background: serialize session, mine convos, refresh wake-up.""" + try: + # Build content in Claude export format (> user turn, then response) + lines: List[str] = [] + for msg in messages: + role = msg.get("role", "") + raw_content = msg.get("content", "") + content = _extract_text_content(raw_content) + if not content or not content.strip(): + continue + if role == "user": + lines.append(f"> {content.strip()}") + elif role == "assistant": + lines.append(content.strip()) + lines.append("") + + if not lines: + return + + # Detect dominant wing for the session + combined = " ".join(lines[:30]) + wing = self._classify_wing(combined[:600], "") + + # Write to a temp directory and mine + with tempfile.TemporaryDirectory(prefix="hermes_mp_") as tmpdir: + session_file = Path(tmpdir) / "session.txt" + session_file.write_text("\n".join(lines), encoding="utf-8") + + from mempalace.convo_miner import mine_convos + mine_convos( + convo_dir=tmpdir, + palace_path=str(self._palace_path), + wing=wing, + agent="hermes", + ) + + self._record_success() + logger.info("MemPalace: session mined into wing=%s", wing) + + # Clear wakeup cache so next session gets fresh context + with self._wakeup_lock: + self._wakeup_cache = "" + + # Runs synchronously in the worker thread — intentional, ensures L1 is + # regenerated before the next session start reads system_prompt_block(). + # Regenerate L1 AAAK + self._refresh_wakeup() + + except Exception as exc: + self._record_failure() + logger.warning("MemPalace on_session_end failed: %s", exc) + + # -- Pre-compress -------------------------------------------------------- + + def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str: + """Extract key exchanges from about-to-be-discarded messages.""" + try: + pairs: List[tuple] = [] + for i in range(len(messages) - 1): + if ( + messages[i].get("role") == "user" + and messages[i + 1].get("role") == "assistant" + ): + u = _extract_text_content(messages[i].get("content", "")) + a = _extract_text_content(messages[i + 1].get("content", "")) + if u and a: + pairs.append((u, a)) + + # File top 5 pairs to palace asynchronously + for u, a in pairs[:5]: + self._enqueue(self._do_sync_turn, u, a) + + if pairs: + return ( + f"MemPalace: {len(pairs)} exchange(s) preserved in palace " + f"before compression." + ) + except Exception as exc: + logger.debug("MemPalace on_pre_compress error: %s", exc) + return "" + + # -- Tool schemas -------------------------------------------------------- + + def get_tool_schemas(self) -> List[Dict[str, Any]]: + return list(_TOOL_SCHEMAS) + + # -- Tool dispatch ------------------------------------------------------- + + def handle_tool_call( + self, tool_name: str, args: Dict[str, Any], **kwargs + ) -> str: + if self._is_breaker_open(): + return json.dumps({ + "error": ( + "MemPalace temporarily unavailable (circuit breaker open " + "after consecutive failures). Will retry automatically." + ) + }) + + try: + if tool_name == "mempalace_search": + return self._tool_search(args) + elif tool_name == "mempalace_status": + return self._tool_status() + elif tool_name == "mempalace_list_wings": + return self._tool_list_wings() + elif tool_name == "mempalace_list_rooms": + return self._tool_list_rooms(args) + elif tool_name == "mempalace_kg_query": + return self._tool_kg_query(args) + elif tool_name == "mempalace_kg_add": + return self._tool_kg_add(args) + elif tool_name == "mempalace_diary_write": + return self._tool_diary_write(args) + elif tool_name == "mempalace_diary_read": + return self._tool_diary_read(args) + else: + return json.dumps({"error": f"Unknown tool: {tool_name}"}) + except Exception as exc: + self._record_failure() + return json.dumps({"error": str(exc)}) + + # -- Tool implementations ------------------------------------------------ + + def _tool_search(self, args: Dict[str, Any]) -> str: + query = args.get("query", "") + if not query: + return json.dumps({"error": "Missing required parameter: query"}) + + wing = args.get("wing") or None + room = args.get("room") or None + + try: + top_k = min(int(args.get("top_k", 5)), _MAX_TOP_K) + from mempalace.searcher import search_memories + result = search_memories( + query=query, + palace_path=str(self._palace_path), + wing=wing, + room=room, + n_results=top_k, + ) + self._record_success() + return json.dumps(result) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"Search failed: {exc}"}) + + def _tool_status(self) -> str: + try: + from mempalace.layers import MemoryStack + stack = MemoryStack( + palace_path=str(self._palace_path), + identity_path=str(self._mempalace_dir / "identity.txt"), + ) + status = stack.status() + status["palace_path"] = str(self._palace_path) + self._record_success() + return json.dumps(status) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"Status failed: {exc}"}) + + def _tool_list_wings(self) -> str: + # FIX 3: use cached collection instead of creating a new PersistentClient + try: + col = self._get_collection() + if col is None: + return json.dumps({"wings": {}, "total_drawers": 0}) + + results = col.get(include=["metadatas"], limit=_MAX_COLLECTION_SCAN) + wings: Dict[str, int] = {} + for meta in results.get("metadatas", []): + w = meta.get("wing", "unknown") + wings[w] = wings.get(w, 0) + 1 + + self._record_success() + return json.dumps({ + "wings": wings, + "total_drawers": sum(wings.values()), + }) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"List wings failed: {exc}"}) + + def _tool_list_rooms(self, args: Dict[str, Any]) -> str: + wing = args.get("wing") or None + # FIX 3: use cached collection instead of creating a new PersistentClient + try: + col = self._get_collection() + if col is None: + return json.dumps({"rooms": {}, "wing": wing}) + + get_kwargs: Dict[str, Any] = {"include": ["metadatas"], "limit": _MAX_COLLECTION_SCAN} + if wing: + get_kwargs["where"] = {"wing": wing} + + results = col.get(**get_kwargs) + rooms: Dict[str, int] = {} + for meta in results.get("metadatas", []): + r = meta.get("room", "unknown") + rooms[r] = rooms.get(r, 0) + 1 + + self._record_success() + return json.dumps({"wing": wing, "rooms": rooms}) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"List rooms failed: {exc}"}) + + def _tool_kg_query(self, args: Dict[str, Any]) -> str: + entity = args.get("entity", "") + if not entity: + return json.dumps({"error": "Missing required parameter: entity"}) + + as_of = args.get("as_of") or None + direction = args.get("direction", "outgoing") + + try: + from mempalace.knowledge_graph import KnowledgeGraph + kg = KnowledgeGraph( + db_path=str(self._mempalace_dir / "knowledge_graph.sqlite3") + ) + results = kg.query_entity(entity, as_of=as_of, direction=direction) + self._record_success() + return json.dumps({"entity": entity, "results": results, "count": len(results)}) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"KG query failed: {exc}"}) + + def _tool_kg_add(self, args: Dict[str, Any]) -> str: + subject = args.get("subject", "") + predicate = args.get("predicate", "") + obj = args.get("object", "") + + if not subject or not predicate or not obj: + return json.dumps({"error": "Missing required: subject, predicate, object"}) + + valid_from = args.get("valid_from") or None + + try: + from mempalace.knowledge_graph import KnowledgeGraph + kg = KnowledgeGraph( + db_path=str(self._mempalace_dir / "knowledge_graph.sqlite3") + ) + triple_id = kg.add_triple( + subject=subject, + predicate=predicate, + obj=obj, + valid_from=valid_from, + ) + self._record_success() + return json.dumps({ + "result": "Triple added.", + "triple_id": triple_id, + "triple": f"{subject} -{predicate}-> {obj}", + }) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"KG add failed: {exc}"}) + + def _tool_diary_write(self, args: Dict[str, Any]) -> str: + entry = args.get("entry", "") + if not entry: + return json.dumps({"error": "Missing required parameter: entry"}) + + tags = args.get("tags", []) + if not isinstance(tags, list): + tags = [tags] if tags else [] + + record = { + "ts": datetime.now().isoformat(), + "entry": entry, + "tags": tags, + } + + try: + self._mempalace_dir.mkdir(parents=True, exist_ok=True) + with open(self._diary_path, "a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") + self._record_success() + return json.dumps({"result": "Diary entry written.", "ts": record["ts"]}) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"Diary write failed: {exc}"}) + + def _tool_diary_read(self, args: Dict[str, Any]) -> str: + tag_filter = args.get("tag") or None + + try: + limit = min(int(args.get("limit", 10)), _MAX_DIARY_LIMIT) + if not self._diary_path.exists(): + return json.dumps({"entries": [], "count": 0}) + + entries = [] + with open(self._diary_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entries.append(json.loads(line)) + except json.JSONDecodeError: + continue + + if tag_filter: + entries = [e for e in entries if tag_filter in e.get("tags", [])] + + # Most recent first + entries = entries[-limit:][::-1] + + self._record_success() + return json.dumps({"entries": entries, "count": len(entries)}) + except Exception as exc: + self._record_failure() + return json.dumps({"error": f"Diary read failed: {exc}"}) + + # -- Shutdown ------------------------------------------------------------ + + def shutdown(self) -> None: + """Flush queued operations and stop the worker thread.""" + if self._worker_thread and self._worker_thread.is_alive(): + self._work_queue.put(None) # sentinel + self._worker_thread.join(timeout=_WORKER_SHUTDOWN_TIMEOUT) + if self._worker_thread.is_alive(): + logger.warning( + "MemPalace: worker thread did not finish within %.0fs — " + "approximately %d queued items may not have been written to the palace.", + _WORKER_SHUTDOWN_TIMEOUT, + self._work_queue.qsize(), + ) + logger.info("MemPalace: shutdown complete.") + + +# --------------------------------------------------------------------------- +# Plugin registration +# --------------------------------------------------------------------------- + +def register(ctx) -> None: + """Register MemPalace as a memory provider plugin.""" + ctx.register_memory_provider(MemPalaceMemoryProvider()) diff --git a/plugins/memory/mempalace/backfill.py b/plugins/memory/mempalace/backfill.py new file mode 100644 index 000000000000..ecae42b7932d --- /dev/null +++ b/plugins/memory/mempalace/backfill.py @@ -0,0 +1,379 @@ +#!/usr/bin/env python3 +"""MemPalace Backfill — mine existing Hermes session history into the palace. + +Reads all sessions from the Hermes SQLite state DB, exports them as +conversation files (Claude export format), then mines each one into the +palace using mempalace's convo_miner. + +Wing classification uses ~/.mempalace/wing_config.json if it exists, +otherwise all sessions go to wing_general. + +Usage: + python -m plugins.memory.mempalace.backfill + python -m plugins.memory.mempalace.backfill --limit 50 + python -m plugins.memory.mempalace.backfill --dry-run + python -m plugins.memory.mempalace.backfill --palace ~/.mempalace/palace + python -m plugins.memory.mempalace.backfill --wing wing_general --source cli +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import tempfile +from pathlib import Path +from typing import Any, Dict, List, Optional + +# FIX 8: import _extract_text_content from the main module — no duplication +from plugins.memory.mempalace import _extract_text_content + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _load_wing_config() -> Dict[str, Any]: + """Load wing config from ~/.mempalace/wing_config.json if it exists.""" + wing_config_path = Path.home() / ".mempalace" / "wing_config.json" + if wing_config_path.exists(): + try: + with open(wing_config_path) as f: + return json.load(f).get("wings", {}) + except Exception: + pass + return {} + + +def _classify_wing_simple(text: str, wing_config: Dict[str, Any]) -> str: + """Keyword-based wing classification using loaded wing_config.""" + if not wing_config: + return "wing_general" + combined = text.lower() + scores: Dict[str, int] = {} + for wing_name, wing_cfg in wing_config.items(): + keywords = wing_cfg.get("keywords", []) + score = sum(1 for kw in keywords if kw in combined) + if score > 0: + scores[wing_name] = score + return max(scores, key=scores.get) if scores else "wing_general" + + +def _messages_to_export_format(messages: List[Dict[str, Any]]) -> str: + """Convert Hermes messages to Claude export format (> User / response pairs).""" + lines: List[str] = [] + for msg in messages: + role = msg.get("role", "") + raw_content = msg.get("content", "") + content = _extract_text_content(raw_content) + if not content or not content.strip(): + continue + # Skip tool calls and tool results — they're noise for mining + if msg.get("tool_calls") or msg.get("tool_name"): + continue + if role == "user": + lines.append(f"> {content.strip()}") + elif role == "assistant": + lines.append(content.strip()) + lines.append("") # blank line between exchanges + return "\n".join(lines) + + +def _find_hermes_state_db() -> Optional[Path]: + """Locate the Hermes state DB.""" + # Primary: use hermes_constants if available + try: + from hermes_constants import get_hermes_home + db = get_hermes_home() / "state.db" + if db.exists(): + return db + except ImportError: + pass + + # Fallback: look in ~/.hermes/ + for candidate in [ + Path.home() / ".hermes" / "state.db", + Path.home() / ".hermes" / "hermes.db", + ]: + if candidate.exists(): + return candidate + + return None + + +# --------------------------------------------------------------------------- +# Main backfill logic +# --------------------------------------------------------------------------- + +def backfill( + palace_path: Optional[str] = None, + limit: int = 0, + dry_run: bool = False, + wing_override: Optional[str] = None, + source_filter: Optional[str] = None, + verbose: bool = False, +) -> int: + """Mine all Hermes sessions into the MemPalace. + + Returns the number of sessions successfully mined. + """ + # ------------------------------------------------------------------ + # 1. Find state DB + # ------------------------------------------------------------------ + db_path = _find_hermes_state_db() + if db_path is None: + print("ERROR: Could not find Hermes state DB (state.db).") + print(" Make sure HERMES_HOME is set or hermes_state is importable.") + return 0 + + print(f"\n Source DB: {db_path}") + + # ------------------------------------------------------------------ + # 2. Resolve palace path + # ------------------------------------------------------------------ + if palace_path: + resolved_palace = os.path.expanduser(palace_path) + else: + try: + from mempalace.config import MempalaceConfig + resolved_palace = MempalaceConfig().palace_path + except ImportError: + resolved_palace = os.path.expanduser("~/.mempalace/palace") + + print(f" Palace: {resolved_palace}") + if dry_run: + print(" DRY RUN — nothing will be filed") + print() + + # ------------------------------------------------------------------ + # 3. Check mempalace is installed + # ------------------------------------------------------------------ + try: + from mempalace.convo_miner import mine_convos + except ImportError: + print("ERROR: mempalace package not installed.") + print(" Run: pip install mempalace") + return 0 + + # ------------------------------------------------------------------ + # 4. Load wing config + # ------------------------------------------------------------------ + wing_config = _load_wing_config() + if wing_config: + print(f" Wing config: {len(wing_config)} wings loaded from ~/.mempalace/wing_config.json") + else: + print(" Wing config: not found — all sessions will go to wing_general") + print(" (run `mempalace init` to configure wings)") + + # ------------------------------------------------------------------ + # 5. Open session DB + # ------------------------------------------------------------------ + try: + # Try the Hermes SessionDB class first (respects WAL mode etc.) + from hermes_state import SessionDB + db = SessionDB(db_path) + use_session_db = True + except Exception as exc: + if verbose: + print(f" Warning: SessionDB unavailable ({exc}), falling back to raw SQLite") + import sqlite3 + db = sqlite3.connect(str(db_path)) + db.row_factory = sqlite3.Row + use_session_db = False + + # ------------------------------------------------------------------ + # 6. Enumerate sessions + # ------------------------------------------------------------------ + try: + if use_session_db: + sessions = db.list_sessions_rich( + source=source_filter, + limit=limit if limit > 0 else 9999, + include_children=False, + ) + else: + query = "SELECT id, source, started_at FROM sessions" + params: List[Any] = [] + if source_filter: + query += " WHERE source = ?" + params.append(source_filter) + query += " ORDER BY started_at DESC" + if limit > 0: + query += f" LIMIT {limit}" + sessions = [dict(r) for r in db.execute(query, params).fetchall()] + except Exception as exc: + print(f"ERROR reading sessions: {exc}") + return 0 + + total = len(sessions) + print(f" Sessions found: {total}") + if total == 0: + print(" Nothing to backfill.") + return 0 + + print() + print(f"{'=' * 60}") + print(" MemPalace Backfill") + print(f"{'=' * 60}\n") + + # ------------------------------------------------------------------ + # 7. Mine each session + # ------------------------------------------------------------------ + mined = 0 + skipped = 0 + errors = 0 + + for idx, session in enumerate(sessions, 1): + session_id = session.get("id", "") + source = session.get("source", "unknown") + preview = session.get("preview", "")[:50] + + # Fetch messages + try: + if use_session_db: + messages = db.get_messages(session_id) + else: + rows = db.execute( + "SELECT role, content FROM messages WHERE session_id = ? ORDER BY timestamp", + (session_id,), + ).fetchall() + messages = [dict(r) for r in rows] + except Exception as exc: + if verbose: + print(f" [{idx:4}/{total}] SKIP {session_id[:8]}... — message fetch failed: {exc}") + errors += 1 + continue + + # Build export text + export_text = _messages_to_export_format(messages) + if len(export_text.strip()) < 50: + if verbose: + print(f" [{idx:4}/{total}] SKIP {session_id[:8]}... — too short") + skipped += 1 + continue + + # Detect wing + if wing_override: + wing = wing_override + else: + wing = _classify_wing_simple(export_text[:1000], wing_config) + + if dry_run: + word_count = len(export_text.split()) + print( + f" [{idx:4}/{total}] DRY {session_id[:8]}... " + f"source={source:8} wing={wing:20} {word_count}w \"{preview}\"" + ) + mined += 1 + continue + + # Write to temp dir and mine + try: + with tempfile.TemporaryDirectory(prefix="hermes_backfill_") as tmpdir: + session_file = Path(tmpdir) / f"session_{session_id[:8]}.txt" + session_file.write_text(export_text, encoding="utf-8") + + # Suppress mine_convos' own print output unless verbose + if not verbose: + import io + import contextlib + with contextlib.redirect_stdout(io.StringIO()): + mine_convos( + convo_dir=tmpdir, + palace_path=resolved_palace, + wing=wing, + agent="hermes-backfill", + ) + else: + mine_convos( + convo_dir=tmpdir, + palace_path=resolved_palace, + wing=wing, + agent="hermes-backfill", + ) + + mined += 1 + print( + f" [{idx:4}/{total}] OK {session_id[:8]}... " + f"source={source:8} wing={wing:20} \"{preview}\"" + ) + except Exception as exc: + errors += 1 + print(f" [{idx:4}/{total}] ERR {session_id[:8]}... — {exc}") + + # ------------------------------------------------------------------ + # 8. Report + # ------------------------------------------------------------------ + print(f"\n{'=' * 60}") + print(" Backfill complete.") + print(f" Sessions mined: {mined}") + print(f" Sessions skipped: {skipped} (too short / no content)") + print(f" Errors: {errors}") + if not dry_run and mined > 0: + print() + print(f" Palace: {resolved_palace}") + print(' Next: mempalace wake-up') + print(f"{'=' * 60}\n") + + return mined + + +# --------------------------------------------------------------------------- +# CLI entry point +# --------------------------------------------------------------------------- + +def main() -> None: + parser = argparse.ArgumentParser( + description="MemPalace Backfill — mine Hermes session history into the palace.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + parser.add_argument( + "--palace", + default=None, + help="Palace directory (default: from ~/.mempalace/config.json or ~/.mempalace/palace)", + ) + parser.add_argument( + "--limit", + type=int, + default=0, + help="Max sessions to process (0 = all)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be mined without writing to palace", + ) + parser.add_argument( + "--wing", + default=None, + help="Override wing for all sessions (default: auto-detect per session)", + ) + parser.add_argument( + "--source", + default=None, + help="Only mine sessions from this source (e.g. cli, telegram, discord)", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Show detailed output including mine_convos progress", + ) + + args = parser.parse_args() + + count = backfill( + palace_path=args.palace, + limit=args.limit, + dry_run=args.dry_run, + wing_override=args.wing, + source_filter=args.source, + verbose=args.verbose, + ) + + sys.exit(0 if count >= 0 else 1) + + +if __name__ == "__main__": + main() diff --git a/plugins/memory/mempalace/plugin.yaml b/plugins/memory/mempalace/plugin.yaml new file mode 100644 index 000000000000..dcc6e0c50de1 --- /dev/null +++ b/plugins/memory/mempalace/plugin.yaml @@ -0,0 +1,10 @@ +name: mempalace +version: 1.0.0 +description: "MemPalace — local palace-structured memory with AAAK compression, ChromaDB semantic search, and knowledge graph. 96.6% LongMemEval R@5, no API key required." +pip_dependencies: + - mempalace>=3.0.0 + - chromadb>=0.4.0 +hooks: + - on_session_end + - on_pre_compress + - on_memory_write