From 9c9b040d428205a729f6d87ea1985e71a2ffd63b Mon Sep 17 00:00:00 2001 From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com> Date: Fri, 17 Jul 2026 14:06:12 +0800 Subject: [PATCH] feat(plugins): add Hermes Agent memory provider (#365) * feat(plugins): add Hermes Agent memory provider * fix(plugins): harden Hermes memory lifecycle * fix(plugins): keep Hermes writer recoverable --- plugins/hermes_agent/README.md | 84 +++++ plugins/hermes_agent/__init__.py | 413 ++++++++++++++++++++++ plugins/hermes_agent/client.py | 76 ++++ plugins/hermes_agent/plugin.yaml | 4 + tests/unit/test_hermes_agent_plugin.py | 457 +++++++++++++++++++++++++ 5 files changed, 1034 insertions(+) create mode 100644 plugins/hermes_agent/README.md create mode 100644 plugins/hermes_agent/__init__.py create mode 100644 plugins/hermes_agent/client.py create mode 100644 plugins/hermes_agent/plugin.yaml create mode 100644 tests/unit/test_hermes_agent_plugin.py diff --git a/plugins/hermes_agent/README.md b/plugins/hermes_agent/README.md new file mode 100644 index 00000000..1d039c55 --- /dev/null +++ b/plugins/hermes_agent/README.md @@ -0,0 +1,84 @@ +# ReMe memory provider for Hermes Agent + +This plugin connects Hermes Agent to a running ReMe HTTP service. It recalls +relevant memory before each model call and records each completed turn through +ReMe's automatic memory job. + +## Prerequisites + +- Python 3.11 or newer +- A working Hermes Agent installation +- ReMe installed with its core dependencies +- One ReMe workspace and endpoint for each Hermes profile that should remain + isolated + +ReMe search currently covers one whole workspace. Pointing multiple Hermes +profiles at the same ReMe workspace therefore shares their recalled memory. Use +a separate ReMe workspace and endpoint when profiles must be isolated. + +## Start ReMe + +Start the HTTP service against a workspace dedicated to the active Hermes +profile: + +```bash +reme start \ + workspace_dir="$HOME/.reme-hermes-default" \ + service.backend=http \ + service.host=127.0.0.1 \ + service.port=2333 +``` + +ReMe needs a working LLM configuration for automatic memory extraction. Its +default search uses BM25, so embedding credentials are optional unless vector +retrieval is enabled. Keep the service running while Hermes is active. + +## Install and configure + +Hermes supports installing a plugin from a repository subdirectory: + +```bash +hermes plugins install agentscope-ai/ReMe/plugins/hermes_agent +hermes memory setup +``` + +Select `reme`, accept `http://127.0.0.1:2333` or enter the endpoint used above. +Setup calls ReMe `health_check` and only replaces an existing provider config +after the endpoint reports healthy. Then start a new Hermes session. +Configuration is stored in +`$HERMES_HOME/reme.json`, so every Hermes profile can point to its own ReMe +workspace. + +The file supports these optional settings: + +```json +{ + "endpoint": "http://127.0.0.1:2333", + "request_timeout": 600.0, + "recall_timeout": 5.0, + "health_timeout": 2.0, + "health_retry_seconds": 30.0, + "shutdown_timeout": 30.0, + "recall_limit": 5 +} +``` + +Run `hermes memory status` to check that the provider is installed and +configured. Starting a Hermes session performs a fresh endpoint health check. + +## Lifecycle and failure behavior + +- `prefetch` calls ReMe `search` and returns only its recalled text. Hermes wraps + that text in its protected memory-context block. +- `sync_turn` queues the completed user/assistant turn for a serial background + writer, which calls ReMe `auto_memory` with a filename-safe ID derived from + the Hermes profile and conversation. +- Cron, flush, and subagent contexts do not write conversational memory. +- A failed health check disables recall and recording until the retry cooldown + expires. Retrieval and recording failures use independent cooldowns, so one + action cannot disable the other while the ReMe service remains healthy. +- Recall uses its own short timeout so a slow ReMe search cannot stall the + Hermes model call for the longer automatic-memory timeout. +- `shutdown` gives queued writes a bounded drain interval; ReMe remains an + independently managed service. An idempotent process-exit hook uses the same + drain path when a Hermes surface does not call provider shutdown directly. diff --git a/plugins/hermes_agent/__init__.py b/plugins/hermes_agent/__init__.py new file mode 100644 index 00000000..8139cb12 --- /dev/null +++ b/plugins/hermes_agent/__init__.py @@ -0,0 +1,413 @@ +"""Hermes Agent memory provider backed by a running ReMe HTTP service.""" + +from __future__ import annotations + +import atexit +import hashlib +import json +import logging +import os +import queue +import re +import tempfile +import threading +import time + +from pathlib import Path +from typing import Any, Dict, List, Optional + +from agent.memory_provider import MemoryProvider + +from .client import ReMeHttpClient, ReMeServiceError + +logger = logging.getLogger(__name__) + +_CONFIG_FILENAME = "reme.json" +_DEFAULT_CONFIG: dict[str, Any] = { + "endpoint": "http://127.0.0.1:2333", + "request_timeout": 600.0, + "recall_timeout": 5.0, + "health_timeout": 2.0, + "health_retry_seconds": 30.0, + "shutdown_timeout": 30.0, + "recall_limit": 5, +} +_NON_FILENAME_CHARS = re.compile(r"[^A-Za-z0-9._-]+") + + +def _config_path(hermes_home: str | Path | None = None) -> Path: + if hermes_home is None: + from hermes_constants import get_hermes_home + + hermes_home = get_hermes_home() + return Path(hermes_home).expanduser() / _CONFIG_FILENAME + + +def _load_config(hermes_home: str | Path | None = None) -> dict[str, Any]: + config = dict(_DEFAULT_CONFIG) + path = _config_path(hermes_home) + if not path.is_file(): + return config + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + logger.warning("Unable to read ReMe provider config %s: %s", path, exc) + return config + if isinstance(loaded, dict): + config.update({key: value for key, value in loaded.items() if value is not None and value != ""}) + return config + + +def _positive_float(config: dict[str, Any], key: str) -> float: + try: + return max(0.1, float(config[key])) + except (KeyError, TypeError, ValueError): + return float(_DEFAULT_CONFIG[key]) + + +def _positive_int(config: dict[str, Any], key: str) -> int: + try: + return max(1, int(config[key])) + except (KeyError, TypeError, ValueError): + return int(_DEFAULT_CONFIG[key]) + + +def _slug(value: str, fallback: str, *, limit: int) -> str: + value = _NON_FILENAME_CHARS.sub("-", str(value or "").strip()).strip("-._") + return (value or fallback)[:limit] + + +def _scoped_session_id(profile_id: str, session_id: str) -> str: + """Create a readable, filename-safe ID without allowing scope collisions.""" + profile = str(profile_id or "default") + session = str(session_id or "session") + digest = hashlib.sha256(f"{profile}\0{session}".encode("utf-8")).hexdigest()[:12] + return f"hermes-{_slug(profile, 'default', limit=32)}-{_slug(session, 'session', limit=64)}-{digest}" + + +class ReMeMemoryProvider(MemoryProvider): + """Use ReMe for automatic cross-session recall and recording in Hermes.""" + + def __init__(self) -> None: + self._client: ReMeHttpClient | None = None + self._endpoint = str(_DEFAULT_CONFIG["endpoint"]) + self._recall_timeout = float(_DEFAULT_CONFIG["recall_timeout"]) + self._health_timeout = float(_DEFAULT_CONFIG["health_timeout"]) + self._health_retry_seconds = float(_DEFAULT_CONFIG["health_retry_seconds"]) + self._shutdown_timeout = float(_DEFAULT_CONFIG["shutdown_timeout"]) + self._recall_limit = int(_DEFAULT_CONFIG["recall_limit"]) + self._service_available = False + self._next_health_probe = 0.0 + self._next_recall_attempt = 0.0 + self._next_write_attempt = 0.0 + self._session_id = "" + self._profile_id = "default" + self._write_enabled = True + self._accept_writes = True + self._write_queue: queue.Queue[dict[str, Any] | None] = queue.Queue() + self._write_thread: threading.Thread | None = None + self._write_thread_lock = threading.Lock() + self._shutdown_started = False + self._atexit_registered = False + + @property + def name(self) -> str: + """Return the provider identifier used by Hermes configuration.""" + return "reme" + + def is_available(self) -> bool: + """Check local configuration only; network probes belong to initialize().""" + try: + config = _load_config() + ReMeHttpClient(str(config["endpoint"]), timeout=_positive_float(config, "request_timeout")) + return True + except (KeyError, TypeError, ValueError, OSError): + return False + + def initialize(self, session_id: str, **kwargs: Any) -> None: + """Load profile configuration and probe ReMe without blocking startup.""" + with self._write_thread_lock: + if self._write_thread is not None and self._write_thread.is_alive(): + raise RuntimeError("Cannot reinitialize ReMe while its previous writer is still running") + hermes_home = str(kwargs.get("hermes_home") or "") or None + config = _load_config(hermes_home) + self._endpoint = str(config["endpoint"]) + self._recall_timeout = _positive_float(config, "recall_timeout") + self._health_timeout = _positive_float(config, "health_timeout") + self._health_retry_seconds = _positive_float(config, "health_retry_seconds") + self._shutdown_timeout = _positive_float(config, "shutdown_timeout") + self._recall_limit = _positive_int(config, "recall_limit") + self._session_id = str(session_id or "") + self._profile_id = str(kwargs.get("agent_identity") or "default") + self._write_enabled = str(kwargs.get("agent_context") or "primary") not in {"cron", "flush", "subagent"} + self._client = ReMeHttpClient(self._endpoint, timeout=_positive_float(config, "request_timeout")) + self._service_available = False + self._next_health_probe = 0.0 + self._next_recall_attempt = 0.0 + self._next_write_attempt = 0.0 + self._accept_writes = True + self._write_queue = queue.Queue() + self._write_thread = None + self._shutdown_started = False + if not self._atexit_registered: + atexit.register(self._atexit_shutdown) + self._atexit_registered = True + + if not self._ensure_service(force=True): + logger.warning( + "ReMe is unavailable at %s; recall is disabled and completed " + "turns will not be recorded until it recovers", + self._endpoint, + ) + + def get_config_schema(self) -> List[Dict[str, Any]]: + """Describe interactive setup fields understood by Hermes.""" + return [ + { + "key": "endpoint", + "description": "ReMe HTTP service endpoint", + "default": str(_DEFAULT_CONFIG["endpoint"]), + "required": True, + }, + ] + + def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: + """Atomically save non-secret settings inside the active Hermes profile.""" + path = _config_path(hermes_home) + existing = _load_config(hermes_home) + existing.update({key: value for key, value in dict(values or {}).items() if value is not None and value != ""}) + + # Validate before replacing a working configuration. + client = ReMeHttpClient( + str(existing["endpoint"]), + timeout=_positive_float(existing, "request_timeout"), + ) + client.health(timeout=_positive_float(existing, "health_timeout")) + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(existing, handle, ensure_ascii=False, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.chmod(tmp_name, 0o600) + os.replace(tmp_name, path) + finally: + if os.path.exists(tmp_name): + os.unlink(tmp_name) + + def get_tool_schemas(self) -> List[Dict[str, Any]]: + """Automatic recall and capture add no model-visible tool schemas.""" + return [] + + def prefetch(self, query: str, *, session_id: str = "") -> str: + """Recall relevant memory before Hermes sends a turn to the model.""" + del session_id + query = str(query or "").strip() + if not query or time.monotonic() < self._next_recall_attempt or not self._ensure_service(): + return "" + assert self._client is not None + try: + response = self._client.call( + "search", + {"query": query, "limit": self._recall_limit}, + timeout=self._recall_timeout, + ) + except ReMeServiceError as exc: + self._next_recall_attempt = time.monotonic() + self._health_retry_seconds + logger.warning("ReMe retrieval failed at %s: %s", self._endpoint, exc) + return "" + answer = response.get("answer") + return answer.strip() if isinstance(answer, str) else "" + + def sync_turn( + self, + user_content: str, + assistant_content: str, + *, + session_id: str = "", + messages: Optional[List[Dict[str, Any]]] = None, + ) -> None: + """Queue one completed turn without blocking Hermes on ReMe's LLM.""" + del messages + user = str(user_content or "").strip() + assistant = str(assistant_content or "").strip() + if not self._write_enabled or not (user or assistant): + return + routed_session = str(session_id or self._session_id) + if not routed_session: + logger.warning("ReMe skipped a completed turn because Hermes supplied no session id") + return + if not self._accept_writes: + logger.warning( + "ReMe did not record completed turn for session %s because the provider is shutting down", + _scoped_session_id(self._profile_id, routed_session), + ) + return + + payload = { + "session_id": _scoped_session_id(self._profile_id, routed_session), + "messages": [ + {"name": "user", "role": "user", "content": user}, + {"name": "assistant", "role": "assistant", "content": assistant}, + ], + } + if not self._enqueue_write(payload): + logger.warning( + "ReMe did not record completed turn for session %s because the provider is shutting down", + payload["session_id"], + ) + + def on_session_switch( + self, + new_session_id: str, + *, + parent_session_id: str = "", + reset: bool = False, + rewound: bool = False, + **kwargs: Any, + ) -> None: + """Update the active conversation boundary after a Hermes switch.""" + del parent_session_id, reset, rewound, kwargs + if new_session_id: + self._session_id = str(new_session_id) + + def shutdown(self) -> None: + """Drain queued writes for a bounded interval, then release state.""" + with self._write_thread_lock: + if self._shutdown_started: + return + self._shutdown_started = True + self._accept_writes = False + thread = self._write_thread + if thread is not None: + self._write_queue.put(None) + if thread is not None: + thread.join(timeout=self._shutdown_timeout) + if thread.is_alive(): + abandoned = self._discard_queued_writes() + logger.warning( + "ReMe shutdown timed out after %.1fs; abandoned %d queued write(s) " + "and the in-flight write may not finish before process exit", + self._shutdown_timeout, + abandoned, + ) + else: + self._client = None + else: + self._client = None + self._service_available = False + self._next_health_probe = 0.0 + + def _atexit_shutdown(self) -> None: + try: + self.shutdown() + except Exception as exc: # pragma: no cover - interpreter teardown safety + logger.debug("ReMe atexit shutdown failed: %s", exc) + + def _discard_queued_writes(self) -> int: + abandoned = 0 + while True: + try: + payload = self._write_queue.get_nowait() + except queue.Empty: + break + try: + if payload is not None: + abandoned += 1 + finally: + self._write_queue.task_done() + self._write_queue.put(None) + return abandoned + + def _enqueue_write(self, payload: dict[str, Any]) -> bool: + with self._write_thread_lock: + if not self._accept_writes: + return False + if self._write_thread is None or not self._write_thread.is_alive(): + self._write_thread = threading.Thread( + target=self._write_loop, + args=(self._write_queue,), + daemon=True, + name="reme-memory-writer", + ) + self._write_thread.start() + self._write_queue.put(payload) + return True + + def _write_loop(self, write_queue: queue.Queue[dict[str, Any] | None]) -> None: + try: + while True: + payload = write_queue.get() + try: + if payload is None: + return + try: + self._record_payload(payload) + except Exception as exc: # keep one bad response from killing the writer + logger.exception( + "Unexpected ReMe recording failure for session %s; the writer will continue: %s", + payload.get("session_id", ""), + exc, + ) + finally: + write_queue.task_done() + finally: + current_thread = threading.current_thread() + with self._write_thread_lock: + if self._write_thread is current_thread: + self._write_thread = None + if not self._accept_writes: + self._client = None + + def _record_payload(self, payload: dict[str, Any]) -> None: + if time.monotonic() < self._next_write_attempt: + logger.warning( + "ReMe did not record completed turn for session %s because writes are cooling down", + payload["session_id"], + ) + return + if not self._ensure_service(): + logger.warning( + "ReMe did not record completed turn for session %s because the service is unavailable", + payload["session_id"], + ) + return + assert self._client is not None + try: + self._client.call("auto_memory", payload) + except ReMeServiceError as exc: + self._next_write_attempt = time.monotonic() + self._health_retry_seconds + logger.warning("ReMe recording failed at %s: %s", self._endpoint, exc) + logger.warning( + "ReMe did not record completed turn for session %s", + payload["session_id"], + ) + + def _ensure_service(self, *, force: bool = False) -> bool: + if self._client is None: + return False + if self._service_available and not force: + return True + now = time.monotonic() + if not force and now < self._next_health_probe: + return False + try: + self._client.health(timeout=self._health_timeout) + except ReMeServiceError as exc: + self._mark_unavailable("health check", exc) + return False + self._service_available = True + self._next_health_probe = 0.0 + return True + + def _mark_unavailable(self, operation: str, error: Exception) -> None: + self._service_available = False + self._next_health_probe = time.monotonic() + self._health_retry_seconds + logger.warning("ReMe %s failed at %s: %s", operation, self._endpoint, error) + + +def register(ctx: Any) -> None: + """Register with Hermes' memory-provider collector.""" + ctx.register_memory_provider(ReMeMemoryProvider()) diff --git a/plugins/hermes_agent/client.py b/plugins/hermes_agent/client.py new file mode 100644 index 00000000..3521a217 --- /dev/null +++ b/plugins/hermes_agent/client.py @@ -0,0 +1,76 @@ +"""Small synchronous client for ReMe's HTTP action service.""" + +from __future__ import annotations + +import json +import socket +import urllib.error +import urllib.request + +from typing import Any +from urllib.parse import urlsplit + + +class ReMeServiceError(RuntimeError): + """Raised when a ReMe action cannot be completed successfully.""" + + +class ReMeHttpClient: + """Call ReMe JSON actions without adding a runtime dependency to Hermes.""" + + def __init__(self, endpoint: str, *, timeout: float) -> None: + endpoint = str(endpoint or "").strip().rstrip("/") + parsed = urlsplit(endpoint) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("ReMe endpoint must be an absolute http(s) URL") + self.endpoint = endpoint + self.timeout = max(0.1, float(timeout)) + + def call( + self, + action: str, + payload: dict[str, Any] | None = None, + *, + timeout: float | None = None, + ) -> dict[str, Any]: + """POST one ReMe action and return its standard response envelope.""" + if not action or not action.replace("_", "").isalnum(): + raise ValueError(f"Invalid ReMe action: {action!r}") + + body = json.dumps(payload or {}, ensure_ascii=False).encode("utf-8") + request = urllib.request.Request( + f"{self.endpoint}/{action}", + data=body, + headers={"Accept": "application/json", "Content-Type": "application/json"}, + method="POST", + ) + request_timeout = self.timeout if timeout is None else max(0.1, float(timeout)) + + try: + with urllib.request.urlopen(request, timeout=request_timeout) as response: + raw = response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace")[:500] + raise ReMeServiceError(f"HTTP {exc.code}: {detail or exc.reason}") from exc + except (urllib.error.URLError, TimeoutError, socket.timeout, OSError) as exc: + reason = getattr(exc, "reason", exc) + raise ReMeServiceError(str(reason)) from exc + + try: + result = json.loads(raw) + except json.JSONDecodeError as exc: + raise ReMeServiceError("ReMe returned invalid JSON") from exc + if not isinstance(result, dict): + raise ReMeServiceError("ReMe returned a non-object response") + if result.get("success") is not True: + raise ReMeServiceError(str(result.get("answer") or "ReMe action did not report success")) + return result + + def health(self, *, timeout: float) -> dict[str, Any]: + """Require both a successful response and a healthy component snapshot.""" + result = self.call("health_check", timeout=timeout) + metadata = result.get("metadata") + health = metadata.get("health") if isinstance(metadata, dict) else None + if not isinstance(health, dict) or health.get("healthy") is not True: + raise ReMeServiceError("ReMe did not report a healthy component snapshot") + return result diff --git a/plugins/hermes_agent/plugin.yaml b/plugins/hermes_agent/plugin.yaml new file mode 100644 index 00000000..74e68ec2 --- /dev/null +++ b/plugins/hermes_agent/plugin.yaml @@ -0,0 +1,4 @@ +name: reme +version: 0.1.0 +description: "ReMe file-native long-term memory for Hermes Agent." +pip_dependencies: [] diff --git a/tests/unit/test_hermes_agent_plugin.py b/tests/unit/test_hermes_agent_plugin.py new file mode 100644 index 00000000..d9507796 --- /dev/null +++ b/tests/unit/test_hermes_agent_plugin.py @@ -0,0 +1,457 @@ +"""Hermes provider contract tests using a real loopback HTTP server.""" + +# pylint: disable=redefined-outer-name + +from __future__ import annotations + +import importlib.util +import http.client +import json +import sys +import threading +import time +import types + +from collections.abc import Iterator +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +import pytest + + +class _MemoryProvider: + """Minimal Hermes ABC stand-in; the real loader is exercised separately.""" + + +@pytest.fixture +def plugin_module(monkeypatch: pytest.MonkeyPatch): + """Load the plugin the same way Hermes loads an isolated plugin package.""" + agent = types.ModuleType("agent") + agent.__path__ = [] # type: ignore[attr-defined] + memory_provider = types.ModuleType("agent.memory_provider") + memory_provider.MemoryProvider = _MemoryProvider + monkeypatch.setitem(sys.modules, "agent", agent) + monkeypatch.setitem(sys.modules, "agent.memory_provider", memory_provider) + + module_name = "_reme_hermes_test_plugin" + for name in list(sys.modules): + if name == module_name or name.startswith(f"{module_name}."): + monkeypatch.delitem(sys.modules, name, raising=False) + plugin_dir = Path(__file__).resolve().parents[2] / "plugins" / "hermes_agent" + spec = importlib.util.spec_from_file_location( + module_name, + plugin_dir / "__init__.py", + submodule_search_locations=[str(plugin_dir)], + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, module_name, module) + spec.loader.exec_module(module) + return module + + +class _ActionHandler(BaseHTTPRequestHandler): + calls: list[tuple[str, dict[str, Any]]] + responses: dict[str, tuple[int, Any] | tuple[int, Any, float]] + + def do_POST(self) -> None: # noqa: N802 - stdlib callback name + """Serve one ReMe-compatible action request.""" + length = int(self.headers.get("Content-Length", "0")) + body = json.loads(self.rfile.read(length) or b"{}") + self.calls.append((self.path, body)) + spec = self.responses.get(self.path, (404, {"detail": "not found"})) + status, response = spec[:2] + if len(spec) == 3: + time.sleep(spec[2]) + encoded = json.dumps(response).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + try: + self.wfile.write(encoded) + except (BrokenPipeError, ConnectionResetError): + pass + + def log_message(self, _format: str, *_args: object) -> None: + return + + +@contextmanager +def action_server( + responses: dict[str, tuple[int, Any] | tuple[int, Any, float]], +) -> Iterator[tuple[str, list[tuple[str, dict[str, Any]]]]]: + """Run a loopback ReMe action server and expose captured requests.""" + calls: list[tuple[str, dict[str, Any]]] = [] + handler = type("Handler", (_ActionHandler,), {"calls": calls, "responses": responses}) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + yield f"http://{host}:{port}", calls + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def _healthy_response(answer: str = "healthy") -> dict[str, Any]: + return { + "success": True, + "answer": answer, + "metadata": {"health": {"healthy": True, "version": "test"}}, + } + + +def _wait_for_call(calls: list[tuple[str, dict[str, Any]]], path: str, timeout: float = 2.0) -> None: + """Wait until the background writer reaches a loopback action.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if any(call_path == path for call_path, _ in calls): + return + time.sleep(0.01) + raise AssertionError(f"timed out waiting for {path}; calls={calls!r}") + + +def _write_config(plugin_module, home: Path, endpoint: str, **overrides: Any) -> None: + """Seed runtime config without exercising the separately tested setup path.""" + del plugin_module + values = {"endpoint": endpoint, "health_retry_seconds": 0.1, **overrides} + home.mkdir(parents=True, exist_ok=True) + (home / "reme.json").write_text(json.dumps(values), encoding="utf-8") + + +def test_setup_is_profile_scoped_and_atomic(plugin_module, tmp_path: Path) -> None: + """Keep endpoint configuration private to each Hermes profile.""" + first = tmp_path / "profile-a" + second = tmp_path / "profile-b" + provider = plugin_module.ReMeMemoryProvider() + + responses = {"/health_check": (200, _healthy_response())} + with action_server(responses) as (first_endpoint, _), action_server(responses) as (second_endpoint, _): + provider.save_config({"endpoint": first_endpoint}, str(first)) + provider.save_config({"endpoint": second_endpoint}, str(second)) + + first_config = json.loads((first / "reme.json").read_text(encoding="utf-8")) + second_config = json.loads((second / "reme.json").read_text(encoding="utf-8")) + assert first_config["endpoint"] == first_endpoint + assert second_config["endpoint"] == second_endpoint + assert (first / "reme.json").stat().st_mode & 0o777 == 0o600 + assert not list(first.glob(".reme.json.*")) + + +def test_setup_rejects_unhealthy_endpoint_without_overwrite(plugin_module, tmp_path: Path) -> None: + """Preserve the last working profile config when endpoint validation fails.""" + healthy = {"/health_check": (200, _healthy_response())} + unhealthy = {"/health_check": (503, {"detail": "starting"})} + provider = plugin_module.ReMeMemoryProvider() + + with action_server(healthy) as (healthy_endpoint, _), action_server(unhealthy) as (unhealthy_endpoint, _): + provider.save_config({"endpoint": healthy_endpoint}, str(tmp_path)) + with pytest.raises(plugin_module.ReMeServiceError, match="HTTP 503"): + provider.save_config({"endpoint": unhealthy_endpoint}, str(tmp_path)) + + saved = json.loads((tmp_path / "reme.json").read_text(encoding="utf-8")) + assert saved["endpoint"] == healthy_endpoint + + +def test_setup_rejects_incomplete_health_envelope(plugin_module, tmp_path: Path) -> None: + """Do not accept an unrelated endpoint that returns generic JSON.""" + responses = {"/health_check": (200, {"success": True, "answer": "ok"})} + with action_server(responses) as (endpoint, _): + with pytest.raises(plugin_module.ReMeServiceError, match="healthy component snapshot"): + plugin_module.ReMeMemoryProvider().save_config({"endpoint": endpoint}, str(tmp_path)) + + +def test_client_requires_explicit_success_envelope(plugin_module) -> None: + """Reject action responses that omit ReMe's explicit success signal.""" + responses = {"/search": (200, {"answer": "not a ReMe envelope"})} + with action_server(responses) as (endpoint, _): + client = plugin_module.ReMeHttpClient(endpoint, timeout=1.0) + with pytest.raises(plugin_module.ReMeServiceError, match="not a ReMe envelope"): + client.call("search", {"query": "test"}) + + +def test_lifecycle_retrieves_records_and_switches_sessions(plugin_module, tmp_path: Path) -> None: + """Exercise recall, recording, session switching, and shutdown.""" + responses = { + "/health_check": (200, _healthy_response()), + "/search": (200, {"success": True, "answer": "remembered project decision", "metadata": {}}), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("conversation-one", hermes_home=str(tmp_path), agent_identity="coder") + + assert provider.prefetch("What did we decide?") == "remembered project decision" + provider.sync_turn("Use SQLite", "Recorded that decision") + provider.on_session_switch("conversation-two") + provider.sync_turn("Use BM25 too", "Recorded the retrieval choice") + provider.shutdown() + + assert [path for path, _ in calls] == [ + "/health_check", + "/search", + "/auto_memory", + "/auto_memory", + ] + first_record = calls[2][1] + second_record = calls[3][1] + assert first_record["session_id"].startswith("hermes-coder-conversation-one-") + assert second_record["session_id"].startswith("hermes-coder-conversation-two-") + assert first_record["session_id"] != second_record["session_id"] + assert first_record["messages"] == [ + {"name": "user", "role": "user", "content": "Use SQLite"}, + {"name": "assistant", "role": "assistant", "content": "Recorded that decision"}, + ] + + +def test_gateway_session_argument_preserves_conversation_boundary(plugin_module, tmp_path: Path) -> None: + """Use per-request gateway sessions instead of cached provider state.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("cached-agent", hermes_home=str(tmp_path), agent_identity="gateway") + provider.sync_turn("first", "one", session_id="chat-a") + provider.sync_turn("second", "two", session_id="chat-b") + provider.shutdown() + + assert calls[1][1]["session_id"].startswith("hermes-gateway-chat-a-") + assert calls[2][1]["session_id"].startswith("hermes-gateway-chat-b-") + + +def test_non_primary_context_does_not_write(plugin_module, tmp_path: Path) -> None: + """Avoid recording internal cron, flush, and subagent turns.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize( + "cron-session", + hermes_home=str(tmp_path), + agent_identity="default", + agent_context="cron", + ) + provider.sync_turn("scheduled system prompt", "scheduled result") + + assert [path for path, _ in calls] == ["/health_check"] + + +def test_unavailable_service_fails_open_and_reports_dropped_write(plugin_module, tmp_path: Path, caplog) -> None: + """Keep Hermes usable while making lost persistence explicit.""" + responses = {"/health_check": (503, {"detail": "starting"})} + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + with caplog.at_level("WARNING"): + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + assert provider.prefetch("question") == "" + provider.sync_turn("important fact", "answer") + provider.shutdown() + + assert [path for path, _ in calls] == ["/health_check"] + assert "recall is disabled" in caplog.text + assert "did not record completed turn" in caplog.text + + +def test_unhealthy_snapshot_is_not_treated_as_available(plugin_module, tmp_path: Path) -> None: + """Reject a successful HTTP response whose health snapshot is unhealthy.""" + responses = { + "/health_check": ( + 200, + { + "success": True, + "answer": "unhealthy", + "metadata": {"health": {"healthy": False}}, + }, + ), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + assert provider.prefetch("question") == "" + + assert [path for path, _ in calls] == ["/health_check"] + + +def test_recall_timeout_does_not_block_hermes_turn(plugin_module, tmp_path: Path) -> None: + """Bound inline recall independently from slow automatic-memory writes.""" + responses = { + "/health_check": (200, _healthy_response()), + "/search": (200, {"success": True, "answer": "too late", "metadata": {}}, 0.5), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint, recall_timeout=0.1) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + + started = time.monotonic() + assert provider.prefetch("question") == "" + elapsed = time.monotonic() - started + assert provider.prefetch("cooldown") == "" + provider.sync_turn("recall timed out", "write still works") + provider.shutdown() + + assert elapsed < 0.4 + assert [path for path, _ in calls] == ["/health_check", "/search", "/auto_memory"] + + +def test_recording_failure_does_not_disable_recall(plugin_module, tmp_path: Path) -> None: + """Keep retrieval healthy when only ReMe's LLM-backed write path fails.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": False, "answer": "LLM unavailable", "metadata": {}}), + "/search": (200, {"success": True, "answer": "recall still works", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + provider.sync_turn("remember this", "attempted") + _wait_for_call(calls, "/auto_memory") + + assert provider.prefetch("existing fact") == "recall still works" + provider.shutdown() + + assert [path for path, _ in calls] == ["/health_check", "/auto_memory", "/search"] + + +def test_writer_continues_after_unexpected_payload_exception( + plugin_module, + tmp_path: Path, + caplog, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Do not let a malformed HTTP response kill all later writes.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + original_record = getattr(provider, "_record_payload") + attempts = 0 + + def record_with_one_broken_response(payload: dict[str, Any]) -> None: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise http.client.IncompleteRead(b"partial", 10) + original_record(payload) + + monkeypatch.setattr(provider, "_record_payload", record_with_one_broken_response) + with caplog.at_level("ERROR"): + provider.sync_turn("first", "broken response") + provider.sync_turn("second", "must still be recorded") + provider.shutdown() + + assert attempts == 2 + assert "Unexpected ReMe recording failure" in caplog.text + assert [path for path, _ in calls] == ["/health_check", "/auto_memory"] + + +def test_enqueue_restarts_a_dead_writer(plugin_module, tmp_path: Path) -> None: + """Replace a stale worker reference before accepting another payload.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + dead_writer = threading.Thread(target=lambda: None) + dead_writer.start() + dead_writer.join(timeout=1) + assert not dead_writer.is_alive() + setattr(provider, "_write_thread", dead_writer) + + provider.sync_turn("after failure", "record this") + provider.shutdown() + + assert getattr(provider, "_write_thread") is None + assert [path for path, _ in calls] == ["/health_check", "/auto_memory"] + + +def test_shutdown_is_bounded_when_write_is_slow(plugin_module, tmp_path: Path, caplog) -> None: + """Do not let an in-flight automatic-memory request wedge Hermes exit.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}, 0.8), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint, shutdown_timeout=0.1) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + provider.sync_turn("slow", "write") + _wait_for_call(calls, "/auto_memory") + provider.sync_turn("queued", "must be abandoned") + + with caplog.at_level("WARNING"): + started = time.monotonic() + provider.shutdown() + elapsed = time.monotonic() - started + + assert elapsed < 0.4 + assert "abandoned 1 queued write(s)" in caplog.text + assert [path for path, _ in calls] == ["/health_check", "/auto_memory"] + + +def test_process_exit_fallback_drains_once(plugin_module, tmp_path: Path) -> None: + """Drain a queued turn when Hermes exits without normal provider cleanup.""" + responses = { + "/health_check": (200, _healthy_response()), + "/auto_memory": (200, {"success": True, "answer": "recorded", "metadata": {}}), + } + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + provider.sync_turn("process", "exit") + + shutdown_at_exit = getattr(provider, "_atexit_shutdown") + shutdown_at_exit() + shutdown_at_exit() # idempotent if normal cleanup also ran + + assert [path for path, _ in calls] == ["/health_check", "/auto_memory"] + + +def test_service_recovers_after_health_retry_cooldown(plugin_module, tmp_path: Path) -> None: + """Resume recall after a previously unavailable ReMe service recovers.""" + responses = {"/health_check": (503, {"detail": "starting"})} + with action_server(responses) as (endpoint, calls): + _write_config(plugin_module, tmp_path, endpoint) + provider = plugin_module.ReMeMemoryProvider() + provider.initialize("session", hermes_home=str(tmp_path), agent_identity="default") + + responses["/health_check"] = (200, _healthy_response()) + responses["/search"] = (200, {"success": True, "answer": "recovered", "metadata": {}}) + time.sleep(0.11) + assert provider.prefetch("question") == "recovered" + + assert [path for path, _ in calls] == ["/health_check", "/health_check", "/search"] + + +def test_register_exposes_provider(plugin_module) -> None: + """Register exactly one provider without model-visible tools.""" + registered: list[Any] = [] + context = types.SimpleNamespace(register_memory_provider=registered.append) + plugin_module.register(context) + assert len(registered) == 1 + assert registered[0].name == "reme" + assert registered[0].get_tool_schemas() == []