Hermes Plugin

Copydef _load_config() -> dict:
    """Env vars provide defaults; $HERMES_HOME/mem0.json overrides individual keys.
    Layering avoids a silent failure when the JSON file exists but lacks fields
    like ``api_key`` that the user set in ``.env``."""
    from hermes_constants import get_hermes_home
 
    # Identity (user/agent id), host and mode are .env values like the key: read them through the
    # profile scope too, or a secondary profile's memories land in the default profile's account.
    # A scope-less multiplex caller raises here on purpose — that is a spawn-site bug, and
    # swallowing it would silently route the turn's memories to the default profile.
    config = {"mode": get_secret("MEM0_MODE", "") or "platform", "host": get_secret("MEM0_HOST", "") or "",
              "agent_id": get_secret("MEM0_AGENT_ID", "") or "hermes", "oss": {}}
    if user_id := get_secret("MEM0_USER_ID", ""):  # only when explicitly configured, so initialize() can fall back to the gateway-native id
        config["user_id"] = user_id
    file_cfg = read_json_or_empty(get_hermes_home() / "mem0.json")
    config.update({k: v for k, v in file_cfg.items() if v is not None and v != ""})
    # MEM0_API_KEY authenticates the Platform and self-hosted HTTP backends; pure OSS mode builds its
    # backend from the local ``oss`` config and has no platform credential to resolve, so a profile
    # scope WITHOUT the key must still load an OSS config (#99121 as it stands today: the caller is
    # scoped, the scope is just empty). Decided after mem0.json overrode the env defaults because
    # the file may be what selects ``oss``. Scope-less callers already raised above.
    if config.get("mode", "platform") == "oss":
        config.setdefault("api_key", "")
    elif not config.get("api_key"):
        config["api_key"] = get_secret("MEM0_API_KEY", "")
    return config

Copydef initialize(self, session_id: str, **kwargs) -> None:
    self._config = cfg = _load_config()
    self._mode, self._api_key, self._host, self._agent_id = cfg.get("mode", "platform"), cfg.get("api_key", ""), cfg.get("host", ""), cfg.get("agent_id", "hermes")
    # user_id precedence: operator-configured (env/mem0.json) > gateway-native id (kwargs) > _DEFAULT_USER_ID.
    # The literal placeholder counts as unset so wizard users still get gateway-native ids.
    configured = cfg.get("user_id")
    self._user_id = (None if configured == _DEFAULT_USER_ID else configured) or kwargs.get("user_id") or _DEFAULT_USER_ID
    # Persisted rerank preference: default for mem0_search when the model omits ``rerank``. Platform-only.
    _rr = cfg.get("rerank", False)
    self._rerank_default = _rr.lower() in ("true", "1", "yes") if isinstance(_rr, str) else bool(_rr)
    self._channel = kwargs.get("platform") or "cli"
    try:
        self._sync_max_chars = int(cfg.get("sync_max_chars") or _SYNC_MSG_MAX_CHARS)
    except (ValueError, TypeError):
        self._sync_max_chars = _SYNC_MSG_MAX_CHARS
    self._backend = self._create_backend()
    if self._backend and not self._atexit_registered:
        atexit.register(self.shutdown)
        self._atexit_registered = True
 

Copydef on_turn_start(self, turn_number: int, message: str, **kwargs) -> None:
    self._start_prefetch(message)
 
def _consume_prefetch_result(self, query: str) -> str | None:
    """Pop the finished prefetch body for ``query`` (None if absent or still running)."""
    with self._prefetch_lock:
        if self._prefetch_query != query or not self._prefetch_done:
            return None
        result, self._prefetch_result, self._prefetch_done = self._prefetch_result, "", False
        return result
 
