feat(memory): add Supermemory setup connection summary

Add post_setup() and get_status_config() to the Supermemory memory
provider so `hermes memory setup` and `hermes memory status` print a
one-line connection summary (container, profile fact count,
auto_recall/auto_capture). Point API-key onboarding at the Hermes
connect URL (app.supermemory.ai/integrations?connect=hermes).

Salvage of #52988. Two fixes folded in:

- Test isolation: the new probe/status tests mocked _SupermemoryClient
  but not the __import__("supermemory") guard inside
  _probe_supermemory_connection, so they passed only where the optional
  supermemory package was installed and failed on a clean checkout / CI
  (the PR shipped with red CI). Added _stub_supermemory_importable()
  mirroring the existing test_is_available_false_when_import_missing
  pattern; the suite now passes with supermemory absent.

- post_setup: `if api_key and api_key not in os.environ` checked whether
  the key's *value* named an env var (always false in practice). Fixed to
  compare the value: `os.environ.get("SUPERMEMORY_API_KEY") != api_key`.

Verified: 38/38 in test_supermemory_provider.py and the full
tests/plugins/memory/ suite green with supermemory not installed.

Closes #52988
This commit is contained in:
Mahesh Sanikommu 2026-06-27 15:01:27 +05:30 committed by kshitij
parent 8827300267
commit 1b75b3fd90
5 changed files with 268 additions and 4 deletions

View file

