""" Coppermind — Persistent AI memory for Hermes Agent. This plugin gives Hermes automatic, persistent memory powered by SurrealDB. Every conversation turn is captured and every prompt is enriched with relevant context — deterministically, via lifecycle hooks. Hooks: - pre_llm_call: Searches memory, injects relevant context into the system prompt - post_llm_call: Captures the conversation turn into memory - on_session_end: Creates and stores a session summary Tools: - coppermind_search: Deep semantic search across all memories - coppermind_recall: Get recent memories for broad context """ import logging import shutil from pathlib import Path from . import client, schemas, tools from .memory import ( distill_turn_for_ingest, format_memories, summarize_conversation, ) logger = logging.getLogger("coppermind") # Track session data for the on_session_end summary _session_turns: dict[str, list[dict]] = {} # --------------------------------------------------------------------------- # Lifecycle hooks # --------------------------------------------------------------------------- def _on_pre_llm_call( session_id: str = "", user_message: str = "", conversation_history: list | None = None, is_first_turn: bool = False, **kwargs, ) -> dict | None: """Called before every LLM call. Searches Coppermind for memories relevant to the user's message and returns context for system prompt injection. Returns: {"context": "..."} to inject into the system prompt, or None. """ if not user_message or not user_message.strip(): return None # Skip retrieval for system messages — skill invocations and internal # directives aren't useful search queries and just pollute context if user_message.strip().startswith("[SYSTEM:"): return None try: # Use the user message as the search query result = client.search_memory(query=user_message.strip(), limit=10) context = format_memories(result) if context: logger.debug( "Injecting %d chars of memory context (session=%s)", len(context), session_id, ) return {"context": context} except client.CoppermindError as e: # Graceful degradation — don't block the agent if memory is unavailable logger.warning("Memory retrieval failed (continuing without context): %s", e) except Exception as e: logger.error("Unexpected error in pre_llm_call: %s", e) return None def _on_post_llm_call( session_id: str = "", user_message: str = "", assistant_response: str = "", conversation_history: list | None = None, **kwargs, ) -> None: """Called after every LLM call. Captures the conversation turn into Coppermind memory. Fire-and-forget — does not block the response. """ if not user_message or not user_message.strip(): return if not assistant_response or not assistant_response.strip(): return # Skip system messages (skill invocations, internal directives) # — they pollute memory with meta-noise and aren't useful to recall if user_message.strip().startswith("[SYSTEM:"): return # Track turns for session summary if session_id: if session_id not in _session_turns: _session_turns[session_id] = [] _session_turns[session_id].append({ "role": "user", "content": user_message, }) _session_turns[session_id].append({ "role": "assistant", "content": assistant_response, }) try: candidate = distill_turn_for_ingest( user_message=user_message.strip(), assistant_response=assistant_response.strip(), session_id=session_id or None, ) if not candidate: return client.ingest_memory( content=candidate["content"], tags=candidate.get("tags"), metadata=candidate.get("metadata"), session_id=session_id or None, ) logger.debug( "Ingested distilled turn memory (%d chars, session=%s)", len(candidate["content"]), session_id, ) except client.CoppermindError as e: # Don't crash on ingest failure — log and continue logger.warning("Memory ingestion failed: %s", e) except Exception as e: logger.error("Unexpected error in post_llm_call: %s", e) def _on_session_end( session_id: str = "", run_conversation: list | None = None, completed: bool = True, **kwargs, ) -> None: """Called when a Hermes session ends. Generates a session summary and stores it as a high-level memory, which is useful for recalling "what we worked on" in future sessions. """ # Use tracked turns or the provided conversation history history = ( _session_turns.pop(session_id, []) or run_conversation or [] ) if not history: return try: summary = summarize_conversation(history) if summary: client.ingest_memory( content=summary, tags=["hermes", "session_summary"], metadata={ "session_id": session_id, "completed": completed, "source_client": "hermes", "capture_kind": "session_summary", "memoryType": "project_state", "importance": 0.58, "retrievalPriority": 1, }, session_id=session_id or None, ) logger.info( "Stored session summary (%d chars, session=%s)", len(summary), session_id, ) except client.CoppermindError as e: logger.warning("Session summary ingestion failed: %s", e) except Exception as e: logger.error("Unexpected error in on_session_end: %s", e) # --------------------------------------------------------------------------- # Skill installer # --------------------------------------------------------------------------- def _install_skill(): """Copy the bundled skill.md to ~/.hermes/skills/coppermind/ on first load.""" try: from hermes_cli.config import get_hermes_home dest = get_hermes_home() / "skills" / "coppermind" / "SKILL.md" except Exception: dest = Path.home() / ".hermes" / "skills" / "coppermind" / "SKILL.md" # Don't overwrite user edits if dest.exists(): return source = Path(__file__).parent / "skill.md" if source.exists(): dest.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, dest) logger.info("Installed Coppermind skill to %s", dest) # --------------------------------------------------------------------------- # Plugin registration # --------------------------------------------------------------------------- def register(ctx): """Wire schemas to handlers and register hooks. Called exactly once at Hermes startup. If this crashes, the plugin is disabled but Hermes continues normally. """ # Cloud mode needs an API key, but the local runtime path is intentionally # keyless for first local success. if client.requires_api_key() and not client.API_KEY: logger.error( "COPPERMIND_API_KEY not set — plugin disabled. " "Get your free key at https://coppermindapi.com" ) return # Register tools ctx.register_tool( name="coppermind_search", toolset="coppermind", schema=schemas.COPPERMIND_SEARCH, handler=tools.coppermind_search, ) ctx.register_tool( name="coppermind_recall", toolset="coppermind", schema=schemas.COPPERMIND_RECALL, handler=tools.coppermind_recall, ) # Register lifecycle hooks ctx.register_hook("pre_llm_call", _on_pre_llm_call) ctx.register_hook("post_llm_call", _on_post_llm_call) ctx.register_hook("on_session_end", _on_session_end) # Install bundled skill _install_skill() logger.info( "Coppermind plugin loaded — API: %s, User: %s", client.BASE_URL, client.USER_ID, )