def _start_prefetch(self, query: str) -> None:
    backend = self._backend
    if not query or backend is None or self._is_breaker_open():
        return
 
    def _run():
        results = self._try(lambda: self._search(query, rerank=self._rerank_default, backend=backend), logger.debug, "Mem0 prefetch failed: %s")
        lines = [r.get("memory", "") for r in (results or []) if r.get("memory")]
        body = "## Mem0 Memory\n" + "\n".join(f"- {line}" for line in lines) if lines else ""
        with self._prefetch_lock:
            if self._prefetch_query == query:
                self._prefetch_result, self._prefetch_done = body, True
 
    with self._prefetch_lock:
        if self._prefetch_query == query and (self._prefetch_done or (self._prefetch_thread and self._prefetch_thread.is_alive())):
            return
        self._prefetch_query, self._prefetch_result, self._prefetch_done = query, "", False
        self._prefetch_thread = spawn_context_thread(_run, name="mem0-prefetch")
        self._prefetch_thread.start()
 
def prefetch(self, query: str, *, session_id: str = "") -> str:
    """Recall memories for the CURRENT question with a short hot-path wait."""
    if (cached := self._consume_prefetch_result(query)) is not None:
        return cached
    self._start_prefetch(query)
    with self._prefetch_lock:
        thread = self._prefetch_thread if self._prefetch_query == query else None
    if thread:
        thread.join(timeout=_PREFETCH_WAIT_SECS)
    return self._consume_prefetch_result(query) or ""  # slow backend: skip injection; mem0_search remains the backstop
 

Copy# Sentence ends recognized when trimming a synced message. Deliberately unordered:
# the LAST boundary of ANY kind wins, so one CJK stop early in a mixed-script turn
# cannot outrank a Latin stop near the end of the window. ``".\n"`` is not listed —
# its index can never exceed the bare ``"."`` it starts with.
_SYNC_SENTENCE_ENDS = ("。", "！", "？", ".", "!", "?")
 
 
def _truncate_for_sync(text: str, max_len: int = _SYNC_MSG_MAX_CHARS) -> str:
    """Cap a synced message at its last sentence boundary within ``max_len``.
 
    Short messages pass through unchanged; long ones keep the last complete
    sentence inside the window so fact extraction still sees coherent statements,
    with a hard cut as fallback when no boundary exists (or one only appears in
    the first third of the window, which usually means unsegmented input).
    """
    if len(text) <= max_len:
        return text
    window = text[:max_len]
    cut = max(window.rfind(sep) for sep in _SYNC_SENTENCE_ENDS)
    if cut > max_len // 3:
        return text[:cut + 1]
    return text[:max_len]
 

Copydef sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
    """Send the turn to Mem0 for server-side fact extraction (non-blocking)."""
    if self._backend is None or self._is_breaker_open():
        return
 
    def _sync():
        if self._backend is not None:
            messages = [
                {"role": "user", "content": _truncate_for_sync(user_content, self._sync_max_chars)},
                {"role": "assistant", "content": _truncate_for_sync(assistant_content, self._sync_max_chars)},
            ]
            self._try(lambda: self._add(messages, infer=True), logger.warning, "Mem0 sync failed: %s")
 
    with self._sync_lock:
        prev = self._sync_thread
    if prev and prev.is_alive():
        prev.join(timeout=5.0)
        if prev.is_alive():
            return
    with self._sync_lock:
        self._sync_thread = spawn_context_thread(_sync, name="mem0-sync")
        self._sync_thread.start()
 

Copy_TOOL_HANDLERS = {
    "mem0_search": (("query",), "Search failed", _tool_search, "skip"),
    "mem0_add": (("content",), "Failed to store", _tool_add, "count"),
    "mem0_update": (("memory_id", "text"), "Update failed", _tool_update, "not_found"),
    "mem0_delete": (("memory_id",), "Delete failed", _tool_delete, "not_found"),
}

Copydef _ensure_owns_memory(self, memory_id: str) -> None:
    """Reject a mutation on a memory that doesn't belong to the caller's user_id."""
    memory = self._backend.get(memory_id)
    if not memory:
        raise ValueError(f"Memory not found: {memory_id}")
    owner = memory.get("user_id") if isinstance(memory, dict) else None
    if owner and owner != self._user_id:
        raise PermissionError(f"Memory {memory_id} does not belong to this user.")
 