@ -5,7 +5,7 @@ Semantic long-term memory with profile recall, semantic search, explicit memory
## Requirements
- `pip install supermemory`
- Supermemory API key from [supermemory.ai](https://supermemory.ai)
- Supermemory API key from [app.supermemory.ai/integrations?connect=hermes](http://app.supermemory.ai/integrations?connect=hermes)
## Setup

View file

@ -32,6 +32,7 @@ _DEFAULT_API_TIMEOUT = 5.0
_MIN_CAPTURE_LENGTH = 10
_MAX_ENTITY_CONTEXT_LENGTH = 1500
_CONVERSATIONS_URL = "https://api.supermemory.ai/v4/conversations"
_API_KEY_URL = "http://app.supermemory.ai/integrations?connect=hermes"
_TRIVIAL_RE = re.compile(
r"^(ok|okay|thanks|thank you|got it|sure|yes|no|yep|nope|k|ty|thx|np)\.?$",
re.IGNORECASE,
@ -387,6 +388,65 @@ class _SupermemoryClient:
return
def _resolve_container_tag_for_setup(hermes_home: str, *, identity: str = "default") -> str:
config = _load_supermemory_config(hermes_home)
env_tag = os.environ.get("SUPERMEMORY_CONTAINER_TAG", "").strip()
raw_tag = env_tag or config["container_tag"]
return _sanitize_tag(raw_tag.replace("{identity}", identity))
def _probe_supermemory_connection(api_key: str, hermes_home: str, *, identity: str = "default") -> dict:
config = _load_supermemory_config(hermes_home)
status = {
"ok": False,
"error": "",
"container_tag": _resolve_container_tag_for_setup(hermes_home, identity=identity),
"profile_facts": 0,
"auto_recall": bool(config["auto_recall"]),
"auto_capture": bool(config["auto_capture"]),
}
if not (api_key or "").strip():
status["error"] = "SUPERMEMORY_API_KEY not set"
return status
try:
__import__("supermemory")
except ImportError:
status["error"] = "supermemory package not installed"
return status
try:
client = _SupermemoryClient(
api_key=api_key.strip(),
timeout=config["api_timeout"],
container_tag=status["container_tag"],
search_mode=config["search_mode"],
)
profile = client.get_profile()
facts = [
fact for fact in (profile.get("static") or []) + (profile.get("dynamic") or [])
if fact and str(fact).strip()
]
status["profile_facts"] = len(facts)
status["ok"] = True
except Exception as exc:
status["error"] = str(exc).strip()[:160] or "connection failed"
return status
def _format_connection_summary(status: dict) -> str:
recall = "on" if status.get("auto_recall") else "off"
capture = "on" if status.get("auto_capture") else "off"
container = status.get("container_tag") or _DEFAULT_CONTAINER_TAG
if status.get("ok"):
facts = int(status.get("profile_facts") or 0)
fact_label = "fact" if facts == 1 else "facts"
return (
f"✓ Connected · container: {container} · {facts} profile {fact_label} · "
f"auto_recall {recall} · auto_capture {capture}"
)
err = status.get("error") or "connection failed"
return f"{err} · container: {container} · auto_recall {recall} · auto_capture {capture}"
STORE_SCHEMA = {
"name": "supermemory_store",
"description": "Store an explicit memory for future recall.",
@ -487,7 +547,7 @@ class SupermemoryMemoryProvider(MemoryProvider):
# All other options are documented for $HERMES_HOME/supermemory.json
# or the SUPERMEMORY_CONTAINER_TAG env var.
return [
{"key": "api_key", "description": "Supermemory API key", "secret": True, "required": True, "env_var": "SUPERMEMORY_API_KEY", "url": "https://supermemory.ai"},
{"key": "api_key", "description": "Supermemory API key", "secret": True, "required": True, "env_var": "SUPERMEMORY_API_KEY", "url": _API_KEY_URL},
]
def save_config(self, values, hermes_home):
@ -498,6 +558,57 @@ class SupermemoryMemoryProvider(MemoryProvider):
sanitized["entity_context"] = _clamp_entity_context(str(sanitized["entity_context"]))
_save_supermemory_config(sanitized, hermes_home)
def get_status_config(self, provider_config: dict) -> dict:
from hermes_constants import get_hermes_home
del provider_config
hermes_home = str(get_hermes_home())
api_key = os.environ.get("SUPERMEMORY_API_KEY", "")
status = _probe_supermemory_connection(api_key, hermes_home)
return {"summary": _format_connection_summary(status)}
def post_setup(self, hermes_home: str, config: dict) -> None:
from pathlib import Path
from hermes_cli.config import save_config
from hermes_cli.memory_setup import _prompt, _write_env_vars
print("\n Configuring supermemory:\n")
print(f" Get your API key at {_API_KEY_URL}\n")
env_writes: dict[str, str] = {}
existing = os.environ.get("SUPERMEMORY_API_KEY", "")
if existing:
masked = f"...{existing[-4:]}" if len(existing) > 4 else "set"
val = _prompt(f"Supermemory API key (current: {masked}, blank to keep)", secret=True)
else:
val = _prompt("Supermemory API key", secret=True)
if val:
env_writes["SUPERMEMORY_API_KEY"] = val
if not isinstance(config.get("memory"), dict):
config["memory"] = {}
config["memory"]["provider"] = self.name
save_config(config)
if env_writes:
_write_env_vars(Path(hermes_home) / ".env", env_writes)
api_key = env_writes.get("SUPERMEMORY_API_KEY") or existing
# Make the freshly-entered key visible to the connection probe below.
# (Checks the VALUE of SUPERMEMORY_API_KEY, not whether the key string
# happens to name some unrelated env var.)
if api_key and os.environ.get("SUPERMEMORY_API_KEY") != api_key:
os.environ["SUPERMEMORY_API_KEY"] = api_key
status = _probe_supermemory_connection(api_key, hermes_home)
print(f"\n {_format_connection_summary(status)}")
print("\n Memory provider: supermemory")
print(" Activation saved to config.yaml")
if env_writes:
print(" API keys saved to .env")
print("\n Start a new session to activate.\n")
def initialize(self, session_id: str, **kwargs) -> None:
from hermes_constants import get_hermes_home
self._hermes_home = kwargs.get("hermes_home") or str(get_hermes_home())

View file

@ -8,8 +8,10 @@ import pytest
from plugins.memory.supermemory import (
SupermemoryMemoryProvider,
_clean_text_for_capture,
_format_connection_summary,
_format_prefetch_context,
_load_supermemory_config,
_probe_supermemory_connection,
_save_supermemory_config,
)
@ -449,6 +451,157 @@ def test_get_config_schema_minimal():
assert schema[0]["secret"] is True
def test_format_connection_summary_ok():
summary = _format_connection_summary({
"ok": True,
"container_tag": "hermes_coder",
"profile_facts": 12,
"auto_recall": True,
"auto_capture": False,
})
assert "✓ Connected" in summary
assert "container: hermes_coder" in summary
assert "12 profile facts" in summary
assert "auto_recall on" in summary
assert "auto_capture off" in summary
def test_format_connection_summary_single_fact_and_error():
one = _format_connection_summary({
"ok": True,
"container_tag": "hermes",
"profile_facts": 1,
"auto_recall": True,
"auto_capture": True,
})
assert "1 profile fact" in one
assert "1 profile facts" not in one
err = _format_connection_summary({
"ok": False,
"error": "invalid API key",
"container_tag": "hermes",
"auto_recall": True,
"auto_capture": True,
})
assert "✗ invalid API key" in err
assert "container: hermes" in err
def test_probe_supermemory_connection_missing_key(tmp_path):
status = _probe_supermemory_connection("", str(tmp_path))
assert status["ok"] is False
assert status["error"] == "SUPERMEMORY_API_KEY not set"
assert status["container_tag"] == "hermes"
def _stub_supermemory_importable(monkeypatch):
"""Make ``__import__("supermemory")`` succeed without the real package.
``_probe_supermemory_connection`` guards on ``__import__("supermemory")``
before using the (mocked) client, so tests that mock ``_SupermemoryClient``
must also satisfy that import guard otherwise they only pass in an
environment where the optional ``supermemory`` package happens to be
installed (and fail on a clean checkout / CI). Mirrors the inverse stub in
``test_is_available_false_when_import_missing``.
"""
import builtins
import types
real_import = builtins.__import__
def fake_import(name, *args, **kwargs):
if name == "supermemory":
return types.ModuleType("supermemory")
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", fake_import)
def test_probe_supermemory_connection_success(monkeypatch, tmp_path):
_stub_supermemory_importable(monkeypatch)
monkeypatch.setattr("plugins.memory.supermemory._SupermemoryClient", FakeClient)
class CountingClient(FakeClient):
def get_profile(self, query=None, *, container_tag=None):
return {
"static": ["Prefers TypeScript"],
"dynamic": ["", "Working on Hermes"],
"search_results": [],
}
monkeypatch.setattr("plugins.memory.supermemory._SupermemoryClient", CountingClient)
status = _probe_supermemory_connection("test-key", str(tmp_path))
assert status["ok"] is True
assert status["profile_facts"] == 2
assert status["auto_recall"] is True
def test_probe_supermemory_connection_client_error(monkeypatch, tmp_path):
_stub_supermemory_importable(monkeypatch)
class BrokenClient(FakeClient):
def get_profile(self, query=None, *, container_tag=None):
raise RuntimeError("API unavailable")
monkeypatch.setattr("plugins.memory.supermemory._SupermemoryClient", BrokenClient)
status = _probe_supermemory_connection("test-key", str(tmp_path))
assert status["ok"] is False
assert "API unavailable" in status["error"]
def test_get_status_config_returns_summary(monkeypatch, tmp_path):
_stub_supermemory_importable(monkeypatch)
monkeypatch.setenv("SUPERMEMORY_API_KEY", "test-key")
monkeypatch.setattr("plugins.memory.supermemory._SupermemoryClient", FakeClient)
monkeypatch.setattr(
"hermes_constants.get_hermes_home",
lambda: tmp_path,
)
result = SupermemoryMemoryProvider().get_status_config({})
assert "summary" in result
assert "✓ Connected" in result["summary"]
assert "container: hermes" in result["summary"]
def test_post_setup_writes_config_and_prints_summary(monkeypatch, tmp_path, capsys):
config: dict = {"memory": {}}
monkeypatch.setenv("SUPERMEMORY_API_KEY", "")
monkeypatch.setattr(
"hermes_cli.memory_setup._prompt",
lambda label, secret=True, default=None: "new-api-key",
)
monkeypatch.setattr(
"plugins.memory.supermemory._probe_supermemory_connection",
lambda api_key, hermes_home, **kwargs: {
"ok": True,
"container_tag": "hermes",
"profile_facts": 3,
"auto_recall": True,
"auto_capture": True,
},
)
saved: dict = {}
def fake_save_config(cfg):
saved.update(cfg)
monkeypatch.setattr("hermes_cli.config.save_config", fake_save_config)
SupermemoryMemoryProvider().post_setup(str(tmp_path), config)
assert config["memory"]["provider"] == "supermemory"
assert saved["memory"]["provider"] == "supermemory"
env_text = (tmp_path / ".env").read_text(encoding="utf-8")
assert "SUPERMEMORY_API_KEY=new-api-key" in env_text
out = capsys.readouterr().out
assert "✓ Connected" in out
assert "3 profile facts" in out
assert "Memory provider: supermemory" in out
@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits not enforced on Windows")
def test_save_config_sets_owner_only_permissions(tmp_path):
"""supermemory.json must be written with 0o600 so API key is not world-readable."""

View file

@ -513,7 +513,7 @@ Semantic long-term memory with profile recall, semantic search, explicit memory
| | |
|---|---|
| **Best for** | Semantic recall with user profiling and session-level graph building |
| **Requires** | `pip install supermemory` + [API key](https://supermemory.ai) |
| **Requires** | `pip install supermemory` + [API key](http://app.supermemory.ai/integrations?connect=hermes) |
| **Data storage** | Supermemory Cloud |
| **Cost** | Supermemory pricing |

View file

@ -467,7 +467,7 @@ hermes config set memory.provider byterover
| | |
|---|---|
| **适合场景** | 带用户 profile 和会话级图谱构建的语义召回 |
| **依赖** | `pip install supermemory` + [API key](https://supermemory.ai) |
| **依赖** | `pip install supermemory` + [API key](http://app.supermemory.ai/integrations?connect=hermes) |
| **数据存储** | Supermemory Cloud |
| **费用** | Supermemory 定价 |