hermes-memory-manager

name	hermes-memory-manager
description	Single-integration-point memory orchestrator — delegates to multiple providers, one external allowed, background async sync so slow providers never block a turn. Lifecycle hooks for compression/session-switch. Source: NousResearch/hermes-agent (MIT).

Implementation (real, runnable — added 2026-06-19)

• Module: core/lib/hermes_adapted/memory_manager.py (registration, prefetch/sync)
  • Lifecycle hooks (on_turn_start/on_session_end/on_session_switch/ on_pre_compress/on_memory_write/on_delegation) + tool-call dispatch (get_all_tool_schemas/handle_tool_call/flush_pending) mixed in from core/lib/hermes_adapted/memory_manager_lifecycle.py
• Tests: tests/test_hermes_memory_manager.py (14 passing)

MemoryManager is condensed from the original (hermes' MemoryProvider ABC and reserved-tool-name set replaced with a plain duck-typed Protocol and an injectable reserved_tool_names set), but the real invariants carry over: exactly one external provider at a time, prefetch/sync run off the calling thread on a single-worker executor (ordering preserved), bounded- timeout shutdown so a wedged provider can't block teardown, and a failing provider never breaks the hook for every other provider.

/hermes-memory-manager

When to Use

• Building an agent that connects to multiple memory backends (built-in L1 + external vector store)
• Need memory sync to not block the turn response (background write)
• Switching between memory providers without rewriting all dependent code
• Any agent memory layer that needs provider isolation and lifecycle hooks

Do NOT use for

• Single-provider memory with no switching needs — direct integration is simpler
• Synchronous memory that must complete before next turn (use direct await)
• See also: [[hermes-streaming-scrubber]] for keeping memory blocks out of UI

---

Design Principles

1. One integration point   — run_agent.py calls MemoryManager, not each provider directly
2. One external provider   — prevents tool schema bloat; multiple built-ins ok
3. Background sync         — sync_all() runs in a separate worker thread; turn never waits
4. Ordered prompt assembly — built-in providers first, then external
5. Reserved tool names      — external providers cannot shadow core tool names

---

Provider Interface

from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor

class MemoryProvider(ABC):
    """Implement this to add a memory backend."""

    @property
    @abstractmethod
    def name(self) -> str: ...

    @property
    def tool_names(self) -> list[str]:
        """Tool names this provider registers. Must not collide with core."""
        return []

    def build_system_prompt_block(self) -> str:
        """Return text to inject into system prompt (stable tier)."""
        return ""

    def prefetch(self, messages: list[dict]) -> None:
        """Called before each turn — load context synchronously."""
        pass

    def sync(self, messages: list[dict]) -> None:
        """Called after each turn — persist in background worker."""
        pass

    # Lifecycle hooks
    def on_turn_start() -> : 
     () -> : 
     () -> : 
     () -> : 
     () -> : 
     () -> : 

---

MemoryManager

CORE_RESERVED_TOOLS = {
    "read_file", "write_file", "terminal", "bash", "web_search",
    "web_fetch", "grep", "glob", "memory_recall",
}

class MemoryManager:
    def __init__(self):
        self._providers:        list[MemoryProvider] = []
        self._tool_to_provider: dict[str, MemoryProvider] = {}
        self._external_count:   int = 0
        self._sync_executor     = ThreadPoolExecutor(max_workers=1)

    def add_provider(self, provider: MemoryProvider, is_external: bool = False) -> None:
        if is_external and self._external_count >= 1:
            raise ValueError(
                f"Only 1 external memory provider allowed. "
                f"Already registered: {[p.name for p in self._providers if p != provider]}"
            )
        for tool in provider.tool_names:
            if tool in CORE_RESERVED_TOOLS:
                raise ValueError(f"Provider '' shadows reserved tool ''")
             tool  ._tool_to_provider:
                 ValueError()
            ._tool_to_provider[tool] = provider

        ._providers.append(provider)
         is_external:
            ._external_count += 

     () -> :
        
        blocks = [p.build_system_prompt_block()  p  ._providers]
         .join(b  b  blocks  b)

     () -> :
        
         provider  ._providers:
            :
                provider.prefetch(messages)
             Exception  e:
                ()

     () -> :
        
         provider  ._providers:
            ._sync_executor.submit(._safe_sync, provider, messages)

     () -> :
        :
            provider.sync(messages)
         Exception  e:
            ()

     () -> :
        
        provider = ._tool_to_provider.get(tool_name)
          provider:
             KeyError()
         provider.handle_tool(tool_name, args)

     () -> :
         provider  ._providers:
            provider.on_session_end()
        ._sync_executor.shutdown(wait=, timeout=drain_timeout)

---

Example: Built-in + external vector store

from mem0 import MemoryClient   # external provider example

class Mem0Provider(MemoryProvider):
    name = "mem0"
    tool_names = ["mem0_recall", "mem0_save"]

    def __init__(self, client: MemoryClient, user_id: str):
        self._client  = client
        self._user_id = user_id
        self._context = ""

    def prefetch(self, messages):
        last_user = next((m["content"] for m in reversed(messages)
                          if m["role"] == "user"), "")
        results    = self._client.search(last_user, user_id=self._user_id, limit=5)
        self._context = "\n".join(r["memory"] for r in results)

    def build_system_prompt_block(self) -> str:
        if not self._context:
            return ""
        return (
            "<memory-context>\n"
            
            
            
        )


manager = MemoryManager()
manager.add_provider(BuiltinL1Provider())
manager.add_provider(Mem0Provider(client, user_id=), is_external=)

---

Anti-Fake-Pass Checklist

❌ sync_all() called synchronously (await) in turn path — defeats background isolation
❌ Two external providers registered — tool schema doubles; second registration raises
❌ Provider shadowing a core tool name — memory_recall hijacked by external backend
❌ No drain timeout on shutdown — process hangs if background worker stalls
❌ prefetch errors swallowed silently without logging — memory silently empty