def _tool_update(self, args: dict) -> str:
    self._ensure_owns_memory(args["memory_id"])
    return json.dumps(self._backend.update(args["memory_id"], args["text"]))
 
def _tool_delete(self, args: dict) -> str:
    self._ensure_owns_memory(args["memory_id"])
    return json.dumps(self._backend.delete(args["memory_id"]))
 

Copy# Circuit breaker: after _BREAKER_THRESHOLD consecutive failures, pause API
# calls for _BREAKER_COOLDOWN_SECS to avoid hammering a down server.
_BREAKER_THRESHOLD, _BREAKER_COOLDOWN_SECS, _PREFETCH_WAIT_SECS = 5, 120, 3
_CLIENT_ERROR_TYPES = ("MemoryNotFoundError", "ValidationError")
# Placeholder user_id. initialize() treats it as "no operator-configured user_id"
# so legacy mem0.json files written by the wizard don't override gateway-native ids.
_DEFAULT_USER_ID = "hermes-user"

Copyclass Mem0Backend(ABC):
    """Unified interface over Platform (MemoryClient), self-hosted (HTTP) and OSS (Memory) backends.
    update()/delete() are template methods: subclasses implement raw ``_update``/``_delete``."""
 
    @abstractmethod
    def search(self, query: str, *, filters: dict, top_k: int = 10, rerank: bool = False) -> list[dict]: ...
    @abstractmethod
    def add(self, messages: list, *, user_id: str, agent_id: str, infer: bool = False, metadata: dict | None = None) -> dict: ...
    @abstractmethod
    def get(self, memory_id: str) -> dict | None: ...
    @abstractmethod
    def _update(self, memory_id: str, text: str) -> None: ...
    @abstractmethod
    def _delete(self, memory_id: str) -> None: ...
 

Copyclass PlatformBackend(Mem0Backend):
    """Wraps mem0.MemoryClient for Mem0 Platform (cloud API)."""
 
    def __init__(self, api_key: str):
        from mem0 import MemoryClient
        self._client = MemoryClient(api_key=api_key)
 
    def search(self, query: str, *, filters: dict, top_k: int = 10, rerank: bool = False) -> list[dict]:
        return _unwrap_results(self._client.search(query, filters=filters, top_k=top_k, rerank=rerank))
 
    def add(self, messages: list, *, user_id: str, agent_id: str, infer: bool = False, metadata: dict | None = None) -> dict:
        return self._client.add(messages, **_add_kwargs(user_id, agent_id, infer, metadata))
 
    def get(self, memory_id: str) -> dict | None:
        return self._client.get(memory_id)
 
    def _update(self, memory_id: str, text: str) -> None:
        self._client.update(memory_id=memory_id, text=text)
 
    def _delete(self, memory_id: str) -> None:
        self._client.delete(memory_id=memory_id)
 

