From 13c5f4e627da7a0a784ff2c1b8fefa08b49e4aab Mon Sep 17 00:00:00 2001 From: Marcin Antas Date: Sat, 18 Jul 2026 20:12:39 +0200 Subject: [PATCH] feat: add Hermes Agent memory provider --- .github/workflows/test.yml | 32 +++ README.md | 11 +- hermes/README.md | 57 ++++ hermes/engram/README.md | 79 ++++++ hermes/engram/__init__.py | 508 ++++++++++++++++++++++++++++++++++ hermes/engram/plugin.yaml | 7 + hermes/install.sh | 139 ++++++++++ hermes/tests/conftest.py | 186 +++++++++++++ hermes/tests/test_provider.py | 292 +++++++++++++++++++ 9 files changed, 1309 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/test.yml create mode 100644 hermes/README.md create mode 100644 hermes/engram/README.md create mode 100644 hermes/engram/__init__.py create mode 100644 hermes/engram/plugin.yaml create mode 100755 hermes/install.sh create mode 100644 hermes/tests/conftest.py create mode 100644 hermes/tests/test_provider.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..5869d13 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,32 @@ +name: test + +on: + push: + branches: [main] + pull_request: + +jobs: + hermes-provider: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12"] + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Syntax checks + run: | + bash -n hermes/install.sh + bash -n plugin/hooks/with-venv.sh + python -m compileall -q hermes/engram plugin/core + + - name: Install pytest + run: pip install pytest + + - name: Run Hermes provider tests + run: python -m pytest hermes/tests -v diff --git a/README.md b/README.md index 4bb17de..1790697 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,16 @@ # Weaviate Engram Integrations -Persistent, cross-session memory for **Claude Code**, backed by -[Weaviate Engram](https://docs.weaviate.io/engram). Claude remembers your preferences, +Persistent, cross-session memory for AI assistants, backed by +[Weaviate Engram](https://docs.weaviate.io/engram). Your assistant remembers your preferences, decisions, and project context across sessions — and recalls what's relevant before it answers. +## Integrations + +- **Claude Code** — the `engram` plugin in [`plugin/`](plugin) (install below). +- **Hermes Agent** — the memory provider in [`hermes/`](hermes); see [`hermes/README.md`](hermes/README.md). + +## Claude Code + - **Recall** — before each answer, relevant memories are fetched and added to the conversation. - **Store** — after each turn, the exchange is saved so it can be recalled later. diff --git a/hermes/README.md b/hermes/README.md new file mode 100644 index 0000000..333310e --- /dev/null +++ b/hermes/README.md @@ -0,0 +1,57 @@ +# Engram × Hermes Agent + +A [Hermes Agent](https://hermes-agent.nousresearch.com) memory provider backed by +[Weaviate Engram](https://docs.weaviate.io/engram) — persistent, cross-session memory with +server-side extraction and scoped recall. + +## Install + +```bash +bash hermes/install.sh +``` + +The installer auto-detects where your Hermes looks for providers: + +- **`$HERMES_HOME/plugins/engram/`** (user-installed providers, survives `hermes update`) — + preferred when supported; +- **`/plugins/memory/engram/`** (bundled) — fallback for older versions, + or forced with `--bundled`. + +Flags: `--link` symlinks instead of copying (development — edits here go live); +pass a checkout path or set `HERMES_REPO` if auto-detection fails. + +Then configure: + +```bash +hermes memory setup # choose "engram", paste your API key +``` + +Get a key at https://console.weaviate.cloud/engram. Memory works from the first turn. + +**Config reference, identity rules, tools, privacy:** see [`engram/README.md`](engram/README.md). + +## Layout + +``` +hermes/ +├── install.sh # installer (copy/symlink into the right plugin location) +├── engram/ # the provider package — self-contained, upstream-PR-ready +│ ├── __init__.py # EngramMemoryProvider + register() +│ ├── plugin.yaml # metadata + pip_dependencies (weaviate-engram) +│ └── README.md # setup + config reference +└── tests/ # pytest suite — no network, no real SDK (stubbed) +``` + +The provider package is deliberately self-contained (no imports from this repo) so it can be +dropped into a `NousResearch/hermes-agent` PR at `plugins/memory/engram/` unchanged. + +## Development + +```bash +bash hermes/install.sh --link # live-edit against your Hermes install +python3 -m venv .venv && .venv/bin/pip install pytest +.venv/bin/python -m pytest hermes/tests -v +``` + +The tests stub the Hermes ABC, `hermes_constants`, and the Engram SDK, and load the provider by +file path — mirroring Hermes' own discovery, so they double as a drop-in compatibility check. diff --git a/hermes/engram/README.md b/hermes/engram/README.md new file mode 100644 index 0000000..284966c --- /dev/null +++ b/hermes/engram/README.md @@ -0,0 +1,79 @@ +# Engram Memory Provider for Hermes Agent + +Persistent, cross-session memory backed by [Weaviate Engram](https://docs.weaviate.io/engram). +Conversation turns are sent to Engram for server-side extraction; relevant memories are recalled +before each turn and searchable on demand. + +| | | +| --- | --- | +| **Best for** | Hands-off memory — Engram handles extraction and organization automatically | +| **Requires** | `pip install weaviate-engram` (auto-installed) + API key | +| **Data storage** | Engram Cloud | +| **Cost** | Engram pricing (cloud) | + +## Setup + +Get an API key at https://console.weaviate.cloud/engram, then: + +```bash +hermes memory setup # select "engram", paste the key +``` + +Or manually: + +```bash +hermes config set memory.provider engram +echo "ENGRAM_API_KEY=your-key" >> ~/.hermes/.env +``` + +**Tools (2):** `engram_search` (semantic search over the user's memories), +`engram_add` (store a durable fact the moment the user states one). + +## How it works + +- **Recall** — before each turn, a background `memories.search` runs against the current prompt; + results are injected as context. The agent can also search on demand with `engram_search`. +- **Store** — after each turn, the exchange is sent to `memories.add` in a daemon thread (never + blocks a response), tagged with `session_id` as the scope property. +- **Mirroring** — writes to Hermes' built-in `MEMORY.md` / `USER.md` are mirrored to Engram + (`add` and `replace` actions). `remove` is **not** propagated — Engram is append-only from this + provider's side, so a built-in deletion can't be undone remotely. +- **Fail-open** — a missing key, missing identity, or a down API disables memory for the session; + it never breaks a conversation. After 5 consecutive API failures, calls pause for 120s + (circuit breaker). + +## Identity + +Memories are isolated per `user_id`, resolved in this order: + +1. `user_id` in `$HERMES_HOME/engram.json` (or `ENGRAM_USER_ID` env) — operator-configured, + applies uniformly across every gateway (CLI, Telegram, Discord, …). +2. The gateway-native user id (Telegram numeric id, Discord snowflake, …). +3. `git config user.email`. + +There is deliberately **no shared default**: a non-unique id would commingle different people's +memories with no way to un-mix them later. If no identity resolves, the provider stays disabled +and logs why. + +## Config + +Secret — in `$HERMES_HOME/.env` or the environment: + +| Variable | Purpose | +| --- | --- | +| `ENGRAM_API_KEY` | Your Engram API key (required). | + +Non-secret — `$HERMES_HOME/engram.json` (written by `hermes memory setup`; env vars read as +fallback): + +| Key | Env var | Default | Description | +| --- | --- | --- | --- | +| `user_id` | `ENGRAM_USER_ID` | — | Canonical user identifier (see Identity above). | +| `base_url` | `ENGRAM_BASE_URL` | `https://api.engram.weaviate.io` | Endpoint override (dev/self-hosted). | + +## Privacy + +Off-device data: conversation turns (user + assistant text), facts stored via `engram_add`, and +mirrored built-in memory writes are sent to Engram Cloud for extraction and storage. Tool calls +and tool results are **not** forwarded. Recall queries (the user's current prompt) are sent on +search. diff --git a/hermes/engram/__init__.py b/hermes/engram/__init__.py new file mode 100644 index 0000000..3876384 --- /dev/null +++ b/hermes/engram/__init__.py @@ -0,0 +1,508 @@ +"""Engram memory plugin — MemoryProvider interface. + +Persistent, cross-session memory backed by Weaviate Engram +(https://docs.weaviate.io/engram). Conversation turns are sent to Engram for +server-side extraction; relevant memories are recalled before each turn. + +Configuration +------------- +Secret (lives in $HERMES_HOME/.env or the environment): + ENGRAM_API_KEY — Engram API key (required). Get one at + https://console.weaviate.cloud/engram + +Behavioral settings (live in $HERMES_HOME/engram.json, set via `hermes memory +setup`; env vars are read as fallback): + user_id — Canonical user identifier (env: ENGRAM_USER_ID). When + set, it applies uniformly across every gateway so the + same human gets one memory store. When unset, the + gateway-native id is used, falling back to + `git config user.email`. NEVER falls back to a shared + literal — a non-unique id would commingle different + people's memories with no way to un-mix them later. + base_url — Engram endpoint override (env: ENGRAM_BASE_URL; + default https://api.engram.weaviate.io). + +Writes attach `session_id` as the scope property (the default Engram project's +only scope property); recall is scoped to the user. Fail-open throughout: a +missing key/identity or a down API disables memory — it never breaks a session. +""" + +from __future__ import annotations + +import json +import logging +import os +import subprocess +import threading +import time +from typing import Any, Dict, List + +from agent.memory_provider import MemoryProvider + +logger = logging.getLogger(__name__) + +DEFAULT_BASE_URL = "https://api.engram.weaviate.io" + +# Circuit breaker: after this many consecutive failures, pause API calls for +# _BREAKER_COOLDOWN_SECS to avoid hammering a down server. +_BREAKER_THRESHOLD = 5 +_BREAKER_COOLDOWN_SECS = 120 +# How long prefetch() waits on an in-flight recall before skipping injection. +_PREFETCH_WAIT_SECS = 3 + + +def _is_client_error(exc: Exception) -> bool: + """True for user-caused errors (bad request, not found) that should NOT trip + the circuit breaker.""" + err_str = str(exc).lower() + return "404" in err_str or "not found" in err_str or "400" in err_str + + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +def _load_config() -> dict: + """Load config from env vars, with $HERMES_HOME/engram.json overrides. + + Environment variables provide defaults; engram.json (if present) overrides + individual keys. Empty values in the file are ignored so a partial file + never blanks out an env var. + """ + from hermes_constants import get_hermes_home + + config = { + "api_key": os.environ.get("ENGRAM_API_KEY", ""), + "user_id": os.environ.get("ENGRAM_USER_ID", ""), + "base_url": os.environ.get("ENGRAM_BASE_URL", DEFAULT_BASE_URL), + } + config_path = get_hermes_home() / "engram.json" + if config_path.exists(): + try: + file_cfg = json.loads(config_path.read_text(encoding="utf-8")) + config.update({k: v for k, v in file_cfg.items() + if v is not None and v != ""}) + except Exception: + pass + return config + + +def _git_email() -> str: + """`git config user.email`, or "" — git missing/timed out/not set all degrade + to no identity.""" + try: + out = subprocess.run( + ["git", "config", "user.email"], + capture_output=True, + text=True, + timeout=2, + ) + except Exception: + return "" + return out.stdout.strip() + + +def _memory_contents(results) -> List[str]: + """Extract content strings from a search result page, tolerating both SDK + objects and plain dicts.""" + lines = [] + for m in results or []: + content = getattr(m, "content", None) + if content is None and isinstance(m, dict): + content = m.get("content") + if content: + lines.append(str(content).strip()) + return lines + + +# --------------------------------------------------------------------------- +# Tool schemas +# --------------------------------------------------------------------------- + +SEARCH_SCHEMA = { + "name": "engram_search", + "description": ( + "Search the user's long-term memories by meaning. Use this before " + "answering any question that may depend on what you know about the " + "user (preferences, facts, history, people, projects, past decisions). " + "For multi-part questions, call it several times with different " + "wording — one search is rarely enough." + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What to search for."}, + }, + "required": ["query"], + }, +} + +ADD_SCHEMA = { + "name": "engram_add", + "description": ( + "Store a durable fact about the user. Call this the moment the user " + "states a lasting preference, correction, decision, or personal detail " + "worth recalling on future turns — don't wait to be asked to remember. " + "Skip transient chit-chat and facts already stored." + ), + "parameters": { + "type": "object", + "properties": { + "content": {"type": "string", "description": "The fact to store."}, + }, + "required": ["content"], + }, +} + + +# --------------------------------------------------------------------------- +# MemoryProvider implementation +# --------------------------------------------------------------------------- + +class EngramMemoryProvider(MemoryProvider): + """Weaviate Engram memory with server-side extraction and semantic recall.""" + + def __init__(self): + self._config = None + self._client = None + self._enabled = False + self._user_id = "" + self._session_id = "" + self._agent_context = "primary" + self._sync_thread = None + self._prefetch_thread = None + self._prefetch_query = "" + self._prefetch_result = "" + self._prefetch_done = False + self._prefetch_lock = threading.Lock() + self._sync_lock = threading.Lock() + self._breaker_lock = threading.Lock() + self._consecutive_failures = 0 + self._breaker_open_until = 0.0 + + @property + def name(self) -> str: + return "engram" + + def is_available(self) -> bool: + """No network calls — just check an API key is configured.""" + try: + return bool(_load_config().get("api_key")) + except Exception: + return False + + # -- Config ------------------------------------------------------------- + + def get_config_schema(self): + return [ + { + "key": "api_key", + "description": "Engram API key", + "secret": True, + "required": True, + "env_var": "ENGRAM_API_KEY", + "url": "https://console.weaviate.cloud/engram", + }, + { + "key": "user_id", + "description": "User identifier (blank → gateway id, then git user.email)", + "required": False, + }, + ] + + def save_config(self, values, hermes_home): + """Write non-secret config to $HERMES_HOME/engram.json (merge, so an + existing base_url survives a re-run of the wizard).""" + import json as _json + from pathlib import Path + + config_path = Path(hermes_home) / "engram.json" + existing = {} + if config_path.exists(): + try: + existing = _json.loads(config_path.read_text(encoding="utf-8")) + except Exception: + pass + existing.update(values) + config_path.write_text(_json.dumps(existing, indent=2) + "\n", encoding="utf-8") + + # -- Lifecycle ---------------------------------------------------------- + + def initialize(self, session_id: str, **kwargs) -> None: + self._config = _load_config() + self._session_id = session_id or "" + self._agent_context = kwargs.get("agent_context") or "primary" + if not (self._config.get("api_key") or "").strip(): + logger.warning("Engram: ENGRAM_API_KEY not set — memory disabled.") + self._enabled = False + return + # Identity: operator-configured user_id → gateway-native id → git email. + # No shared literal fallback — memories are tagged with this id + # permanently, so a non-unique id would commingle different people's + # memories with no way to un-mix them later. + self._user_id = ( + (self._config.get("user_id") or "").strip() + or (kwargs.get("user_id") or "").strip() + or _git_email() + ) + if not self._user_id: + logger.warning( + "Engram: no stable identity — set user_id in engram.json / " + "ENGRAM_USER_ID, or git config user.email. Memory disabled " + "(prevents mixing memories between users)." + ) + self._enabled = False + return + try: + from engram import EngramClient + + self._client = EngramClient( + api_key=self._config["api_key"], + base_url=self._config.get("base_url") or DEFAULT_BASE_URL, + ) + except Exception as e: + logger.error("Engram client failed to initialize: %s", e) + self._client = None + self._enabled = False + return + self._enabled = True + + def system_prompt_block(self) -> str: + if not self._enabled: + return "" + return ( + "# Engram Memory\n" + f"Active. User: {self._user_id}.\n" + "You have persistent memory of this user from past conversations. " + "Call engram_search before answering anything that could depend on " + "prior context (the user's preferences, facts, history, people, " + "projects, or earlier decisions) — do not rely on the chat window " + "alone, and do not assume you have no memory.\n" + "Call engram_add the moment the user states a lasting preference, " + "correction, decision, or personal detail worth recalling later.\n" + "Tools: engram_search to find memories, engram_add to store facts." + ) + + def shutdown(self) -> None: + for t in (self._prefetch_thread, self._sync_thread): + if t and t.is_alive(): + t.join(timeout=5.0) + try: + if self._client and hasattr(self._client, "close"): + self._client.close() + except Exception: + pass + self._client = None + self._enabled = False + + # -- Circuit breaker ------------------------------------------------------ + + def _is_breaker_open(self) -> bool: + with self._breaker_lock: + if self._consecutive_failures < _BREAKER_THRESHOLD: + return False + if time.monotonic() >= self._breaker_open_until: + self._consecutive_failures = 0 + return False + return True + + def _record_success(self): + with self._breaker_lock: + self._consecutive_failures = 0 + + def _record_failure(self): + with self._breaker_lock: + self._consecutive_failures += 1 + count = self._consecutive_failures + if count >= _BREAKER_THRESHOLD: + self._breaker_open_until = time.monotonic() + _BREAKER_COOLDOWN_SECS + else: + count = 0 + if count >= _BREAKER_THRESHOLD: + logger.warning( + "Engram circuit breaker tripped after %d consecutive failures. " + "Pausing API calls for %ds.", + count, _BREAKER_COOLDOWN_SECS, + ) + + # -- Recall (background prefetch + hot-path consume) ---------------------- + + def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None: + self._start_prefetch(message) + + def queue_prefetch(self, query: str, *, session_id: str = "") -> None: + self._start_prefetch(query) + + def _consume_prefetch_result(self, query: str): + with self._prefetch_lock: + if self._prefetch_query != query or not self._prefetch_done: + return None + result = self._prefetch_result + self._prefetch_result = "" + self._prefetch_done = False + return result + + def _start_prefetch(self, query: str) -> None: + if not query or not self._enabled or self._client is None: + return + if self._is_breaker_open(): + return + client = self._client + user_id = self._user_id + with self._prefetch_lock: + if self._prefetch_query == query: + if self._prefetch_done: + return + if self._prefetch_thread and self._prefetch_thread.is_alive(): + return + self._prefetch_query = query + self._prefetch_result = "" + self._prefetch_done = False + + def _run(): + body = "" + try: + results = client.memories.search(query=query, user_id=user_id) + lines = _memory_contents(results) + if lines: + body = "## Engram Memory\n" + "\n".join(f"- {l}" for l in lines) + self._record_success() + except Exception as e: + self._record_failure() + logger.debug("Engram prefetch failed: %s", e) + with self._prefetch_lock: + if self._prefetch_query == query: + self._prefetch_result = body + self._prefetch_done = True + + t = threading.Thread(target=_run, daemon=True, name="engram-prefetch") + with self._prefetch_lock: + self._prefetch_thread = t + t.start() + + def prefetch(self, query: str, *, session_id: str = "") -> str: + """Recall memories for the CURRENT question with a short hot-path wait.""" + if not self._enabled: + return "" + cached = self._consume_prefetch_result(query) + if cached 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) + cached = self._consume_prefetch_result(query) + if cached is not None: + return cached + # Slow backend: skip injection; engram_search remains as the backstop. + return "" + + # -- Store (always background, never blocks a turn) ------------------------ + + def _start_add(self, messages: List[Dict[str, str]], session_id: str = "") -> None: + """Fire a background memories.add. Serializes with any in-flight add: + joins it briefly and skips on contention rather than duplicating.""" + if not self._enabled or self._client is None or self._is_breaker_open(): + return + client = self._client + user_id = self._user_id + sid = session_id or self._session_id + properties = {"session_id": sid} if sid else None + + def _add(): + try: + client.memories.add(messages, user_id=user_id, properties=properties) + self._record_success() + except Exception as e: + self._record_failure() + logger.warning("Engram sync failed: %s", e) + + with self._sync_lock: + if self._sync_thread and self._sync_thread.is_alive(): + self._sync_thread.join(timeout=5.0) + # Still alive after the wait → skip to avoid duplicate ingestion. + if self._sync_thread and self._sync_thread.is_alive(): + return + self._sync_thread = threading.Thread(target=_add, daemon=True, name="engram-sync") + self._sync_thread.start() + + def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None: + """Persist a completed turn (non-blocking). Skipped for non-primary + contexts — subagent/cron turns would corrupt the user's memory.""" + if self._agent_context not in ("primary", ""): + return + messages = [] + if (user_content or "").strip(): + messages.append({"role": "user", "content": user_content}) + if (assistant_content or "").strip(): + messages.append({"role": "assistant", "content": assistant_content}) + if not messages: + return + self._start_add(messages, session_id=session_id) + + def on_memory_write(self, action: str, target: str, content: str) -> None: + """Mirror built-in MEMORY.md / USER.md writes into Engram. The backend + is append-only: a 'replace' is stored as the new/corrected fact, and + 'remove' can't be propagated — logged instead.""" + if action not in ("add", "replace"): + logger.debug( + "Engram: not mirroring built-in memory %r on %s — append-only backend", + action, target, + ) + return + text = (content or "").strip() + if not text: + return + self._start_add([{"role": "user", "content": text}]) + + # -- Tools ----------------------------------------------------------------- + + def get_tool_schemas(self) -> List[Dict[str, Any]]: + return [SEARCH_SCHEMA, ADD_SCHEMA] + + def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str: + if not self._enabled or self._client is None: + return json.dumps({"error": "Engram memory is not active (missing API key or identity)."}) + if self._is_breaker_open(): + return json.dumps({"error": "Engram temporarily unavailable (multiple consecutive failures). Will retry automatically."}) + + if tool_name == "engram_search": + query = (args.get("query") or "").strip() + if not query: + return json.dumps({"error": "Missing required parameter: query"}) + try: + results = self._client.memories.search(query=query, user_id=self._user_id) + self._record_success() + lines = _memory_contents(results) + if not lines: + return json.dumps({"result": "No relevant memories found."}) + return json.dumps({"results": lines, "count": len(lines)}) + except Exception as e: + if not _is_client_error(e): + self._record_failure() + return json.dumps({"error": f"Search failed: {e}"}) + + if tool_name == "engram_add": + content = (args.get("content") or "").strip() + if not content: + return json.dumps({"error": "Missing required parameter: content"}) + try: + self._client.memories.add( + [{"role": "user", "content": content}], + user_id=self._user_id, + properties={"session_id": self._session_id} if self._session_id else None, + ) + self._record_success() + return json.dumps({"result": "Fact stored."}) + except Exception as e: + if not _is_client_error(e): + self._record_failure() + return json.dumps({"error": f"Failed to store: {e}"}) + + return json.dumps({"error": f"Unknown tool: {tool_name}"}) + + +def register(ctx) -> None: + """Register Engram as a memory provider plugin.""" + ctx.register_memory_provider(EngramMemoryProvider()) diff --git a/hermes/engram/plugin.yaml b/hermes/engram/plugin.yaml new file mode 100644 index 0000000..c7e7a10 --- /dev/null +++ b/hermes/engram/plugin.yaml @@ -0,0 +1,7 @@ +name: engram +version: 1.0.0 +description: "Weaviate Engram — persistent cross-session memory with server-side extraction and scoped semantic recall." +pip_dependencies: + - weaviate-engram>=1.0.0,<2.0.0 +hooks: + - on_memory_write diff --git a/hermes/install.sh b/hermes/install.sh new file mode 100755 index 0000000..f2ff646 --- /dev/null +++ b/hermes/install.sh @@ -0,0 +1,139 @@ +#!/usr/bin/env bash +# Install the Engram memory provider for Hermes Agent. +# +# Hermes discovers memory providers in two places: +# 1. $HERMES_HOME/plugins// (user-installed — survives `hermes update`) +# 2. /plugins/memory// (bundled — the only option on older versions) +# This script picks the best target automatically: the user-plugins dir when the +# installed Hermes supports it, otherwise the checkout's bundled dir. +# +# Usage: +# bash install.sh [--link] [--bundled] [HERMES_REPO] +# +# --link symlink instead of copying (development — edits here go live) +# --bundled force install into the checkout's plugins/memory/ (e.g. for an +# upstream PR worktree) even if user-plugins are supported +# HERMES_REPO path to the Hermes checkout; auto-detected when omitted: +# $HERMES_REPO → $HERMES_HOME/hermes-agent (~/.hermes default) +# → /usr/local/lib/hermes-agent (root installs) + +set -euo pipefail + +usage() { + sed -n '14,23p' "${BASH_SOURCE[0]}" | sed 's/^#\{1,\} \{0,1\}//' +} + +LINK=false +BUNDLED=false +REPO_ARG="" +for arg in "$@"; do + case "$arg" in + --link) LINK=true ;; + --bundled) BUNDLED=true ;; + -h|--help) usage; exit 0 ;; + *) REPO_ARG="$arg" ;; + esac +done + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +SRC="$SCRIPT_DIR/engram" +HH="${HERMES_HOME:-$HOME/.hermes}" + +if [ ! -f "$SRC/__init__.py" ]; then + echo "error: provider package not found at $SRC" >&2 + exit 1 +fi + +find_repo() { + local candidates=() + if [ -n "$REPO_ARG" ]; then + candidates+=("$REPO_ARG") + fi + if [ -n "${HERMES_REPO:-}" ]; then + candidates+=("$HERMES_REPO") + fi + candidates+=("$HH/hermes-agent") + candidates+=("/usr/local/lib/hermes-agent") + local c + for c in "${candidates[@]}"; do + if [ -d "$c/plugins/memory" ]; then + echo "$c" + return 0 + fi + done + return 1 +} + +REPO="$(find_repo || true)" + +supports_user_plugins() { + # User-installed providers ($HERMES_HOME/plugins/) were added to Hermes' + # discovery after the first bundled-only versions — detect by the marker + # function in the checkout's plugins/memory/__init__.py. + [ -n "$1" ] && grep -q "_get_user_plugins_dir" "$1/plugins/memory/__init__.py" 2>/dev/null +} + +DEST="" +if [ "$BUNDLED" = false ]; then + if supports_user_plugins "$REPO"; then + DEST="$HH/plugins/engram" + elif [ -z "$REPO" ] && [ -d "$HH/plugins" ]; then + # No checkout in the usual places (e.g. desktop install), but the + # user-plugins dir exists — Hermes clearly supports it. + DEST="$HH/plugins/engram" + fi +fi + +if [ -z "$DEST" ] && [ -n "$REPO" ]; then + DEST="$REPO/plugins/memory/engram" +fi + +if [ -z "$DEST" ]; then + cat >&2 < $SRC" +else + # tar-to-tar copy excludes Python caches + tar -cf - --exclude='__pycache__' --exclude='*.pyc' -C "$SCRIPT_DIR" engram \ + | tar -xf - -C "$(dirname "$DEST")" + echo "Copied $SRC -> $DEST" +fi + +cat <> ~/.hermes/.env + 3. Start a chat — memory works from the first turn. +EOF diff --git a/hermes/tests/conftest.py b/hermes/tests/conftest.py new file mode 100644 index 0000000..5cb8bd0 --- /dev/null +++ b/hermes/tests/conftest.py @@ -0,0 +1,186 @@ +"""Test harness: stub the Hermes internals and the Engram SDK in sys.modules, +then load the provider package by file path — mirroring Hermes' own +plugins/memory discovery (_load_provider_from_dir), so these tests double as a +drop-in compatibility check for an upstream PR. + +No network, no real SDK, no Hermes checkout needed. +""" + +import importlib.util +import os +import sys +import types +from abc import ABC, abstractmethod +from pathlib import Path + +import pytest + +ENGRAM_PKG = Path(__file__).resolve().parents[1] / "engram" + + +# --- Stub: agent.memory_provider (the ABC the provider imports at module top) --- + +agent_mod = types.ModuleType("agent") +mp_mod = types.ModuleType("agent.memory_provider") + + +class MemoryProvider(ABC): + """Minimal mirror of the real ABC's abstract surface (plus the optional + hooks our provider overrides).""" + + @property + @abstractmethod + def name(self) -> str: ... + + @abstractmethod + def is_available(self) -> bool: ... + + @abstractmethod + def initialize(self, session_id: str, **kwargs) -> None: ... + + @abstractmethod + def get_tool_schemas(self): ... + + def system_prompt_block(self) -> str: + return "" + + def prefetch(self, query: str, *, session_id: str = "") -> str: + return "" + + def queue_prefetch(self, query: str, *, session_id: str = "") -> None: + pass + + def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None: + pass + + def handle_tool_call(self, tool_name: str, args, **kwargs) -> str: + raise NotImplementedError + + def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None: + pass + + def on_memory_write(self, action: str, target: str, content: str) -> None: + pass + + def shutdown(self) -> None: + pass + + def get_config_schema(self): + return [] + + def save_config(self, values, hermes_home) -> None: + pass + + +mp_mod.MemoryProvider = MemoryProvider +agent_mod.memory_provider = mp_mod +sys.modules.setdefault("agent", agent_mod) +sys.modules["agent.memory_provider"] = mp_mod + + +# --- Stub: hermes_constants (imported lazily inside provider functions) --- + +hermes_constants = types.ModuleType("hermes_constants") + + +def get_hermes_home(): + # Read at call time so each test's HERMES_HOME (tmp_path) applies. + return Path(os.environ.get("HERMES_HOME", "/nonexistent-hermes-home")) + + +hermes_constants.get_hermes_home = get_hermes_home +sys.modules["hermes_constants"] = hermes_constants + + +# --- Stub: engram SDK (imported lazily in initialize) --- + +class FakeMemories: + def __init__(self): + self.add_calls = [] + self.search_calls = [] + self.search_results = [] + self.error = None # raised by add/search when set + + def add(self, messages, user_id=None, properties=None): + if self.error: + raise self.error + self.add_calls.append( + {"messages": messages, "user_id": user_id, "properties": properties} + ) + return {} + + def search(self, query=None, user_id=None, topics=None, properties=None): + if self.error: + raise self.error + self.search_calls.append( + {"query": query, "user_id": user_id, "topics": topics, "properties": properties} + ) + return self.search_results + + +class FakeEngramClient: + instances = [] + + def __init__(self, api_key=None, base_url=None): + self.api_key = api_key + self.base_url = base_url + self.memories = FakeMemories() + self.closed = False + FakeEngramClient.instances.append(self) + + def close(self): + self.closed = True + + +engram_mod = types.ModuleType("engram") +engram_mod.EngramClient = FakeEngramClient +sys.modules["engram"] = engram_mod + + +# --- Provider loading (register-pattern, like Hermes discovery) --- + +def load_provider_module(): + init_file = ENGRAM_PKG / "__init__.py" + spec = importlib.util.spec_from_file_location("plugins.memory.engram", str(init_file)) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class _Ctx: + def __init__(self): + self.provider = None + + def register_memory_provider(self, provider): + self.provider = provider + + +@pytest.fixture +def mod(): + return load_provider_module() + + +@pytest.fixture(autouse=True) +def clean_env(tmp_path, monkeypatch): + for var in ("ENGRAM_API_KEY", "ENGRAM_USER_ID", "ENGRAM_BASE_URL"): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + FakeEngramClient.instances.clear() + yield + + +@pytest.fixture +def provider(mod): + ctx = _Ctx() + mod.register(ctx) + p = ctx.provider + assert p is not None, "register(ctx) must register a provider instance" + yield p + p.shutdown() + + +def drain(p, timeout=5.0): + """Wait for the provider's background threads (sync/prefetch) to finish.""" + for t in (p._sync_thread, p._prefetch_thread): + if t and t.is_alive(): + t.join(timeout=timeout) diff --git a/hermes/tests/test_provider.py b/hermes/tests/test_provider.py new file mode 100644 index 0000000..ce921a5 --- /dev/null +++ b/hermes/tests/test_provider.py @@ -0,0 +1,292 @@ +"""Unit tests for the Engram Hermes memory provider. + +Everything runs against the fakes in conftest.py — no network, no real SDK. +""" + +import json +from types import SimpleNamespace + +from conftest import FakeEngramClient, drain + + +# --- Registration / availability ------------------------------------------- + +def test_register_and_name(provider, mod): + assert provider.name == "engram" + assert isinstance(provider, mod.MemoryProvider) + + +def test_is_available_requires_api_key(provider, monkeypatch): + assert provider.is_available() is False + monkeypatch.setenv("ENGRAM_API_KEY", "k") + assert provider.is_available() is True + + +def test_config_file_overrides_env(provider, monkeypatch, tmp_path): + monkeypatch.setenv("ENGRAM_API_KEY", "env-key") + monkeypatch.setenv("ENGRAM_USER_ID", "env-user") + (tmp_path / "engram.json").write_text( + json.dumps({"api_key": "file-key", "user_id": "", "base_url": "http://x"}) + ) + from conftest import load_provider_module + + cfg = load_provider_module()._load_config() + assert cfg["api_key"] == "file-key" # file wins over env + assert cfg["user_id"] == "env-user" # empty file value ignored → env kept + assert cfg["base_url"] == "http://x" + + +def test_get_config_schema(provider): + schema = {f["key"]: f for f in provider.get_config_schema()} + assert schema["api_key"]["secret"] is True + assert schema["api_key"]["required"] is True + assert schema["api_key"]["env_var"] == "ENGRAM_API_KEY" + assert "user_id" in schema + + +def test_save_config_merges(provider, tmp_path): + (tmp_path / "engram.json").write_text(json.dumps({"base_url": "http://keep"})) + provider.save_config({"user_id": "u1"}, str(tmp_path)) + cfg = json.loads((tmp_path / "engram.json").read_text()) + assert cfg == {"base_url": "http://keep", "user_id": "u1"} + + +# --- initialize: guards and identity ---------------------------------------- + +def _no_git(monkeypatch, mod): + """Simulate git being unusable / unset.""" + monkeypatch.setattr(mod.subprocess, "run", + lambda *a, **k: SimpleNamespace(stdout="")) + + +def test_initialize_disabled_without_api_key(provider): + provider.initialize("s1", user_id="u") + assert provider._enabled is False + assert provider.system_prompt_block() == "" + assert provider.prefetch("q") == "" + + +def test_initialize_disabled_without_identity(provider, mod, monkeypatch): + monkeypatch.setenv("ENGRAM_API_KEY", "k") + _no_git(monkeypatch, mod) + provider.initialize("s1") # no config user_id, no gateway id, no git email + assert provider._enabled is False + assert provider._client is None + # hooks are no-ops, never raise + provider.sync_turn("hi", "hello") + assert provider.prefetch("q") == "" + assert "not active" in provider.handle_tool_call("engram_search", {"query": "q"}) + + +def test_identity_resolution_order(provider, mod, monkeypatch, tmp_path): + monkeypatch.setenv("ENGRAM_API_KEY", "k") + monkeypatch.setattr(mod.subprocess, "run", + lambda *a, **k: SimpleNamespace(stdout="git@example.com\n")) + + # 1. configured user_id beats the gateway id + (tmp_path / "engram.json").write_text(json.dumps({"user_id": "configured"})) + provider.initialize("s1", user_id="gateway") + assert provider._user_id == "configured" + + # 2. gateway id beats git email + (tmp_path / "engram.json").unlink() + provider.initialize("s1", user_id="gateway") + assert provider._user_id == "gateway" + + # 3. git email is the last resort + provider.initialize("s1") + assert provider._user_id == "git@example.com" + + +def test_initialize_creates_client(provider, monkeypatch): + monkeypatch.setenv("ENGRAM_API_KEY", "k") + provider.initialize("sess-1", user_id="u1") + assert provider._enabled is True + client = FakeEngramClient.instances[-1] + assert client.api_key == "k" + assert client.base_url == "https://api.engram.weaviate.io" + + +# --- sync_turn --------------------------------------------------------------- + +def _initialized(provider, monkeypatch, **kwargs): + monkeypatch.setenv("ENGRAM_API_KEY", "k") + kwargs.setdefault("user_id", "u1") + provider.initialize("sess-1", **kwargs) + return FakeEngramClient.instances[-1] + + +def test_sync_turn_payload(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + provider.sync_turn("hi there", "hello!") + drain(provider) + assert len(client.memories.add_calls) == 1 + call = client.memories.add_calls[0] + assert call["messages"] == [ + {"role": "user", "content": "hi there"}, + {"role": "assistant", "content": "hello!"}, + ] + assert call["user_id"] == "u1" + assert call["properties"] == {"session_id": "sess-1"} + + +def test_sync_turn_session_id_override(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + provider.sync_turn("a", "b", session_id="other-session") + drain(provider) + assert client.memories.add_calls[0]["properties"] == {"session_id": "other-session"} + + +def test_sync_turn_skips_non_primary_context(provider, monkeypatch): + client = _initialized(provider, monkeypatch, agent_context="subagent") + provider.sync_turn("hi", "hello") + drain(provider) + assert client.memories.add_calls == [] + + +def test_sync_turn_empty_turn_ignored(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + provider.sync_turn("", " ") + drain(provider) + assert client.memories.add_calls == [] + + +# --- prefetch / recall -------------------------------------------------------- + +def test_prefetch_formats_memories(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + client.memories.search_results = [ + {"content": "likes dark mode"}, + SimpleNamespace(content="works at Weaviate"), + {"content": ""}, # blank entries dropped + ] + body = provider.prefetch("what does the user like?") + assert body == "## Engram Memory\n- likes dark mode\n- works at Weaviate" + # search ran user-scoped + assert client.memories.search_calls[0]["user_id"] == "u1" + + +def test_prefetch_empty_when_nothing_found(provider, monkeypatch): + _initialized(provider, monkeypatch) + assert provider.prefetch("anything") == "" + + +def test_prefetch_failure_is_silent(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + client.memories.error = RuntimeError("boom") + assert provider.prefetch("q") == "" + assert provider._consecutive_failures == 1 + + +def test_on_turn_start_warms_cache(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + client.memories.search_results = [{"content": "cached fact"}] + provider.on_turn_start(1, "tell me about myself") + assert provider.prefetch("tell me about myself") == "## Engram Memory\n- cached fact" + # exactly one search — prefetch consumed the warmed result + assert len(client.memories.search_calls) == 1 + + +# --- tools -------------------------------------------------------------------- + +def test_tool_search(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + client.memories.search_results = [{"content": "fact one"}, {"content": "fact two"}] + out = json.loads(provider.handle_tool_call("engram_search", {"query": "q"})) + assert out == {"results": ["fact one", "fact two"], "count": 2} + + +def test_tool_search_no_results(provider, monkeypatch): + _initialized(provider, monkeypatch) + out = json.loads(provider.handle_tool_call("engram_search", {"query": "q"})) + assert out["result"] == "No relevant memories found." + + +def test_tool_search_missing_query(provider, monkeypatch): + _initialized(provider, monkeypatch) + out = json.loads(provider.handle_tool_call("engram_search", {})) + assert "error" in out + + +def test_tool_add(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + out = json.loads(provider.handle_tool_call("engram_add", {"content": "I prefer vim"})) + assert out["result"] == "Fact stored." + assert client.memories.add_calls[0]["messages"] == [ + {"role": "user", "content": "I prefer vim"} + ] + + +def test_tool_add_missing_content(provider, monkeypatch): + _initialized(provider, monkeypatch) + out = json.loads(provider.handle_tool_call("engram_add", {})) + assert "error" in out + + +def test_tool_error_does_not_trip_breaker_on_client_error(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + client.memories.error = RuntimeError("404 not found") + for _ in range(6): + provider.handle_tool_call("engram_search", {"query": "q"}) + assert provider._consecutive_failures == 0 # client errors don't count + + +def test_unknown_tool(provider, monkeypatch): + _initialized(provider, monkeypatch) + out = json.loads(provider.handle_tool_call("engram_nope", {})) + assert "Unknown tool" in out["error"] + + +def test_tool_schemas(provider): + names = [s["name"] for s in provider.get_tool_schemas()] + assert names == ["engram_search", "engram_add"] + for schema in provider.get_tool_schemas(): + assert schema["parameters"]["type"] == "object" + assert schema["parameters"]["required"] + + +# --- on_memory_write mirroring ------------------------------------------------- + +def test_on_memory_write_mirrors_add_and_replace(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + provider.on_memory_write("add", "memory", "user likes tea") + drain(provider) + provider.on_memory_write("replace", "user", "user likes coffee") + drain(provider) + assert len(client.memories.add_calls) == 2 + assert client.memories.add_calls[0]["messages"][0]["content"] == "user likes tea" + assert client.memories.add_calls[1]["messages"][0]["content"] == "user likes coffee" + + +def test_on_memory_write_ignores_remove(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + provider.on_memory_write("remove", "memory", "obsolete fact") + drain(provider) + assert client.memories.add_calls == [] + + +# --- circuit breaker ------------------------------------------------------------ + +def test_circuit_breaker_opens_after_repeated_failures(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + client.memories.error = RuntimeError("connection refused") + for _ in range(5): + provider.sync_turn("a", "b") + drain(provider) + assert provider._is_breaker_open() is True + # while open, tools short-circuit and background work doesn't start + out = json.loads(provider.handle_tool_call("engram_search", {"query": "q"})) + assert "temporarily unavailable" in out["error"] + thread_before = provider._sync_thread + provider.sync_turn("a", "b") + assert provider._sync_thread is thread_before + + +# --- shutdown ------------------------------------------------------------------- + +def test_shutdown_closes_client(provider, monkeypatch): + client = _initialized(provider, monkeypatch) + provider.sync_turn("a", "b") + provider.shutdown() + assert client.closed is True + assert provider._enabled is False