From a6b45ffb1fa195aca9a2170e6efca54bc62f6bcf Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 21:32:00 -0700
Subject: [PATCH] refactor(memory/supermemory): _quietly helper for
log-and-return, tuple-driven tool schemas, inline single-use helpers
---
plugins/memory/supermemory/__init__.py | 304 +++++++++----------------
1 file changed, 111 insertions(+), 193 deletions(-)
diff --git a/plugins/memory/supermemory/__init__.py b/plugins/memory/supermemory/__init__.py
index 6ece3cf514..0b339660b5 100644
--- a/plugins/memory/supermemory/__init__.py
+++ b/plugins/memory/supermemory/__init__.py
@@ -1,8 +1,8 @@
-"""Supermemory memory plugin (MemoryProvider): profile recall, semantic search, explicit memory tools,
-cleaned turn capture, and session-end conversation ingest."""
+"""Supermemory memory plugin (MemoryProvider): profile recall, semantic search, memory tools, turn capture, session ingest."""
from __future__ import annotations
+import importlib
import json
import logging
import os
@@ -11,7 +11,7 @@ import threading
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
from agent.memory_provider import MemoryProvider
from agent.secret_scope import get_secret, is_multiplex_active
@@ -40,6 +40,16 @@ _ALIAS_TO_TOOL = {kebab: snake for snake, kebab in _KEBAB_ALIASES.items()}
_BOOL_WORDS = {**dict.fromkeys(("true", "1", "yes", "y", "on"), True), **dict.fromkeys(("false", "0", "no", "n", "off"), False)}
+def _quietly(fn: Callable[[], Any], fail_msg: str = "", *args: Any, level: int = logging.DEBUG, default: Any = None) -> Any:
+ """Run ``fn()``; on any exception log ``fail_msg`` (if given) with traceback and return ``default``."""
+ try:
+ return fn()
+ except Exception:
+ if fail_msg:
+ logger.log(level, fail_msg, *args, exc_info=True)
+ return default
+
+
def _sanitize_tag(raw: str) -> str:
return re.sub(r"_+", "_", re.sub(r"[^a-zA-Z0-9_]", "_", raw or "")).strip("_") or _DEFAULT_CONTAINER_TAG
@@ -56,21 +66,12 @@ def _clamp_entity_context(text: str) -> str:
def _as_bool(value: Any, default: bool) -> bool:
"""bool passthrough; common true/false words parsed; anything else (incl. ints) -> default."""
- if isinstance(value, bool):
- return value
- return _BOOL_WORDS.get(value.strip().lower(), default) if isinstance(value, str) else default
-
-
-def _valid_search_mode(mode: str) -> str:
- return mode if mode in _VALID_SEARCH_MODES else "hybrid"
+ return value if isinstance(value, bool) else _BOOL_WORDS.get(value.strip().lower(), default) if isinstance(value, str) else default
def _clamp_number(value: Any, default, lo, hi, cast):
"""Cast ``value`` and clamp it to [lo, hi]; fall back to ``default`` on any conversion error."""
- try:
- return max(lo, min(hi, cast(value)))
- except Exception:
- return default
+ return _quietly(lambda: max(lo, min(hi, cast(value))), default=default)
# config key -> (default, normalizer applied to the raw/merged value). Order = supermemory.json layout.
@@ -83,11 +84,10 @@ _CONFIG_SPEC: Dict[str, tuple] = {
"max_recall_results": (10, lambda v: _clamp_number(v, 10, 1, 20, int)),
"profile_frequency": (50, lambda v: _clamp_number(v, 50, 1, 500, int)),
"capture_mode": ("all", lambda v: "everything" if v == "everything" else "all"),
- "search_mode": ("hybrid", lambda v: _valid_search_mode(str(v).strip().lower())),
+ "search_mode": ("hybrid", lambda v: v if (v := str(v).strip().lower()) in _VALID_SEARCH_MODES else "hybrid"),
"entity_context": (_DEFAULT_ENTITY_CONTEXT, lambda v: _clamp_entity_context(str(v))),
"api_timeout": (5.0, lambda v: _clamp_number(v, 5.0, 0.5, 15.0, float)),
"base_url": ("", lambda v: str(v or "").strip()),
- # Multi-container support
"enable_custom_container_tags": (False, lambda v: _as_bool(v, False)),
"custom_containers": ([], lambda v: [_sanitize_tag(str(t)) for t in v if t] if isinstance(v, list) else []),
"custom_container_instructions": ("", lambda v: str(v).strip()),
@@ -95,12 +95,7 @@ _CONFIG_SPEC: Dict[str, tuple] = {
def _read_json_dict(path: Path) -> dict:
- """JSON object stored at ``path`` or {} if missing/invalid."""
- try:
- raw = json.loads(path.read_text(encoding="utf-8")) if path.exists() else None
- except Exception:
- logger.debug("Failed to parse %s", path, exc_info=True)
- return {}
+ raw = _quietly(lambda: json.loads(path.read_text(encoding="utf-8")), "Failed to parse %s", path) if path.exists() else None
return raw if isinstance(raw, dict) else {}
@@ -115,11 +110,9 @@ def _load_supermemory_config(hermes_home: Optional[str] = None) -> dict:
def _save_supermemory_config(values: dict, hermes_home: str) -> None:
- config_path = Path(hermes_home) / "supermemory.json"
- existing = _read_json_dict(config_path)
- existing.update(values)
from utils import atomic_json_write
- atomic_json_write(config_path, existing, mode=0o600, sort_keys=True)
+ config_path = Path(hermes_home) / "supermemory.json"
+ atomic_json_write(config_path, {**_read_json_dict(config_path), **values}, mode=0o600, sort_keys=True)
def _detect_category(text: str) -> str:
@@ -129,24 +122,20 @@ def _detect_category(text: str) -> str:
def _format_relative_time(iso_timestamp: str) -> str:
- try:
- dt = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00"))
- now = datetime.now(timezone.utc)
+ """'just now' / '5m ago' / '3h ago' / '2d ago' / '%d %b[ %Y]'; '' when unparseable."""
+ def _fmt():
+ dt, now = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")), datetime.now(timezone.utc)
seconds = (now - dt).total_seconds()
for limit, unit, label in ((1800, 0, "just now"), (3600, 60, "m ago"), (86400, 3600, "h ago"), (604800, 86400, "d ago")):
if seconds < limit:
return f"{int(seconds / unit)}{label}" if unit else label
return dt.strftime("%d %b" if dt.year == now.year else "%d %b %Y")
- except Exception:
- return ""
+ return _quietly(_fmt, default="")
def _similarity_pct(value: Any) -> Optional[int]:
"""0..1 similarity -> whole percent; None when absent or unparseable."""
- try:
- return None if value is None else round(float(value) * 100)
- except Exception:
- return None
+ return _quietly(lambda: None if value is None else round(float(value) * 100))
def _profile_sections(static_facts: list, dynamic_facts: list) -> list[str]:
@@ -160,21 +149,15 @@ def _format_prefetch_context(static_facts: list, dynamic_facts: list, search_res
def _unique(items, key=lambda x: x): # set.add() returns None, so `not seen.add(k)` records k and keeps the item
return [i for i in items or [] if (k := key(i)) and k not in seen and not seen.add(k)][:max_results]
-
sections = _profile_sections(_unique(static_facts), _unique(dynamic_facts))
lines = []
for item in _unique(search_results, key=lambda i: i.get("memory", "")):
rel = _format_relative_time(item.get("updated_at") or item.get("updatedAt") or "")
pct = _similarity_pct(item.get("similarity"))
- prefix_bits = ([f"[{rel}]"] if rel else []) + ([f"[{pct}%]"] if pct is not None else [])
- lines.append(f"- {' '.join(prefix_bits)} {item['memory']}".strip())
- if lines:
- sections.append("## Relevant Memories\n" + "\n".join(lines))
- if not sections:
- return ""
- intro = ("The following is background context from long-term memory. Use it silently when relevant. "
- "Do not force memories into the conversation.")
- return f"\n{intro}\n\n" + "\n\n".join(sections) + "\n"
+ lines.append(f"- {' '.join(([f'[{rel}]'] if rel else []) + ([f'[{pct}%]'] if pct is not None else []))} {item['memory']}".strip())
+ sections += ["## Relevant Memories\n" + "\n".join(lines)] if lines else []
+ intro = "The following is background context from long-term memory. Use it silently when relevant. Do not force memories into the conversation."
+ return f"\n{intro}\n\n" + "\n\n".join(sections) + "\n" if sections else ""
def _clean_text_for_capture(text: str) -> str:
@@ -183,10 +166,9 @@ def _clean_text_for_capture(text: str) -> str:
def _memory_fields(item: Any, *keys: str) -> dict:
"""Pick SDK result attrs into a plain dict; ``updated_at`` also accepts camelCase ``updatedAt``."""
- values = {"id": getattr(item, "id", ""), "memory": getattr(item, "memory", ""), "similarity": getattr(item, "similarity", None),
- "updated_at": getattr(item, "updated_at", None) or getattr(item, "updatedAt", None),
- "metadata": getattr(item, "metadata", None)}
- return {k: values[k] for k in keys}
+ defaults = {"id": "", "memory": "", "similarity": None, "metadata": None}
+ return {k: getattr(item, "updated_at", None) or getattr(item, "updatedAt", None) if k == "updated_at" else getattr(item, k, defaults[k])
+ for k in keys}
class _SupermemoryClient:
@@ -194,15 +176,11 @@ class _SupermemoryClient:
search_mode: str = "hybrid", base_url: str = ""):
# Lazy-install the SDK on demand (honors security.allow_lazy_installs and sealed Docker
# venvs). On failure fall through so the raw import produces the canonical ImportError.
- try:
- from tools.lazy_deps import ensure as _lazy_ensure
- _lazy_ensure("memory.supermemory", prompt=False)
- except Exception:
- pass
+ _quietly(lambda: importlib.import_module("tools.lazy_deps").ensure("memory.supermemory", prompt=False))
from supermemory import Supermemory
-
self._api_key, self._container_tag, self._timeout = api_key, container_tag, timeout
- self._search_mode, self._base_url = _valid_search_mode(search_mode), _resolve_base_url(base_url)
+ self._search_mode = search_mode if search_mode in _VALID_SEARCH_MODES else "hybrid"
+ self._base_url = _resolve_base_url(base_url)
self._client = Supermemory(api_key=api_key, base_url=self._base_url, timeout=timeout, max_retries=0,
default_headers={"x-sm-source": "hermes"})
@@ -210,8 +188,7 @@ class _SupermemoryClient:
# sm_source routes Hermes writes into the "Hermes" Space in the Supermemory app so the user
# can filter / bulk-manage them per source agent (a routing key for the user, not telemetry).
merged = {"sm_source": "hermes", **(metadata or {})}
- legacy_source = merged.pop("source", None)
- if legacy_source and "type" not in merged:
+ if (legacy_source := merged.pop("source", None)) and "type" not in merged:
merged["type"] = str(legacy_source)
return merged
@@ -238,8 +215,7 @@ class _SupermemoryClient:
search_data = getattr(response, "search_results", None) or getattr(response, "searchResults", None)
raw_results = getattr(search_data, "results", None) or search_data or []
return {
- "static": (getattr(profile_data, "static", []) or []) if profile_data else [],
- "dynamic": (getattr(profile_data, "dynamic", []) or []) if profile_data else [],
+ **{k: (getattr(profile_data, k, []) or []) if profile_data else [] for k in ("static", "dynamic")},
"search_results": [item if isinstance(item, dict) else _memory_fields(item, "memory", "updated_at", "similarity")
for item in raw_results] if isinstance(raw_results, list) else [],
}
@@ -258,10 +234,9 @@ class _SupermemoryClient:
def ingest_conversation(self, session_id: str, messages: list[dict], metadata: dict | None = None) -> None:
payload: dict = {"conversationId": session_id, "messages": messages, "containerTags": [self._container_tag],
**({"metadata": self._merge_metadata(metadata)} if metadata else {})}
- req = urllib.request.Request(
- f"{self._base_url}/v4/conversations", data=json.dumps(payload).encode("utf-8"), method="POST",
- headers={"Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", "x-sm-source": "hermes"},
- )
+ req = urllib.request.Request(f"{self._base_url}/v4/conversations", data=json.dumps(payload).encode("utf-8"), method="POST",
+ headers={"Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json",
+ "x-sm-source": "hermes"})
with urllib.request.urlopen(req, timeout=self._timeout + 3):
return
@@ -282,20 +257,17 @@ def _probe_supermemory_connection(api_key: str, hermes_home: str, *, identity: s
status = {"ok": False, "error": "", "profile_facts": 0, "container_tag": _resolve_container_tag(config["container_tag"], identity),
"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
+ return {**status, "error": "SUPERMEMORY_API_KEY not set"}
try:
__import__("supermemory")
except ImportError:
- status["error"] = "supermemory package not installed"
- return status
+ return {**status, "error": "supermemory package not installed"}
try:
profile = _build_client(api_key.strip(), config, status["container_tag"]).get_profile()
- status["profile_facts"] = sum(1 for f in (profile.get("static") or []) + (profile.get("dynamic") or []) if f and str(f).strip())
- status["ok"] = True
except Exception as exc:
- status["error"] = str(exc).strip()[:160] or "connection failed"
- return status
+ return {**status, "error": str(exc).strip()[:160] or "connection failed"}
+ facts = sum(1 for f in (profile.get("static") or []) + (profile.get("dynamic") or []) if f and str(f).strip())
+ return {**status, "ok": True, "profile_facts": facts}
def _format_connection_summary(status: dict) -> str:
@@ -307,24 +279,19 @@ def _format_connection_summary(status: dict) -> str:
return f"✗ {status.get('error') or 'connection failed'} · container: {container} · {flags}"
-def _str_prop(description: str) -> dict:
- return {"type": "string", "description": description}
-
-
-# (name, description, properties, required) -> tool schema; kebab aliases are added in get_tool_schemas().
+# (name, description, ((prop, type, description), ...), required) -> tool schema; kebab aliases are added in get_tool_schemas().
_BASE_SCHEMAS = [
- {"name": name, "description": description, "parameters": {"type": "object", "properties": props, **({"required": req} if req else {})}}
+ {"name": name, "description": description,
+ "parameters": {"type": "object", "properties": {p: {"type": t, "description": d} for p, t, d in props}, **({"required": req} if req else {})}}
for name, description, props, req in (
("supermemory_store", "Store an explicit memory for future recall.",
- {"content": _str_prop("The memory content to store."),
- "metadata": {"type": "object", "description": "Optional metadata attached to the memory."}}, ["content"]),
+ (("content", "string", "The memory content to store."), ("metadata", "object", "Optional metadata attached to the memory.")), ["content"]),
("supermemory_search", "Search long-term memory by semantic similarity.",
- {"query": _str_prop("What to search for."),
- "limit": {"type": "integer", "description": "Maximum results to return, 1 to 20."}}, ["query"]),
+ (("query", "string", "What to search for."), ("limit", "integer", "Maximum results to return, 1 to 20.")), ["query"]),
("supermemory_forget", "Forget a memory by exact id or by best-match query.",
- {"id": _str_prop("Exact memory id to delete."), "query": _str_prop("Query used to find the memory to forget.")}, None),
+ (("id", "string", "Exact memory id to delete."), ("query", "string", "Query used to find the memory to forget.")), None),
("supermemory_profile", "Retrieve persistent profile facts and recent memory context.",
- {"query": _str_prop("Optional query to focus the profile response.")}, None),
+ (("query", "string", "Optional query to focus the profile response."),), None),
)
]
@@ -341,22 +308,17 @@ class SupermemoryMemoryProvider(MemoryProvider):
def __init__(self):
self._api_key = self._session_id = self._hermes_home = ""
self._client: Optional[_SupermemoryClient] = None
- self._container_tag = _DEFAULT_CONTAINER_TAG
- self._turn_count = 0
+ self._container_tag, self._turn_count, self._write_enabled, self._active = _DEFAULT_CONTAINER_TAG, 0, True, False
self._prefetch_thread = self._sync_thread = self._write_thread = None # only _write_thread is ever started
- self._write_enabled = True
- self._active = False
self._session_turns: List[Dict[str, str]] = []
self._apply_config(_load_supermemory_config())
- self._base_url = _DEFAULT_BASE_URL # env var is only consulted in initialize()
- self._allowed_containers = []
+ self._base_url, self._allowed_containers = _DEFAULT_BASE_URL, [] # env var is only consulted in initialize()
def _apply_config(self, config: dict) -> None:
for key in ("auto_recall", "auto_capture", "max_recall_results", "profile_frequency", "capture_mode",
"search_mode", "entity_context", "api_timeout", "custom_containers", "custom_container_instructions"):
setattr(self, f"_{key}", config[key])
- self._base_url = _resolve_base_url(config["base_url"])
- self._enable_custom_containers = config["enable_custom_container_tags"]
+ self._base_url, self._enable_custom_containers = _resolve_base_url(config["base_url"]), config["enable_custom_container_tags"]
self._allowed_containers: List[str] = [self._container_tag] + list(self._custom_containers)
@property
@@ -364,43 +326,37 @@ class SupermemoryMemoryProvider(MemoryProvider):
return "supermemory"
def is_available(self) -> bool:
- # Key presence only, no SDK import check: the SDK is lazy-installed when the client is first constructed
- # in initialize(), so gating on importability here is a chicken-and-egg trap on sealed venvs. Mirrors honcho/mem0.
+ # Key presence only, no SDK import check: the SDK is lazy-installed in initialize(), so gating on
+ # importability here is a chicken-and-egg trap on sealed venvs. Mirrors honcho/mem0.
return bool(get_secret("SUPERMEMORY_API_KEY", ""))
def get_config_schema(self):
- # Only the API key is prompted during `hermes memory setup`; other options live in
- # $HERMES_HOME/supermemory.json or SUPERMEMORY_CONTAINER_TAG.
+ # Only the API key is prompted during `hermes memory setup`; other options live in supermemory.json / env.
return [{"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):
sanitized = dict(values or {})
- if "container_tag" in sanitized:
- sanitized["container_tag"] = _sanitize_tag(str(sanitized["container_tag"]))
- if "entity_context" in sanitized:
- sanitized["entity_context"] = _clamp_entity_context(str(sanitized["entity_context"]))
+ for key, fix in (("container_tag", _sanitize_tag), ("entity_context", _clamp_entity_context)):
+ if key in sanitized:
+ sanitized[key] = fix(str(sanitized[key]))
_save_supermemory_config(sanitized, hermes_home)
def get_status_config(self, provider_config: dict) -> dict:
from hermes_constants import get_hermes_home
- status = _probe_supermemory_connection(get_secret("SUPERMEMORY_API_KEY", "") or "", str(get_hermes_home()))
- return {"summary": _format_connection_summary(status)}
+ return {"summary": _format_connection_summary(_probe_supermemory_connection(get_secret("SUPERMEMORY_API_KEY", "") or "", str(get_hermes_home())))}
def post_setup(self, hermes_home: str, config: dict) -> None:
from hermes_cli.config import save_config
from hermes_cli.memory_setup import _prompt, _write_env_vars
-
print(f"\n Configuring supermemory:\n\n Get your API key at {_API_KEY_URL}\n")
existing = os.environ.get("SUPERMEMORY_API_KEY", "")
masked = f"...{existing[-4:]}" if len(existing) > 4 else "set"
val = _prompt(f"Supermemory API key (current: {masked}, blank to keep)" if existing else "Supermemory API key", secret=True)
- if not isinstance(config.get("memory"), dict):
- config["memory"] = {}
- config["memory"]["provider"] = self.name
+ memory = config["memory"] = config["memory"] if isinstance(config.get("memory"), dict) else {}
+ memory["provider"] = self.name
save_config(config)
if val:
_write_env_vars({"SUPERMEMORY_API_KEY": val}, hermes_home=hermes_home)
-
api_key = val or existing
# Make the freshly-entered key visible to the probe below. Single-profile only: under a multiplexed
# gateway, writing to the process-global environ would leak the key to sibling profiles and their subprocesses.
@@ -415,40 +371,27 @@ class SupermemoryMemoryProvider(MemoryProvider):
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())
- self._session_id = session_id
- self._turn_count = 0
+ self._session_id, self._turn_count, self._session_turns = session_id, 0, []
config = _load_supermemory_config(self._hermes_home)
self._api_key = get_secret("SUPERMEMORY_API_KEY", "") or ""
self._container_tag = _resolve_container_tag(config["container_tag"], kwargs.get("agent_identity", "default"))
self._apply_config(config)
- self._session_turns = []
self._write_enabled = kwargs.get("agent_context", "") not in {"cron", "flush", "subagent"}
- self._active = bool(self._api_key)
- self._client = None
- if self._active:
- try:
- self._client = _build_client(self._api_key, config, self._container_tag)
- except Exception:
- logger.warning("Supermemory initialization failed", exc_info=True)
- self._active = False
+ self._client = _quietly(lambda: _build_client(self._api_key, config, self._container_tag),
+ "Supermemory initialization failed", level=logging.WARNING) if self._api_key else None
+ self._active = self._client is not None
def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None:
self._turn_count = max(turn_number, 0)
def system_prompt_block(self) -> str:
- if not self._active:
- return ""
- lines = [
- "# Supermemory",
- f"Active. Container: {self._container_tag}.",
- "Use supermemory-search, supermemory-save, supermemory-forget, and supermemory-profile (aliases: supermemory_search, supermemory_store, supermemory_forget, supermemory_profile).",
- ]
+ lines = ["# Supermemory", f"Active. Container: {self._container_tag}.",
+ "Use supermemory-search, supermemory-save, supermemory-forget, and supermemory-profile (aliases: supermemory_search, supermemory_store, supermemory_forget, supermemory_profile)."]
if self._enable_custom_containers and self._custom_containers:
- lines.append(f"\nMulti-container mode enabled. Available containers: {', '.join(self._allowed_containers)}.")
- lines.append("Pass an optional container_tag to supermemory_search, supermemory_store, supermemory_forget, and supermemory_profile to target a specific container.")
- if self._custom_container_instructions:
- lines.append(f"\n{self._custom_container_instructions}")
- return "\n".join(lines)
+ lines += [f"\nMulti-container mode enabled. Available containers: {', '.join(self._allowed_containers)}.",
+ "Pass an optional container_tag to supermemory_search, supermemory_store, supermemory_forget, and supermemory_profile to target a specific container."]
+ lines += [f"\n{self._custom_container_instructions}"] if self._custom_container_instructions else []
+ return "\n".join(lines) if self._active else ""
def _can_write(self) -> bool:
return bool(self._active and self._write_enabled and self._client)
@@ -456,45 +399,37 @@ class SupermemoryMemoryProvider(MemoryProvider):
def prefetch(self, query: str, *, session_id: str = "") -> str:
if not self._active or not self._auto_recall or not self._client or not query.strip():
return ""
- try:
+ def _recall():
profile = self._client.get_profile(query=query[:200])
include_profile = self._turn_count <= 1 or (self._turn_count % self._profile_frequency == 0)
- return _format_prefetch_context(static_facts=profile["static"] if include_profile else [],
- dynamic_facts=profile["dynamic"] if include_profile else [],
- search_results=profile["search_results"], max_results=self._max_recall_results)
- except Exception:
- logger.debug("Supermemory prefetch failed", exc_info=True)
- return ""
+ return _format_prefetch_context(profile["static"] if include_profile else [], profile["dynamic"] if include_profile else [],
+ profile["search_results"], self._max_recall_results)
+ return _quietly(_recall, "Supermemory prefetch failed", default="")
def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
if not self._can_write() or not self._auto_capture:
return
- clean_user, clean_assistant = _clean_text_for_capture(user_content), _clean_text_for_capture(assistant_content)
- if clean_user or clean_assistant: # buffered for the single full-session document written at end/switch/shutdown
- self._session_turns.append({"user": clean_user, "assistant": clean_assistant})
+ turn = {"user": _clean_text_for_capture(user_content), "assistant": _clean_text_for_capture(assistant_content)}
+ if any(turn.values()): # buffered for the single full-session document written at end/switch/shutdown
+ self._session_turns.append(turn)
- def _ingest_session(self, session_id: str, messages: list[dict], metadata: dict, fail_msg: str, level: int = logging.DEBUG) -> None:
- try:
- self._client.ingest_conversation(session_id, messages, metadata={"type": "full_session", "session_id": session_id, **metadata})
- except Exception:
- logger.log(level, fail_msg, exc_info=True)
+ def _ingest(self, session_id: str, messages: list[dict], metadata: dict, fail_msg: str, level: int = logging.DEBUG) -> None:
+ metadata = {"type": "full_session", "session_id": session_id, **metadata}
+ _quietly(lambda: self._client.ingest_conversation(session_id, messages, metadata=metadata), fail_msg, level=level)
- def _ingest_buffered_turns(self, session_id: str, *, partial: bool, fail_msg: str) -> None:
- # message_count reports 2 per buffered turn regardless of empty sides.
- turns = self._session_turns
+ def _flush_turns(self, session_id: str, *, partial: bool, fail_msg: str) -> None:
+ turns = self._session_turns # message_count reports 2 per buffered turn regardless of empty sides
messages = [{"role": role, "content": t[role]} for t in turns for role in ("user", "assistant") if t.get(role)]
- self._ingest_session(session_id, messages, {"message_count": len(turns) * 2, "partial": partial}, fail_msg)
+ self._ingest(session_id, messages, {"message_count": len(turns) * 2, "partial": partial}, fail_msg)
def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
if not self._can_write() or not self._session_id:
return
- cleaned = [{"role": m.get("role"), "content": _clean_text_for_capture(str(m.get("content", "")))}
- for m in messages or [] if m.get("role") in {"user", "assistant"}]
- cleaned = [m for m in cleaned if m["content"]]
+ cleaned = [{"role": m.get("role"), "content": content} for m in messages or []
+ if m.get("role") in {"user", "assistant"} and (content := _clean_text_for_capture(str(m.get("content", ""))))]
if not cleaned or (len(cleaned) == 1 and len(cleaned[0]["content"]) < 20):
return
- self._ingest_session(self._session_id, cleaned, {"message_count": len(cleaned)}, "Supermemory session ingest failed",
- level=logging.WARNING)
+ self._ingest(self._session_id, cleaned, {"message_count": len(cleaned)}, "Supermemory session ingest failed", level=logging.WARNING)
self._session_turns = [] # so shutdown() doesn't duplicate on normal exit
def on_session_switch(self, new_session_id: str, *, parent_session_id: str = "", reset: bool = False, **kwargs) -> None:
@@ -502,7 +437,7 @@ class SupermemoryMemoryProvider(MemoryProvider):
old_session_id = self._session_id
if self._can_write():
if self._session_turns and old_session_id:
- self._ingest_buffered_turns(old_session_id, partial=not reset, fail_msg="Supermemory session-switch ingest failed")
+ self._flush_turns(old_session_id, partial=not reset, fail_msg="Supermemory session-switch ingest failed")
self._turn_count = 0
self._session_id = str(new_session_id or "").strip() or old_session_id
self._session_turns = []
@@ -510,46 +445,36 @@ class SupermemoryMemoryProvider(MemoryProvider):
def on_memory_write(self, action: str, target: str, content: str) -> None:
if not self._can_write() or action != "add" or not (content or "").strip():
return
-
- def _run():
- try:
- self._client.add_memory(content.strip(), metadata={"target": target, "type": "explicit_memory"},
- entity_context=self._entity_context)
- except Exception:
- logger.debug("Supermemory on_memory_write failed", exc_info=True)
-
if self._write_thread and self._write_thread.is_alive():
self._write_thread.join(timeout=2.0)
- self._write_thread = threading.Thread(target=_run, daemon=False, name="supermemory-memory-write")
+ self._write_thread = threading.Thread(
+ target=_quietly, daemon=False, name="supermemory-memory-write",
+ args=(lambda: self._client.add_memory(content.strip(), metadata={"target": target, "type": "explicit_memory"},
+ entity_context=self._entity_context), "Supermemory on_memory_write failed"))
self._write_thread.start()
def shutdown(self) -> None:
# Emergency fallback (crashes only). Buffer is cleared on normal on_session_end().
if self._can_write() and self._session_turns and self._session_id:
logger.warning("Supermemory: Saving session via shutdown (session=%s, turns=%d)", self._session_id, len(self._session_turns))
- self._ingest_buffered_turns(self._session_id, partial=True, fail_msg="Supermemory shutdown ingest failed")
+ self._flush_turns(self._session_id, partial=True, fail_msg="Supermemory shutdown ingest failed")
if self._write_thread and self._write_thread.is_alive():
self._write_thread.join(timeout=5.0)
self._prefetch_thread = self._sync_thread = self._write_thread = None
def get_tool_schemas(self) -> List[Dict[str, Any]]:
schemas = [json.loads(json.dumps(base)) for base in _BASE_SCHEMAS] # deep copies
- if self._enable_custom_containers:
- for schema in schemas: # multi-container mode: every tool takes an optional container_tag
- schema["parameters"]["properties"]["container_tag"] = {
- "type": "string",
- "description": f"Optional container tag. Allowed: {', '.join(self._allowed_containers)}. Defaults to primary ({self._container_tag}).",
- }
+ for schema in schemas if self._enable_custom_containers else (): # multi-container mode: every tool takes container_tag
+ schema["parameters"]["properties"]["container_tag"] = {
+ "type": "string", "description": f"Optional container tag. Allowed: {', '.join(self._allowed_containers)}. Defaults to primary ({self._container_tag})."}
# Kebab-case aliases are appended after all snake_case schemas (deep-copied, name swapped).
return schemas + [{**json.loads(json.dumps(s)), "name": _KEBAB_ALIASES[s["name"]]} for s in schemas]
def _tool_container_tag(self, args: dict) -> Optional[str]:
"""Validated container_tag from args; None = primary. Raises _TagError when not whitelisted."""
- tag = str(args.get("container_tag") or "").strip() if self._enable_custom_containers else ""
- if not tag:
- return None
- tag = _sanitize_tag(tag)
- if tag not in self._allowed_containers:
+ raw = str(args.get("container_tag") or "").strip() if self._enable_custom_containers else ""
+ tag = _sanitize_tag(raw) if raw else None
+ if tag and tag not in self._allowed_containers:
raise _TagError(f"Container tag '{tag}' is not allowed. Allowed: {', '.join(self._allowed_containers)}")
return tag
@@ -557,9 +482,7 @@ class SupermemoryMemoryProvider(MemoryProvider):
content = str(args.get("content") or "").strip()
if not content:
return tool_error("content is required")
- metadata = args.get("metadata") or {}
- if not isinstance(metadata, dict):
- metadata = {}
+ metadata = args.get("metadata") if isinstance(args.get("metadata"), dict) else {}
metadata.setdefault("type", _detect_category(content))
metadata.pop("source", None)
tag = self._tool_container_tag(args)
@@ -572,22 +495,19 @@ class SupermemoryMemoryProvider(MemoryProvider):
return tool_error("query is required")
limit = _clamp_number(args.get("limit", 5) or 5, 5, 1, 20, int)
tag = self._tool_container_tag(args)
- formatted = []
- for item in self._client.search_memories(query, limit=limit, container_tag=tag):
- pct = _similarity_pct(item.get("similarity"))
- formatted.append({"id": item.get("id", ""), "content": item.get("memory", ""), **({"similarity": pct} if pct is not None else {})})
- return _tagged({"results": formatted, "count": len(formatted)}, tag)
+ results = [{"id": i.get("id", ""), "content": i.get("memory", ""), **({"similarity": pct} if (pct := _similarity_pct(i.get("similarity"))) is not None else {})}
+ for i in self._client.search_memories(query, limit=limit, container_tag=tag)]
+ return _tagged({"results": results, "count": len(results)}, tag)
def _tool_forget(self, args: dict) -> dict | str:
- memory_id = str(args.get("id") or "").strip()
- query = str(args.get("query") or "").strip()
+ memory_id, query = str(args.get("id") or "").strip(), str(args.get("query") or "").strip()
if not memory_id and not query:
return tool_error("Provide either id or query")
tag = self._tool_container_tag(args) # not echoed in the response
- if memory_id:
- self._client.forget_memory(memory_id, container_tag=tag)
- return {"forgotten": True, "id": memory_id}
- return self._client.forget_by_query(query, container_tag=tag)
+ if not memory_id:
+ return self._client.forget_by_query(query, container_tag=tag)
+ self._client.forget_memory(memory_id, container_tag=tag)
+ return {"forgotten": True, "id": memory_id}
def _tool_profile(self, args: dict) -> dict:
tag = self._tool_container_tag(args)
@@ -605,10 +525,8 @@ class SupermemoryMemoryProvider(MemoryProvider):
handler, fail_prefix = self._TOOL_HANDLERS[tool_name]
try:
resp = handler(self, args)
- except _TagError as exc:
- return tool_error(str(exc))
except Exception as exc:
- return tool_error(f"{fail_prefix}: {exc}")
+ return tool_error(str(exc) if isinstance(exc, _TagError) else f"{fail_prefix}: {exc}")
return resp if isinstance(resp, str) else json.dumps(resp)
# snake_case tool name -> (handler, error prefix); kebab aliases are folded in via _ALIAS_TO_TOOL first.