Copyclass SelfHostedBackend(Mem0Backend):
    """Direct HTTP backend for a self-hosted Mem0 server (the FastAPI ``server/``).
    mem0.MemoryClient is hardwired to the cloud API (``Authorization: Token``, ``GET /v1/ping/`` in ``__init__``),
    so this speaks the server's real contract: ``X-API-Key`` auth and the ``/memories`` / ``/search`` routes."""
 
    def __init__(self, api_key: str, host: str, transport=None):
        import httpx
        headers = {"Content-Type": "application/json", **({"X-API-Key": api_key} if api_key else {})}  # key omitted only for AUTH_DISABLED servers
        # Connect-level retries keep one dropped SYN from counting toward the breaker. ``transport`` is injectable for tests.
        self._client = httpx.Client(base_url=host.rstrip("/"), headers=headers, timeout=30.0, transport=transport or httpx.HTTPTransport(retries=2))
        self._capture_timeout = httpx.Timeout(120.0, connect=30.0)
 
    def _json(self, method: str, path: str, **kwargs) -> Any:
        resp = self._client.request(method, path, **kwargs)
        resp.raise_for_status()
        return resp.json() if resp.content else {}
 
    def search(self, query: str, *, filters: dict, top_k: int = 10, rerank: bool = False) -> list[dict]:
        # rerank is platform-only; the self-hosted /search ignores it. user_id belongs in filters (top-level is deprecated).
        return _unwrap_results(self._json("POST", "/search", json={"query": query, "top_k": top_k, **({"filters": filters} if filters else {})}))
 
    def add(self, messages: list, *, user_id: str, agent_id: str, infer: bool = False, metadata: dict | None = None) -> dict:
        # Server-side extraction takes longer than a search or verbatim write.
        return self._json("POST", "/memories", json={"messages": messages, **_add_kwargs(user_id, agent_id, infer, metadata)},
                          timeout=self._capture_timeout if infer else self._client.timeout)
 
    def get(self, memory_id: str) -> dict | None:
        return self._json("GET", f"/memories/{memory_id}")
 
    def _update(self, memory_id: str, text: str) -> None:
        self._json("PUT", f"/memories/{memory_id}", json={"text": text})
 
    def _delete(self, memory_id: str) -> None:
        self._json("DELETE", f"/memories/{memory_id}")
 
    def close(self) -> None:
        with suppress(Exception):
            self._client.close()
 

Copyclass OSSBackend(Mem0Backend):
    """Wraps mem0.Memory for self-hosted (OSS) mode."""
 
    def __init__(self, oss_config: dict):
        from ._oss_providers import EMBEDDER_PROVIDERS, KNOWN_DIMS, LLM_PROVIDERS
 
        self._local_path = None
        self._owner = None
        self._lock = nullcontext()
        self._closed = False
 
        def _provider_block(name: str, registry: dict) -> dict:
            """Copy of oss_config[name] with the legacy ``api_base`` key mapped to the provider's canonical base-URL key."""
            block = dict(oss_config[name])
            provider_config = dict(block.get("config", {}))
            legacy_base = provider_config.pop("api_base", None)
            canonical_key = registry.get(str(block.get("provider") or "").strip().lower(), {}).get("base_url_key")
            if legacy_base and canonical_key:
                provider_config.setdefault(canonical_key, legacy_base)
            if str(block.get("provider") or "").strip().lower() == "openai":
                from agent.secret_scope import get_secret
 
                # Resolve profile secrets before comparing configurations for sharing.
                provider_config["api_key"] = provider_config.get("api_key") or get_secret("OPENAI_API_KEY", "")
                if not provider_config["api_key"]:
                    raise ValueError(f"OpenAI API key is required for the Hermes Mem0 OSS {name}")
                provider_config["openai_base_url"] = (
                    provider_config.get("openai_base_url") or get_secret("OPENAI_API_BASE", "")
                    or get_secret("OPENAI_BASE_URL", "") or "https://api.openai.com/v1"
                )
            block["config"] = provider_config
            return block
 
        vector_store = dict(oss_config["vector_store"])
        vs_config = dict(vector_store.get("config", {}))
        if vs_config.get("path"):
            vs_config["path"] = os.path.expanduser(vs_config["path"])
        embedder_config = oss_config.get("embedder", {}).get("config", {})
        dims = embedder_config.get("embedding_dims") or KNOWN_DIMS.get(embedder_config.get("model", ""))
        if dims:
            vs_config["embedding_model_dims"] = dims
        remote = (vs_config.get("host") and vs_config.get("port")) or vs_config.get("url") or vs_config.get("api_key")
        if (vector_store.get("provider", "qdrant") == "qdrant" and not vs_config.get("client")
                and not remote and vs_config.get("https") is None):
            from mem0.configs.vector_stores.qdrant import QdrantConfig
            path = vs_config.get("path", QdrantConfig.model_fields["path"].default)
            if path:
                self._local_path = vs_config["path"] = os.path.realpath(os.path.expanduser(path))
        vector_store["config"] = vs_config
        config = {"vector_store": vector_store, "llm": _provider_block("llm", LLM_PROVIDERS), "embedder": _provider_block("embedder", EMBEDDER_PROVIDERS), "version": "v1.1"}
        if self._local_path:
            from hermes_constants import get_hermes_home
            profile = os.path.realpath(get_hermes_home())
            with _LOCAL_QDRANT_LOCK:
                owner = _LOCAL_QDRANT_MEMORIES.get(self._local_path)
                if owner is None:
                    owner = _LocalQdrantMemory(self._create_memory(config, dims), deepcopy(config), profile)
                    _LOCAL_QDRANT_MEMORIES[self._local_path] = owner
                else:
                    if owner.profile != profile or owner.config != config:
                        raise ValueError("Local Qdrant storage is already open with a different profile or configuration. "
                                         "Existing memories were preserved. Close its active sessions before changing settings, "
                                         "or use a separate storage path.")
                    owner.users += 1
                self._owner, self._lock, self._memory = owner, owner.lock, owner.memory
        else:
            self._memory = self._create_memory(config, dims)
 

