""" Coppermind Memory Provider for Hermes Agent. This provider integrates Coppermind (SurrealDB-backed memory) as a memory provider plugin. It uses the same API as the hook-based plugin but implements the official MemoryProvider ABC for clean Hermes integration. Configuration via environment variables: - COPPERMIND_BASE_URL: API endpoint (required) - COPPERMIND_USER_ID: user identifier (default: "default") - COPPERMIND_TIMEOUT: request timeout in seconds (default: 10) - COPPERMIND_MAX_QUERY_LENGTH: max query length for search (default: 2000) """ import json import logging import os import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, List, Optional from agent.memory_provider import MemoryProvider logger = logging.getLogger(__name__) class CoppermindMemoryProvider(MemoryProvider): """Memory provider that uses Coppermind for persistent storage.""" @property def name(self) -> str: return "coppermind" def is_available(self) -> bool: """Check if COPPERMIND_BASE_URL is set.""" return bool(os.environ.get("COPPERMIND_BASE_URL")) def initialize(self, session_id: str, **kwargs) -> None: """Initialize provider configuration from environment.""" self.session_id = session_id self.user_id = os.environ.get("COPPERMIND_USER_ID", "default") self.base_url = os.environ["COPPERMIND_BASE_URL"].rstrip("/") self.timeout = int(os.environ.get("COPPERMIND_TIMEOUT", "10")) self.max_query_length = int( os.environ.get("COPPERMIND_MAX_QUERY_LENGTH", "2000") ) logger.debug( "CoppermindMemoryProvider initialized: base_url=%s, user_id=%s, max_query=%d", self.base_url, self.user_id, self.max_query_length, ) # ------------------------------------------------------------------------ # Core lifecycle # ------------------------------------------------------------------------ def prefetch(self, query: str, *, session_id: str = "") -> str: """Recall relevant memories for the upcoming turn. Args: query: The user's query/message to search against. session_id: Optional session identifier for scoping. Returns: Formatted context string to inject, or empty string if none. """ if not query.strip(): return "" try: # Truncate to avoid header size limits (431 errors) safe_query = query.strip()[: self.max_query_length] result = self._search_memory(safe_query, limit=5) return self._format_context(result) except Exception as e: logger.warning("Coppermind prefetch failed: %s", e) return "" def sync_turn( self, user_content: str, assistant_content: str, *, session_id: str = "" ) -> None: """Persist a completed conversation turn to Coppermind. This should be non-blocking; any heavy work should be queued. For simplicity we do it inline but with timeout protection. """ if not user_content.strip() or not assistant_content.strip(): return try: content = self._format_conversation_for_ingest( user_content, assistant_content, session_id or self.session_id ) self._ingest_memory(content, tags=["hermes", "conversation"]) except Exception as e: logger.warning("Coppermind sync_turn failed: %s", e) def get_tool_schemas(self) -> List[Dict[str, Any]]: """No tools exposed by this provider (automatic recall only).""" return [] def handle_tool_call( self, tool_name: str, args: Dict[str, Any], **kwargs ) -> str: raise NotImplementedError( f"Provider {self.name} does not handle tool {tool_name}" ) def shutdown(self) -> None: """No special cleanup needed.""" pass # ------------------------------------------------------------------------ # Optional hooks # ------------------------------------------------------------------------ def on_session_end(self, messages: List[Dict[str, Any]]) -> None: """Store a session summary when the session ends.""" user_messages = [ msg["content"] for msg in messages if msg.get("role") == "user" and msg.get("content") ] if not user_messages: return parts = [f"Session summary ({len(user_messages)} user messages):"] for i, msg in enumerate(user_messages[:10], 1): preview = msg[:200].rstrip() + ("…" if len(msg) > 200 else "") parts.append(f" {i}. {preview}") if len(user_messages) > 10: parts.append(f" ... and {len(user_messages) - 10} more messages") summary = "\n".join(parts) try: self._ingest_memory( summary, tags=["hermes", "session_summary"], metadata={"completed": True}, ) except Exception as e: logger.warning("Failed to store session summary: %s", e) def get_config_schema(self) -> List[Dict[str, Any]]: """Provider uses environment variables; no interactive setup needed.""" return [] def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: """No native config file to write.""" pass # ------------------------------------------------------------------------ # Internal HTTP helpers # ------------------------------------------------------------------------ def _search_memory(self, query: str, limit: int = 5) -> Dict[str, Any]: """GET /v1/memory/search Note: Uses GET with query parameters. Query should be pre-truncated to avoid 431 Request Header Fields Too Large errors. """ url = f"{self.base_url}/v1/memory/search" params = { "userId": self.user_id, "query": query, "limit": str(limit), } full_url = f"{url}?{urllib.parse.urlencode(params)}" req = urllib.request.Request(full_url, method="GET") with urllib.request.urlopen(req, timeout=self.timeout) as resp: return json.loads(resp.read().decode("utf-8")) def _ingest_memory( self, content: str, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, ) -> None: """POST /v1/ingest""" url = f"{self.base_url}/v1/ingest" body = { "userId": self.user_id, "data": [ { "content": content, **({"tags": tags} if tags else {}), **({"metadata": metadata} if metadata else {}), } ], } data = json.dumps(body).encode("utf-8") req = urllib.request.Request( url, data=data, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=self.timeout): pass # 200 OK # ------------------------------------------------------------------------ # Formatting utilities # ------------------------------------------------------------------------ def _format_context(self, api_response: Dict[str, Any]) -> str: """Convert search response into an XML context block.""" fragments = api_response.get("fragments", []) if not fragments: return "" lines = [] total_chars = 0 max_context_chars = 3000 max_fragment_chars = 500 for frag in fragments: content = ( frag.get("content") or frag.get("contentSummary") or frag.get("text") or "" ) if not content: continue content_len = len(content) if content_len > max_fragment_chars: content = content[:max_fragment_chars].rstrip() + "…" content_len = max_fragment_chars new_total = total_chars + content_len if new_total > max_context_chars: break lines.append(f" - {content}") total_chars = new_total if not lines: return "" # Build the context block with actual newlines inner = "\n".join(lines) return f"\n{inner}\n" def _format_conversation_for_ingest( self, user_msg: str, assistant_msg: str, session_id: str ) -> str: """Create a structured representation for storage.""" parts = [f"[Session: {session_id}]", f"User: {user_msg}", f"Assistant: {assistant_msg}"] # Truncate very long responses to avoid excessive storage if len(assistant_msg) > 2000: assistant_msg = assistant_msg[:2000].rstrip() + "… [truncated]" return "\n".join(parts) # ------------------------------------------------------------------------ # Plugin entry point # ------------------------------------------------------------------------ def register(ctx) -> None: """Called by the memory plugin discovery system.""" ctx.register_memory_provider(CoppermindMemoryProvider())