"""Real Hermes + Mem0 SDK + local Qdrant; only the model API is simulated. HERMES_SOURCE=/path/to/hermes-agent python tests/smoke_hermes.py Requires the Hermes dependencies, mem0ai, and qdrant-client in the active environment. All configuration, credentials, and database files are temporary. """ import importlib import io import json import os import shutil import sys import tempfile import threading from contextlib import redirect_stdout from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from unittest.mock import patch class ModelAPI(BaseHTTPRequestHandler): calls = [] def log_message(self, *args): pass def do_POST(self): if self.headers.get("Authorization") != "Bearer local-test": self.send_error(401) return body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) self.calls.append(self.path) if self.path == "/v1/embeddings": result = { "object": "list", "data": [{"object": "embedding", "index": i, "embedding": [1.0, 0.0, 0.0]} for i, _ in enumerate(body["input"])], "model": "test-embedding", "usage": {"prompt_tokens": 1, "total_tokens": 1}, } elif self.path == "/v1/chat/completions": result = { "id": "test-completion", "object": "chat.completion", "created": 0, "model": "gpt-4.1-mini", "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": json.dumps({"facts": ["Prefers green tea."], "memory": [ {"id": "0", "text": "Prefers green tea.", "event": "ADD"}]})}}], } else: self.send_error(404) return encoded = json.dumps(result).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def main(): hermes_source = Path(os.environ["HERMES_SOURCE"]).resolve() sys.path.insert(0, str(hermes_source)) with tempfile.TemporaryDirectory() as temporary: home = Path(temporary) os.environ.update(HERMES_HOME=temporary, MEM0_DIR=str(home / "mem0-data"), MEM0_TELEMETRY="false") (home / "mem0-data").mkdir() server = ThreadingHTTPServer(("127.0.0.1", 0), ModelAPI) worker = threading.Thread(target=server.serve_forever, daemon=True) worker.start() try: url = f"http://127.0.0.1:{server.server_port}/v1" config = { "mode": "oss", "user_id": "existing-user", "agent_id": "hermes", "oss": { "llm": {"provider": "openai", "config": { "model": "gpt-4.1-mini", "api_key": "local-test", "openai_base_url": url}}, "embedder": {"provider": "openai", "config": { "model": "test-embedding", "embedding_dims": 3, "api_key": "local-test", "openai_base_url": url}}, "vector_store": {"provider": "qdrant", "config": {"path": str(home / "qdrant")}}, }, } (home / "mem0.json").write_text(json.dumps(config)) destination = home / "plugins" / "mem0" shutil.copytree(Path(__file__).resolve().parents[1], destination) import plugins.memory as memory_plugins from agent.memory_manager import MemoryManager # Simulate bundled-provider removal only inside this test process. memory_plugins._MEMORY_PLUGINS_DIR = home / "empty-bundled" memory_plugins._MEMORY_PLUGINS_DIR.mkdir() assert memory_plugins.find_provider_dir("mem0") == destination provider = memory_plugins.load_memory_provider("mem0", register_skills=False) assert provider is not None and provider.__class__.__module__.startswith("_hermes_user_memory.") setup = importlib.import_module(f"{provider.__class__.__module__}._setup") from hermes_cli.memory_setup import cmd_setup_provider, cmd_status output = io.StringIO() arguments = ["hermes", "memory", "setup", "mem0", "--mode", "platform", "--api-key", "local-test"] with patch.object(sys, "argv", arguments), patch.object(sys, "stdin", io.StringIO("\n\n")): with patch.object(setup, "_curses_select", return_value=0), redirect_stdout(output): cmd_setup_provider("mem0") assert json.loads((home / "mem0.json").read_text())["user_id"] == "existing-user" provider.save_config(config, home) assert (home / "mem0.json").stat().st_mode & 0o777 == 0o600 with redirect_stdout(output): cmd_status(None) assert "installed ✓" in output.getvalue() and "available ✓" in output.getvalue() assert (home / ".env").stat().st_mode & 0o777 == 0o600 manager = MemoryManager() manager.add_provider(provider) manager.initialize_all("smoke-session", platform="cli", user_id="gateway-user") assert provider._backend is not None, getattr(provider, "_init_error", "initialization failed") def tool(name, **arguments): result = json.loads(manager.handle_tool_call(name, arguments)) assert "error" not in result, result return result try: assert tool("mem0_add", content="Prefers Python.")["result"] == "Fact stored." memories = tool("mem0_search", query="language")["results"] assert len(memories) == 1 and memories[0]["memory"] == "Prefers Python." memory_id = memories[0]["id"] tool("mem0_update", memory_id=memory_id, text="Prefers Rust.") assert "Prefers Rust." in provider.prefetch("language preference") assert tool("mem0_search", query="language")["results"][0]["memory"] == "Prefers Rust." tool("mem0_delete", memory_id=memory_id) assert tool("mem0_search", query="language")["result"] == "No relevant memories found." manager.sync_all("I prefer green tea.", "Noted.", session_id="smoke-session") assert provider._user_id == "existing-user" finally: manager.shutdown_all() assert provider._backend is None # A changed embedder must fail without destroying the existing collection. config["oss"]["embedder"]["config"]["embedding_dims"] = 4 provider.save_config(config, home) mismatched = type(provider)() mismatched.initialize("mismatched-session") assert mismatched._backend is None and "Existing memories were preserved" in mismatched._init_error config["oss"]["embedder"]["config"]["embedding_dims"] = 3 # Resolve embedder credentials from the active profile, not another profile's process env. del config["oss"]["embedder"]["config"]["api_key"] del config["oss"]["embedder"]["config"]["openai_base_url"] provider.save_config(config, home) from agent.secret_scope import ( reset_secret_scope, set_multiplex_active, set_secret_scope, ) resumed = type(provider)() with patch.dict(os.environ, {"OPENAI_API_KEY": "wrong-profile", "OPENAI_API_BASE": "http://127.0.0.1:1/v1"}): set_multiplex_active(True) token = set_secret_scope({"OPENAI_API_KEY": "local-test", "OPENAI_BASE_URL": url}) try: resumed.initialize("resumed-session", user_id="different-gateway-user") finally: reset_secret_scope(token) set_multiplex_active(False) try: assert resumed._backend is not None found = json.loads(resumed.handle_tool_call("mem0_search", {"query": "drink"})) assert "results" in found, found assert found["results"][0]["memory"] == "Prefers green tea." finally: resumed.shutdown() assert "/v1/chat/completions" in ModelAPI.calls print("PASS: external Hermes loader, CLI setup/status, real Mem0/Qdrant CRUD, recall, background extraction,") print(" existing identity, profile credentials, private files, shutdown, dimension safety and persistence across restart. Model responses simulated locally.") finally: server.shutdown() server.server_close() worker.join(timeout=5) if __name__ == "__main__": main()