Copydef _detect_current_dims(provider: str, vs_config: dict, collection_name: str) -> int | None:
    """Current embedding dimension of ``collection_name``, or None if it doesn't exist yet.
    Raises on any failure to connect/inspect so the caller can decide whether to skip the guard."""
    if provider == "qdrant":
        from qdrant_client import QdrantClient
        path, url, host = vs_config.get("path"), vs_config.get("url"), vs_config.get("host")
        if path:
            client = QdrantClient(path=path)
        elif url:
            client = QdrantClient(url=url, api_key=vs_config.get("api_key"))
        elif host:
            client = QdrantClient(host=host, port=vs_config.get("port") or 6333, api_key=vs_config.get("api_key"))
        else:
            return None
        with closing(client):
            if not client.collection_exists(collection_name):
                return None
            vectors = client.get_collection(collection_name).config.params.vectors
            # Named-vector collections expose a dict; unnamed expose an object with .size.
            if isinstance(vectors, dict):
                vectors = next(iter(vectors.values()), None)
            return getattr(vectors, "size", None)
    elif provider == "pgvector":
        import psycopg2
        conn_params = {k: vs_config[k] for k in ("host", "port", "user", "password", "dbname", "sslmode") if vs_config.get(k)}
        with closing(psycopg2.connect(**conn_params)) as conn:
            conn.autocommit = True
            with closing(conn.cursor()) as cur:
                cur.execute("SELECT atttypmod FROM pg_attribute WHERE attrelid = %s::regclass AND attname = 'vector'", (collection_name,))
                row = cur.fetchone()
                return row[0] if row and row[0] > 0 else None
    return None

Copydef _reject_dimension_mismatch(provider: str, vs_config: dict, expected_dims: int) -> None:
    """Reject embedding dimension changes without deleting existing memories."""
    collection_name = vs_config.get("collection_name", "mem0")
    try:
        current_dims = OSSBackend._detect_current_dims(provider, vs_config, collection_name)
    except Exception as dimension_detection_error:
        logger.warning(
            "Could not determine embedding dimensions for collection %r (%s): %s. Skipping dimension-change guard.",
            collection_name, provider, dimension_detection_error,
        )
        return
    if current_dims is not None and current_dims != expected_dims:
        raise ValueError(
            f"Collection {collection_name!r} has {current_dims} embedding dimensions, but {expected_dims} are configured. "
            "Existing memories were preserved. Restore the previous embedder or use a new collection_name."
        )
 

Copy@dataclass
class _LocalQdrantMemory:
    memory: Any
    config: dict
    profile: str
    lock: Any = field(default_factory=RLock)
    users: int = 1
 
 
_LOCAL_QDRANT_MEMORIES: dict[str, _LocalQdrantMemory] = {}
_LOCAL_QDRANT_LOCK = RLock()
 

Copyif self._local_path:
    from hermes_constants import get_hermes_home
    profile = os.path.realpath(get_hermes_home())
    with _LOCAL_QDRANT_LOCK:
        owner = _LOCAL_QDRANT_MEMORIES.get(self._local_path)
        if owner is None:
            owner = _LocalQdrantMemory(self._create_memory(config, dims), deepcopy(config), profile)
            _LOCAL_QDRANT_MEMORIES[self._local_path] = owner
        else:
            if owner.profile != profile or owner.config != config:
                raise ValueError("Local Qdrant storage is already open with a different profile or configuration. "
                                 "Existing memories were preserved. Close its active sessions before changing settings, "
                                 "or use a separate storage path.")
            owner.users += 1
        self._owner, self._lock, self._memory = owner, owner.lock, owner.memory
else:
    self._memory = self._create_memory(config, dims)
 

Copyclass DirectOpenAILLM(OpenAILLM):
    """Use OpenAI credentials and requests regardless of router environment."""
 
    def __init__(self, config: Optional[Union[BaseLlmConfig, OpenAIConfig, Dict]] = None):
        if config is None:
            config = OpenAIConfig()
        elif isinstance(config, dict):
            config = OpenAIConfig(**config)
        elif isinstance(config, BaseLlmConfig) and not isinstance(config, OpenAIConfig):
            fields = {k: getattr(config, k) for k in _COPIED_FIELDS}
            fields.update({k: getattr(config, k, None) for k in _OPTIONAL_FIELDS})
            config = OpenAIConfig(**fields)

Copyif not config.model:
    config.model = "gpt-5-mini"
# Configs predating the setup marker: keep the default model reasoning-safe
# without overriding an explicit user choice.
if config.model == "gpt-5-mini" and config.is_reasoning_model is None:
    config.is_reasoning_model = True
# Bypass OpenAILLM.__init__ (it picks OpenRouter when OPENROUTER_API_KEY is
# set); LLMBase still owns validation and supported-parameter filtering.
LLMBase.__init__(self, config)
# OPENAI_API_KEY / OPENAI_BASE_URL are profile credentials: read them through the secret
# scope, never raw os.environ, or a multiplexed secondary's memory extraction runs on the
# default profile's OpenAI account (and its proxy).
from agent.secret_scope import get_secret
api_key = self.config.api_key or get_secret("OPENAI_API_KEY", "")
if not api_key:
    raise ValueError("OpenAI API key is required for the Hermes Mem0 OSS provider")
from openai import OpenAI
self.client = OpenAI(api_key=api_key, base_url=self.config.openai_base_url or get_secret("OPENAI_API_BASE", "") or get_secret("OPENAI_BASE_URL", "") or "https://api.openai.com/v1", timeout=120.0)
 

Copy# BaseLlmConfig fields copied into OpenAIConfig; the last two may be absent on older mem0.
_COPIED_FIELDS = ("model", "temperature", "api_key", "max_tokens", "top_p", "top_k", "enable_vision", "vision_details", "http_client_proxies")
_OPTIONAL_FIELDS = ("reasoning_effort", "is_reasoning_model")
 

Copy# OPENAI_API_KEY / OPENAI_BASE_URL are profile credentials: read them through the secret
# scope, never raw os.environ, or a multiplexed secondary's memory extraction runs on the
# default profile's OpenAI account (and its proxy).
from agent.secret_scope import get_secret
api_key = self.config.api_key or get_secret("OPENAI_API_KEY", "")
if not api_key:
    raise ValueError("OpenAI API key is required for the Hermes Mem0 OSS provider")
from openai import OpenAI
self.client = OpenAI(api_key=api_key, base_url=self.config.openai_base_url or get_secret("OPENAI_API_BASE", "") or get_secret("OPENAI_BASE_URL", "") or "https://api.openai.com/v1", timeout=120.0)

Copydef generate_response(self, messages: List[Dict[str, str]], response_format=None, tools: Optional[List[Dict]] = None, tool_choice: str = "auto", **kwargs):
    params = self._get_supported_params(messages=messages, **kwargs)
    params.update({"model": self.config.model, "messages": messages})
    # No OpenRouter-only fields; ``store`` is opt-in so OpenAI-compatible endpoints never receive unknown fields.
    if self.config.store is not None:
        params["store"] = self.config.store
    if response_format:
        params["response_format"] = response_format
    if tools:
        params["tools"], params["tool_choice"] = tools, tool_choice
    response = self.client.chat.completions.create(**params)
    parsed_response = self._parse_response(response, tools)
    if self.config.response_callback:
        try:
            self.config.response_callback(self, response, params)
        except Exception:
            logging.error("Error running Mem0 OpenAI response callback")
    return parsed_response

CopyKNOWN_DIMS: dict[str, int] = {"text-embedding-3-small": 1536, "text-embedding-3-large": 3072, "text-embedding-ada-002": 1536, "nomic-embed-text": 768}

Copydef validate_oss_config(oss_config: dict) -> list[str]:
    """Validate an OSS config dict. Returns list of error strings (empty = valid)."""
    errors: list[str] = []
    for section, registry in SECTION_REGISTRIES:
        block = oss_config.get(section)
        if not block or not isinstance(block, dict):
            errors.append(f"Missing required section: {section}")
        elif block.get("provider", "") not in registry:
            errors.append(f"Unknown {section} provider '{block.get('provider', '')}'. Valid: {', '.join(registry.keys())}")
    vs = oss_config.get("vector_store", {})
    if vs.get("provider") == "pgvector" and not vs.get("config", {}).get("user"):
        errors.append("PGVector requires 'user' in vector_store.config")
    return errors

Copydef vector_default_config(provider_id: str) -> dict[str, Any]:
    """A vector store's ``default_config`` with callable defaults resolved for the active profile."""
    return {k: (v() if callable(v) else v) for k, v in VECTOR_PROVIDERS[provider_id]["default_config"].items()}
 

Copydef _ensure_pgvector_extension(pg_config: dict) -> None:
    try:
        import psycopg2
    except ImportError:
        return
    defaults = {"host": "localhost", "port": 5432, "user": "postgres", "dbname": "postgres"}
    try:
        conn = psycopg2.connect(**(defaults | {k: v for k, v in pg_config.items() if k in defaults or (k == "password" and v)}))
        conn.autocommit = True
        conn.cursor().execute("CREATE EXTENSION IF NOT EXISTS vector")
        conn.close()
        print("  ✓ pgvector extension enabled")
    except Exception as e:
        print(f"  Warning: Could not enable pgvector extension: {e}")
 

Copy_MODE_HANDLERS = {"oss": _setup_oss, "selfhosted": _setup_selfhosted, "self-hosted": _setup_selfhosted, "platform": _setup_platform}
# Interactive picker order: Platform, Self-hosted server, Open Source.
_MODE_ITEMS = [("Platform", "Mem0 Cloud API (lightweight, just needs an API key)"), ("Self-hosted server", "Connect to an existing self-hosted Mem0 server (Docker/FastAPI)"), ("Open Source", "Run Mem0 locally (self-hosted LLM + vector store)")]
_MODE_PICKER = (_setup_platform, _setup_selfhosted, _setup_oss)
 
 
def post_setup(hermes_home: str, config: dict) -> None:
    """Entry point for `hermes memory setup`: routes on --mode (platform / selfhosted / oss), else shows a picker.
    OSS is non-interactive only when the mode came from the flag."""
    with suppress(ImportError):  # mem0ai must meet the minimum version from plugin.yaml
        import mem0
        installed_ver = getattr(mem0, "__version__", None)
        if installed_ver and _version_tuple(installed_ver) < (2, 0, 10):
            print(f"\n  ⚠ mem0ai {installed_ver} installed but >=2.0.10 required.\n  Run: uv pip install --python {sys.executable} 'mem0ai>=2.0.10'")
    flags = parse_flags(sys.argv[1:])
    handler = _MODE_HANDLERS.get(flags["mode"])
    flags["_mode_from_flag"] = handler is not None
    if handler is None:
        handler = _MODE_PICKER[_curses_select("  Select mode", _MODE_ITEMS, 0)]
    handler(hermes_home, config, flags)
 

Copyif os.environ.get("MEM0_HOST", "").strip():
    print(f"\n  ⚠ MEM0_HOST is set in your environment ({os.environ['MEM0_HOST']}). It overrides platform mode — remove it from ~/.hermes/.env (or unset it) or Hermes will keep routing to the self-hosted server.")
_persist_provider_config(hermes_home, config, provider_config, env_writes, "mem0", "API keys saved to .env")

Copydef _write_env(env_path: Path, env_writes: dict[str, str]) -> None:
    env_path.parent.mkdir(parents=True, exist_ok=True)
    # utf-8-sig like the canonical .env readers: a BOM'd first line would miss the key match and get duplicated.
    existing_lines = env_path.read_text(encoding="utf-8-sig").splitlines() if env_path.exists() else []
    keys = [line.split("=", 1)[0].strip() if "=" in line and not line.startswith("#") else None for line in existing_lines]
    new_lines = [f"{k}={env_writes[k]}" if k in env_writes else line for k, line in zip(keys, existing_lines)]
    new_lines += [f"{k}={v}" for k, v in env_writes.items() if k not in keys]
    from utils import atomic_write_text
 
    atomic_write_text(env_path, "\n".join(new_lines) + "\n", mode=0o600)
 

Copy_PGVECTOR_CONTAINER, _PGVECTOR_IMAGE = "hermes-pgvector", "pgvector/pgvector:pg17"
…
def _ensure_pgvector(host: str = "localhost", port: int = 5432) -> dict | None:
    """Ensure pgvector is reachable, offering Docker if not; returns the started container's vector_config, else None."""
    if _check_pgvector(host, port)[0]:
        print(f"  ✓ PostgreSQL reachable at {host}:{port}")
        return None
    print(f"  PostgreSQL not reachable at {host}:{port}")
    if not shutil.which("docker"):
        print("  Docker not found. Install Docker to auto-start pgvector,\n  or run PostgreSQL with pgvector manually.")
        return None
    with suppress(Exception):  # restart our own container if it exists but is stopped
        result = _docker("inspect", _PGVECTOR_CONTAINER, "--format", "{{.State.Status}}", timeout=10, text=True, encoding='utf-8', errors='replace')
        if result.returncode == 0 and "exited" in result.stdout:
            print(f"  Found stopped container '{_PGVECTOR_CONTAINER}', restarting...")
            _docker("start", _PGVECTOR_CONTAINER, timeout=15)
            if _pg_ready(host, port, 15):
                print("  ✓ PostgreSQL container restarted")
                return None
    if input("  Start pgvector via Docker? [Y/n]: ").strip().lower() not in ("", "y", "yes"):
        print("  Skipping Docker setup. Make sure PostgreSQL with pgvector is running.")
        return None
    password = secrets.token_urlsafe(24)
    try:
        print(f"  Pulling {_PGVECTOR_IMAGE}...")
        _docker("pull", _PGVECTOR_IMAGE, timeout=120)
        print(f"  Starting container '{_PGVECTOR_CONTAINER}' on port {port}...")
        _docker("run", "-d", "--name", _PGVECTOR_CONTAINER, "-e", f"POSTGRES_PASSWORD={password}", "-p", f"127.0.0.1:{port}:5432", _PGVECTOR_IMAGE, timeout=30, check=True)
        if _pg_ready(host, port, 20):
            print(f"  ✓ pgvector running on {host}:{port}")
        else:
            print("  Warning: Container started but PostgreSQL not yet accepting connections.\n  It may need a few more seconds. Config will be saved; retry later.")
        return {"host": host, "port": port, "user": "postgres", "password": password, "dbname": "postgres"}
    except subprocess.CalledProcessError as e:
        print(f"  Failed to start Docker container: {_scrub(str(e), password)}")
    except Exception as e:
        print(f"  Docker error: {_scrub(str(e), password)}")
    return None