From 04fdb78180d3370349b6eb38df021254035d099f Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:26:19 -0400 Subject: [PATCH 1/7] feat(convo): canonical write path for single verbatim exchanges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live agent integrations and their backfills need to file one conversation exchange at a time, but hand-rolling the upsert leaves drawers without hall / entities / filed_at / extract_mode metadata — silently invisible to hallway traversal, entity search, and the since/before date filters. file_conversation_exchange() builds the same metadata the convo miner writes, and make_exchange_drawer_id() moves the ID construction into ids.py per its single-source-of-truth contract (full-content hash, no prefix collisions; filed_at keeps repeated exchanges distinct). Co-Authored-By: Claude Fable 5 --- mempalace/convo_miner.py | 58 +++++++++++++++++++++++++++++++++++++++- mempalace/ids.py | 21 +++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/mempalace/convo_miner.py b/mempalace/convo_miner.py index 0ea784b..abe66b4 100644 --- a/mempalace/convo_miner.py +++ b/mempalace/convo_miner.py @@ -19,7 +19,12 @@ from collections import defaultdict from typing import Optional from .collision_scan import assert_no_collisions -from .ids import ID_RECIPE, make_convo_drawer_id, make_convo_sentinel_id +from .ids import ( + ID_RECIPE, + make_convo_drawer_id, + make_convo_sentinel_id, + make_exchange_drawer_id, +) from .normalize import normalize from .entities import entities_metadata from .palace import ( @@ -57,6 +62,57 @@ def _detect_hall_cached(content: str) -> str: return max(scores, key=scores.get) if scores else "general" +def file_conversation_exchange( + collection, + *, + wing: str, + room: str, + text: str, + source_file: str, + agent: str, + authored_at: Optional[str] = None, + extra_metadata: Optional[dict] = None, +) -> Optional[str]: + """File one verbatim conversation exchange as a single drawer. + + Canonical write path for live agent integrations (e.g. Hermes) and + their backfills — both must route here so routing, normalization, + and metadata conventions stay identical between live and historical + ingest. Builds the same metadata the convo miner writes so hallway + traversal, entity search, and since/before date filters see + integration drawers exactly like mined ones. + + ``extra_metadata`` lets callers append integration-specific fields + (e.g. ``source`` / ``session_id``) but cannot be used to *drop* the + canonical keys. Returns the drawer id, or None when ``text`` is + empty after stripping. + """ + text = (text or "").strip() + if not text: + return None + filed_at = datetime.now().isoformat() + drawer_id = make_exchange_drawer_id(wing, room, source_file, filed_at, text) + metadata = { + "wing": wing, + "room": room, + "hall": _detect_hall_cached(text), + "source_file": source_file, + "chunk_index": 0, + "added_by": agent, + "filed_at": filed_at, + "entities": entities_metadata(text), + "authored_at": authored_at if authored_at is not None else filed_at, + "ingest_mode": "convos", + "extract_mode": "exchange", + "normalize_version": NORMALIZE_VERSION, + "id_recipe": ID_RECIPE, + } + if extra_metadata: + metadata.update(extra_metadata) + collection.upsert(ids=[drawer_id], documents=[text], metadatas=[metadata]) + return drawer_id + + # File types that might contain conversations CONVO_EXTENSIONS = { ".txt", diff --git a/mempalace/ids.py b/mempalace/ids.py index de4c7ab..c9ad625 100644 --- a/mempalace/ids.py +++ b/mempalace/ids.py @@ -108,6 +108,27 @@ def make_convo_sentinel_id(source_file: str, extract_mode: str) -> str: return f"_reg_{_delimited_sha256((source_file, extract_mode), _HASH_TRUNC_DRAWER)}" +def make_exchange_drawer_id( + wing: str, room: str, source_file: str, filed_at: str, content: str +) -> str: + """Drawer ID for a single verbatim conversation exchange. + + Used by live agent integrations (e.g. Hermes) and their backfills via + ``convo_miner.file_conversation_exchange``. Hashes the FULL content, + not a prefix — prefix hashing collided on common openings ("User: hi + can you help me with…") and ChromaDB's upsert silently overwrote the + earlier drawer. ``filed_at`` is included so genuinely repeated + exchanges stay distinct drawers (verbatim always — repetition is + signal, not noise). + + Hash input is ``f"{source_file}|{filed_at}|{content}"``. + """ + return ( + f"drawer_{wing}_{room}_" + f"{_delimited_sha256((source_file, filed_at, content), _HASH_TRUNC_DRAWER)}" + ) + + def make_triple_id( sub_id: str, predicate: str, obj_id: str, valid_from: str, recorded_at: str ) -> str: From 9553ed99406424d5797949247b714f3dd0302c62 Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:26:19 -0400 Subject: [PATCH 2/7] feat(integrations): Hermes memory provider core MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Hermes provider from feat/hermes-integration, split out per review on #1684 — provider + tests only; backfill, the hermes install CLI, and docs follow in a stacked PR. Changes vs the original branch: - _file_turn routes through convo_miner.file_conversation_exchange() instead of a hand-rolled col.upsert, so live turns carry canonical drawer metadata and the ids.py ID recipe. - The backfill/live wing-routing parity test moves to the backfill PR. Co-Authored-By: Claude Fable 5 --- mempalace/integrations/__init__.py | 0 mempalace/integrations/hermes/__init__.py | 1393 +++++++++++++++++++++ tests/test_hermes_integration.py | 636 ++++++++++ 3 files changed, 2029 insertions(+) create mode 100644 mempalace/integrations/__init__.py create mode 100644 mempalace/integrations/hermes/__init__.py create mode 100644 tests/test_hermes_integration.py diff --git a/mempalace/integrations/__init__.py b/mempalace/integrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mempalace/integrations/hermes/__init__.py b/mempalace/integrations/hermes/__init__.py new file mode 100644 index 0000000..833b70f --- /dev/null +++ b/mempalace/integrations/hermes/__init__.py @@ -0,0 +1,1393 @@ +"""MemPalace memory provider for Hermes. + +Implements the Hermes ``MemoryProvider`` ABC (``agent/memory_provider.py``) +so MemPalace can be selected as ``memory.provider: mempalace`` in +``~/.hermes/config.yaml``. + +Design notes +------------ + +* ChromaDB access goes through ``mempalace.backends.chroma.ChromaBackend`` + rather than a raw ``chromadb.PersistentClient``. This ensures the + embedding function returned by ``mempalace.embedding.get_embedding_function`` + is bound to the collection, fixing the embedding-dimension mismatch that + silently broke the three earlier Hermes-side PRs (NousResearch/hermes-agent + #5671, #12203, #9761) on existing palaces. + +* Per-turn writes go through a bounded background queue. The agent loop + never blocks on ChromaDB or SQLite. + +* The provider is **inactive** under ``agent_context in {"cron", "flush"}`` + or ``platform == "cron"``. Cron-context turns are system-generated and + would otherwise corrupt the user's representation. + +* Configuration precedence: ``$HERMES_HOME/mempalace.json`` is read + first, then env vars override (``MEMPALACE_PALACE_PATH``, + ``MEMPALACE_IDENTITY_PATH``, ``MEMPALACE_WING``). An empty env var + is ignored — ``export MEMPALACE_WING=`` is intent to unset. Defaults + fill in anything still missing. ``collection_name`` is intentionally + not user-configurable here: the provider writes through + ``self._collection_name`` while ``search_memories`` (used by + ``prefetch`` and ``_tool_search``) reads its own configured collection + name from ``~/.mempalace/config.json``, and exposing two ways to set + it would let the two diverge silently. + +* ``~/.mempalace/identity.txt`` (L0) and ``~/.mempalace/wing_config.json`` + are loaded if present but never created here. Run + ``mempalace init `` to generate them. +""" + +from __future__ import annotations + +import json +import logging +import os +import queue +import re +import threading +from collections import deque +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +from mempalace.backends.chroma import ChromaBackend +from mempalace.convo_miner import file_conversation_exchange +from mempalace.knowledge_graph import KnowledgeGraph +from mempalace.layers import MemoryStack +from mempalace.searcher import search_memories + +# When this plugin is loaded by Hermes, ``agent.memory_provider`` is on the +# import path. When mempalace's own test suite imports this module (or a tool +# inspects it without Hermes installed) the import fails — fall back to a stub +# so the module still imports cleanly. ``isinstance(provider, MemoryProvider)`` +# checks remain meaningful because the real ABC binds at plugin-load time. +try: + from agent.memory_provider import MemoryProvider # type: ignore[import-not-found] +except ImportError: # pragma: no cover - Hermes not installed + + class MemoryProvider: # type: ignore[no-redef] + """Stub used when ``agent.memory_provider`` cannot be imported.""" + + +logger = logging.getLogger("mempalace.hermes") + + +def _match_wing_by_keywords(text: str, wing_config: Dict[str, Any]) -> str: + """Return the first wing whose keywords match a whole word in ``text``. + + Word boundaries matter — bare substring matching routes turns mentioning + ``said`` into a wing whose keyword is ``ai``. Fall back to ``wing_general``. + + Lives at module scope so ``backfill.py`` can use exactly the same matching + logic the live provider uses (it cannot import from this file as a + relative import — it's run by ``importlib.util.spec_from_file_location``). + The duplicate below in ``backfill.py`` must be kept in sync. + """ + if not wing_config: + return "wing_general" + text_lower = text.lower() + for wing_name, wing_def in wing_config.items(): + keywords = wing_def.get("keywords", []) if isinstance(wing_def, dict) else [] + for kw in keywords: + if not kw: + continue + pattern = r"\b" + re.escape(kw.lower()) + r"\b" + if re.search(pattern, text_lower): + return wing_name + return "wing_general" + + +def _normalize_content(content: Any) -> str: + """Flatten an Anthropic/OpenAI ``content`` field to a plain string. + + Hermes turns frequently carry ``content`` as a list of typed parts + (``[{"type": "text", "text": "..."}, {"type": "tool_use", ...}]``). + A naive ``f"User: {content}"`` would persist the literal ``repr`` of + the list and corrupt semantic search recall over the palace. Concatenate + the ``text`` blocks (and surface a tool-use marker so search hits still + say a tool was called) instead. + """ + if not content: + return "" + if isinstance(content, str): + return content + if isinstance(content, list): + parts: List[str] = [] + for block in content: + if not isinstance(block, dict): + parts.append(str(block)) + continue + btype = block.get("type", "") + if btype == "text": + text = block.get("text", "") + if text: + parts.append(text) + elif btype == "tool_use": + name = block.get("name", "?") + parts.append(f"[tool_use: {name}]") + elif btype == "tool_result": + result = block.get("content") + parts.append(f"[tool_result] {_normalize_content(result)}") + else: + # Unknown block type — fall back to text field or skip. + text = block.get("text", "") + if text: + parts.append(text) + return "\n".join(parts) + return str(content) + + +# --------------------------------------------------------------------------- +# Tool schemas (OpenAI function-calling format; no `handler` field — dispatch +# happens via ``MempalaceProvider.handle_tool_call``). +# --------------------------------------------------------------------------- + +TOOL_SCHEMAS: List[Dict[str, Any]] = [ + { + "name": "mempalace_search", + "description": "Semantic search across the palace. Returns verbatim drawers ranked by relevance.", + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "Natural language search query."}, + "wing": {"type": "string", "description": "Limit results to one wing (optional)."}, + "room": { + "type": "string", + "description": "Limit results to one room within a wing (optional).", + }, + "n_results": { + "type": "integer", + "description": "Number of results (1-50, default 5).", + }, + }, + "required": ["query"], + }, + }, + { + "name": "mempalace_status", + "description": "Palace overview: total drawers, per-wing counts, palace path.", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "mempalace_list_wings", + "description": "List all wings with their drawer counts.", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "mempalace_list_rooms", + "description": "List rooms (and counts) within a wing.", + "parameters": { + "type": "object", + "properties": { + "wing": {"type": "string", "description": "Wing name."}, + }, + "required": ["wing"], + }, + }, + { + "name": "mempalace_kg_query", + "description": "Query the knowledge graph for relationships involving an entity, with optional time filtering.", + "parameters": { + "type": "object", + "properties": { + "entity": {"type": "string", "description": "Entity name."}, + "since": {"type": "string", "description": "ISO date lower bound (optional)."}, + }, + "required": ["entity"], + }, + }, + { + "name": "mempalace_kg_add", + "description": "Add a (subject, predicate, object) fact to the knowledge graph.", + "parameters": { + "type": "object", + "properties": { + "subject": {"type": "string"}, + "predicate": {"type": "string"}, + "object": {"type": "string"}, + }, + "required": ["subject", "predicate", "object"], + }, + }, + { + "name": "mempalace_diary_write", + "description": "Append an AAAK diary entry.", + "parameters": { + "type": "object", + "properties": { + "entry": {"type": "string", "description": "Diary entry text."}, + }, + "required": ["entry"], + }, + }, + { + "name": "mempalace_diary_read", + "description": "Read the most recent diary entries.", + "parameters": { + "type": "object", + "properties": { + "n": { + "type": "integer", + "description": "Number of entries to return (default 10).", + }, + }, + }, + }, + { + "name": "mempalace_add_drawer", + "description": ( + "File a verbatim drawer into the palace. Use for explicit " + "structured content the user dictates or you decide to " + "persist — the per-turn auto-filing happens separately." + ), + "parameters": { + "type": "object", + "properties": { + "wing": {"type": "string", "description": "Wing name."}, + "room": {"type": "string", "description": "Room within the wing."}, + "content": {"type": "string", "description": "Verbatim drawer content."}, + "source_file": { + "type": "string", + "description": "Optional source-file annotation.", + }, + }, + "required": ["wing", "room", "content"], + }, + }, + { + "name": "mempalace_update_drawer", + "description": ( + "Edit an existing drawer in place. Prefer adding a new " + "drawer that supersedes the old one — mempalace is " + "append-first." + ), + "parameters": { + "type": "object", + "properties": { + "drawer_id": {"type": "string"}, + "content": {"type": "string", "description": "New verbatim content (optional)."}, + "wing": {"type": "string", "description": "New wing (optional)."}, + "room": {"type": "string", "description": "New room (optional)."}, + }, + "required": ["drawer_id"], + }, + }, + { + "name": "mempalace_delete_drawer", + "description": ( + "Remove a drawer. Reserve for PII cleanup or correcting a " + "wrong filing — mempalace's design prefers superseding adds." + ), + "parameters": { + "type": "object", + "properties": {"drawer_id": {"type": "string"}}, + "required": ["drawer_id"], + }, + }, + { + "name": "mempalace_list_drawers", + "description": "List drawers in a wing/room with their previews.", + "parameters": { + "type": "object", + "properties": { + "wing": {"type": "string"}, + "room": {"type": "string"}, + "limit": {"type": "integer", "description": "Default 20."}, + "offset": {"type": "integer", "description": "Default 0."}, + }, + }, + }, + { + "name": "mempalace_get_drawer", + "description": "Fetch a drawer's full verbatim content by id.", + "parameters": { + "type": "object", + "properties": {"drawer_id": {"type": "string"}}, + "required": ["drawer_id"], + }, + }, + { + "name": "mempalace_check_duplicate", + "description": ( + "Check whether content similar to the given text already " + "exists in the palace before filing. Returns closest match " + "and similarity score." + ), + "parameters": { + "type": "object", + "properties": { + "content": {"type": "string"}, + "threshold": {"type": "number", "description": "Default 0.9."}, + }, + "required": ["content"], + }, + }, + { + "name": "mempalace_kg_invalidate", + "description": ( + "Mark a (subject, predicate, object) fact as no longer valid " + "from a given date — per the palace protocol's step 5." + ), + "parameters": { + "type": "object", + "properties": { + "subject": {"type": "string"}, + "predicate": {"type": "string"}, + "object": {"type": "string"}, + "ended": { + "type": "string", + "description": ( + "ISO date the fact stopped being true (optional, defaults to now)." + ), + }, + }, + "required": ["subject", "predicate", "object"], + }, + }, + { + "name": "mempalace_kg_timeline", + "description": "Full temporal timeline for an entity in the knowledge graph.", + "parameters": { + "type": "object", + "properties": { + "entity": { + "type": "string", + "description": "Entity name (optional — all entities if omitted).", + }, + }, + }, + }, + { + "name": "mempalace_kg_stats", + "description": "Knowledge graph summary statistics.", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "mempalace_get_taxonomy", + "description": "Full wing → room → drawer-count tree of the palace.", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "mempalace_get_aaak_spec", + "description": ( + "Return the full AAAK compression dialect specification " + "(also injected in the wake-up block)." + ), + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "mempalace_traverse", + "description": "Traverse the room graph from a starting room, following hallway links.", + "parameters": { + "type": "object", + "properties": { + "start_room": {"type": "string"}, + "max_hops": {"type": "integer", "description": "Default 2."}, + }, + "required": ["start_room"], + }, + }, + { + "name": "mempalace_graph_stats", + "description": "Palace graph statistics — rooms, hallways, cross-wing tunnels.", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "mempalace_find_tunnels", + "description": ( + "Find cross-wing tunnels — direct semantic links between rooms in different wings." + ), + "parameters": { + "type": "object", + "properties": { + "wing_a": {"type": "string"}, + "wing_b": {"type": "string"}, + }, + }, + }, + { + "name": "mempalace_create_tunnel", + "description": ( + "Create a tunnel between two rooms across wings to bridge cross-cutting entities." + ), + "parameters": { + "type": "object", + "properties": { + "wing_a": {"type": "string"}, + "room_a": {"type": "string"}, + "wing_b": {"type": "string"}, + "room_b": {"type": "string"}, + "description": {"type": "string"}, + }, + "required": ["wing_a", "room_a", "wing_b", "room_b"], + }, + }, + { + "name": "mempalace_list_tunnels", + "description": "List tunnels, optionally scoped to a wing.", + "parameters": { + "type": "object", + "properties": {"wing": {"type": "string"}}, + }, + }, + { + "name": "mempalace_delete_tunnel", + "description": "Remove a tunnel by id.", + "parameters": { + "type": "object", + "properties": {"tunnel_id": {"type": "string"}}, + "required": ["tunnel_id"], + }, + }, + { + "name": "mempalace_follow_tunnels", + "description": ( + "Follow tunnels outward from a (wing, room) pair to discover connected rooms." + ), + "parameters": { + "type": "object", + "properties": { + "wing": {"type": "string"}, + "room": {"type": "string"}, + }, + "required": ["wing", "room"], + }, + }, + { + "name": "mempalace_memories_filed_away", + "description": "Show drawers filed (with metadata) during the current session.", + "parameters": {"type": "object", "properties": {}}, + }, +] + + +# --------------------------------------------------------------------------- +# Provider +# --------------------------------------------------------------------------- + + +class MempalaceProvider(MemoryProvider): # type: ignore[misc] + """Hermes memory provider backed by MemPalace.""" + + DEFAULT_COLLECTION_NAME = "mempalace_drawers" + DEFAULT_PALACE_PATH = "~/.mempalace/palace" + DEFAULT_IDENTITY_PATH = "~/.mempalace/identity.txt" + WORKER_QUEUE_MAX = 500 + # Cap for status-style metadata scans. On large palaces (200k+ drawers) + # an unbounded ``col.get(include=["metadatas"])`` would materialize every + # row into Python memory just to compute counts — multi-second hangs and + # OOM risk on small hosts. Above this cap, breakdowns are sampled from + # the first ``STATUS_SCAN_LIMIT`` drawers and the response carries + # ``truncated: True`` plus a ``scanned`` count so the caller knows + # exactly how partial the view is and can compute coverage against the + # palace total it already has. + STATUS_SCAN_LIMIT = 5000 + + def __init__(self) -> None: + # Config + lifecycle state + self._config: Dict[str, Any] = {} + self._palace_path: str = "" + self._collection_name: str = self.DEFAULT_COLLECTION_NAME + self._wing_config: Dict[str, Any] = {} + self._identity: str = "" + self._wake_up_cache: str = "" + self._initialized = False + self._cron_skipped = False + + # Per-session bookkeeping + self._session_id: str = "" + self._hermes_home: str = "" + self._turn_count = 0 + + # ChromaDB access through mempalace's own backend (matches embedding + # function, fixes the dim-mismatch bug from prior PRs). + self._backend = None + self._collection = None + self._collection_lock = threading.Lock() + + # Background worker for non-blocking writes. + self._worker_queue: queue.Queue = queue.Queue(maxsize=self.WORKER_QUEUE_MAX) + self._worker_thread: Optional[threading.Thread] = None + self._worker_stop = threading.Event() + + # initialize() must be serialised against concurrent re-entries so we + # don't spawn two worker threads sharing one queue. + self._init_lock = threading.Lock() + + # ----- Required ABC ------------------------------------------------------ + + @property + def name(self) -> str: + return "mempalace" + + def is_available(self) -> bool: + """Always True — module-level imports prove mempalace is installed. + + If mempalace were missing, ``import`` of this module would have failed + before Hermes' plugin loader called ``is_available``. The check is + kept for ABC conformance and so a future config flag can disable the + provider here without surgery elsewhere. + """ + return True + + def initialize(self, session_id: str, **kwargs: Any) -> None: + # System-generated contexts (cron, flush) would corrupt user representation. + agent_context = kwargs.get("agent_context", "") + platform = kwargs.get("platform", "cli") + if agent_context in {"cron", "flush"} or platform == "cron": + logger.debug( + "MemPalace inactive: agent_context=%s, platform=%s", + agent_context, + platform, + ) + with self._init_lock: + self._cron_skipped = True + return + + # Serialise the rest: producers all read _initialized / _collection / + # _worker_thread; a re-entrant initialize() must not duplicate the + # worker or leave torn-up state visible to a parallel sync_turn. + with self._init_lock: + # Clear the cron-skip flag — a previous cron-context initialize on + # the same instance must not leave the provider permanently inert. + self._cron_skipped = False + + self._session_id = session_id or "" + self._hermes_home = str(kwargs.get("hermes_home", "") or "") + + self._config = self._load_config() + self._palace_path = str( + Path(self._config.get("palace_path", self.DEFAULT_PALACE_PATH)).expanduser() + ) + # Collection name is intentionally **not** configurable. ``_file_turn`` + # writes through ``self._collection``; ``prefetch`` / ``_tool_search`` + # go through ``search_memories``, which reads its own configured + # collection name from ``~/.mempalace/config.json``. Exposing two + # ways to set the name invites write-here, read-there mismatches + # that silently make the provider look mute. + self._collection_name = self.DEFAULT_COLLECTION_NAME + + self._load_wing_config() + self._load_identity() + + # Backend init: failures (slow disk, locked SQLite, missing palace) + # must not hang Hermes startup. The agent runs without palace + # context until the next successful initialize(). + backend_ready = False + try: + self._backend = ChromaBackend() + with self._collection_lock: + self._collection = self._backend.get_or_create_collection( + self._palace_path, + self._collection_name, + ) + logger.info( + "MemPalace: collection '%s' ready (palace=%s)", + self._collection_name, + self._palace_path, + ) + backend_ready = True + except Exception as exc: + logger.warning("MemPalace backend init failed: %s", exc) + with self._collection_lock: + self._collection = None + + # Background worker for filing — only when the backend opened. + # Starting a worker against a None collection invites the + # on_pre_compress / sync_turn data-loss path where the hint + # promises persistence the worker can't deliver. + if backend_ready and ( + self._worker_thread is None or not self._worker_thread.is_alive() + ): + self._worker_stop.clear() + self._worker_thread = threading.Thread( + target=self._background_worker, + daemon=True, + name="mempalace-worker", + ) + self._worker_thread.start() + + # Warm the wake-up cache without blocking startup, but only if the + # backend is up — otherwise MemoryStack reads a half-set palace. + if backend_ready: + threading.Thread( + target=self._refresh_wake_up_cache, + daemon=True, + name="mempalace-wakeup", + ).start() + + # ``_initialized`` reflects readiness — get_tool_schemas / + # handle_tool_call / prefetch / sync_turn / on_pre_compress key + # off this. A failed backend init leaves it False so callers see + # a uniformly inactive provider rather than half-broken state. + self._initialized = backend_ready + + def get_tool_schemas(self) -> List[Dict[str, Any]]: + # Schemas describe the *interface*, not runtime readiness. Hermes' + # ``agent.memory_manager._register_provider`` snapshots schemas at + # registration time (BEFORE ``initialize()`` runs) to build its + # tool-name → provider routing table; if we returned ``[]`` there, + # the dispatcher would never learn our tool names and every later + # call would hit ``"Unknown tool: "`` from the dispatcher + # without reaching ``handle_tool_call`` at all. Backend readiness + # is checked at call time in ``handle_tool_call``. + if self._cron_skipped: + return [] + return list(TOOL_SCHEMAS) + + # ----- Optional: prompt + recall ---------------------------------------- + + def system_prompt_block(self) -> str: + if self._cron_skipped or not self._initialized: + return "" + if not self._identity and not self._wake_up_cache: + return "" + parts = ["# MemPalace context"] + if self._identity: + parts.append(self._identity) + if self._wake_up_cache: + parts.append(self._wake_up_cache) + return "\n\n".join(parts) + + def prefetch(self, query: str, *, session_id: str = "") -> str: + if self._cron_skipped or not self._initialized or not query: + return "" + try: + n = max(1, min(int(self._config.get("n_prefetch", 3)), 20)) + result = search_memories( + query, + palace_path=self._palace_path, + n_results=n, + ) + hits = result.get("results", []) if isinstance(result, dict) else [] + if not hits: + return "" + lines = ["## MemPalace — relevant context"] + for r in hits: + wing = r.get("wing", "") + room = r.get("room", "") + tag = f"[{wing}/{room}] " if wing else "" + text = (r.get("text") or "").strip() + if text: + lines.append(f"{tag}{text}") + return "\n\n".join(lines) + except Exception as exc: + logger.debug("MemPalace prefetch error: %s", exc) + return "" + + def sync_turn( + self, + user_content: str, + assistant_content: str, + *, + session_id: str = "", + messages: Optional[List[Dict[str, Any]]] = None, + ) -> None: + if self._cron_skipped or not self._initialized: + return + user_text = _normalize_content(user_content) + assistant_text = _normalize_content(assistant_content) + if not user_text and not assistant_text: + return + try: + self._worker_queue.put_nowait( + ( + "file_turn", + { + "user": user_text, + "assistant": assistant_text, + "session_id": session_id or self._session_id, + }, + ) + ) + except queue.Full: + # Loud, not silent: the verbatim invariant is what mempalace sells. + # If the queue saturates we want operators to see it. + logger.warning( + "MemPalace worker queue full (maxsize=%d) — turn dropped; " + "writes likely stalled on disk or ChromaDB", + self.WORKER_QUEUE_MAX, + ) + + # ----- Optional lifecycle hooks ---------------------------------------- + + def on_turn_start(self, turn_number: int, message: str, **kwargs: Any) -> None: + self._turn_count = turn_number + + def on_session_end(self, messages: List[Dict[str, Any]]) -> None: + # Skip when the provider never came up — enqueueing to a queue whose + # worker never started would silently fill the bounded buffer with + # tasks that can never drain. + if self._cron_skipped or not self._initialized: + return + try: + self._worker_queue.put_nowait( + ( + "session_end", + {"messages": list(messages or []), "session_id": self._session_id}, + ) + ) + except queue.Full: + logger.warning( + "MemPalace queue full at session_end — %d messages will not be filed", + len(messages or []), + ) + # Regenerate the AAAK wake-up cache for the next session. + threading.Thread( + target=self._refresh_wake_up_cache, + daemon=True, + name="mempalace-wakeup", + ).start() + + def on_session_switch( + self, + new_session_id: str, + *, + parent_session_id: str = "", + reset: bool = False, + rewound: bool = False, + **kwargs: Any, + ) -> None: + # Repoint subsequent writes at the new session. /reset and /new flush + # per-session counters; /resume and /branch keep them. + self._session_id = new_session_id or "" + if reset: + self._turn_count = 0 + + def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str: + """File the messages about to be discarded and signal verbatim persistence. + + The hint is **only** returned when the worker can actually persist the + payload — if the backend never came up or the queue is saturated, we + say nothing so the summarizer falls back to its default conservative + discarding rather than acting on a false promise. + """ + if self._cron_skipped or not self._initialized: + return "" + try: + self._worker_queue.put_nowait( + ( + "pre_compress", + {"messages": list(messages or []), "session_id": self._session_id}, + ) + ) + except queue.Full: + logger.warning( + "MemPalace queue full at pre_compress — %d messages will not be filed", + len(messages or []), + ) + return "" + return ( + "MemPalace has filed every message in this window verbatim. " + "Compressed content remains searchable via the `mempalace_search` " + "tool — the summarizer can be aggressive about discarding raw turns." + ) + + def on_memory_write( + self, + action: str, + target: str, + content: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> None: + if ( + self._cron_skipped + or not self._initialized # worker isn't running; queueing leaks + or action != "add" + or target != "user" + or not content + ): + return + try: + self._worker_queue.put_nowait( + ( + "mem_write", + {"content": content, "metadata": dict(metadata or {})}, + ) + ) + except queue.Full: + logger.warning("MemPalace queue full at memory_write — entry dropped") + + def on_delegation( + self, + task: str, + result: str, + *, + child_session_id: str = "", + **kwargs: Any, + ) -> None: + # Record the (task, result) pair as a synthetic turn so the parent + # session's recall surfaces the delegated work. + self.sync_turn( + f"[delegated task]\n{task}", + f"[subagent {child_session_id} returned]\n{result}", + session_id=self._session_id, + ) + + # ----- Tool dispatch --------------------------------------------------- + + def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs: Any) -> str: + if self._cron_skipped: + return json.dumps({"error": "MemPalace not active (cron context)."}) + if not self._initialized: + return json.dumps({"error": "MemPalace not initialized."}) + args = args or {} + try: + if tool_name == "mempalace_search": + return json.dumps( + self._tool_search( + query=args.get("query", ""), + wing=args.get("wing"), + room=args.get("room"), + n_results=int(args.get("n_results", 5)), + ) + ) + if tool_name == "mempalace_status": + return json.dumps(self._tool_status()) + if tool_name == "mempalace_list_wings": + return json.dumps(self._tool_list_wings()) + if tool_name == "mempalace_list_rooms": + return json.dumps(self._tool_list_rooms(args.get("wing", ""))) + if tool_name == "mempalace_kg_query": + return json.dumps( + self._tool_kg_query( + entity=args.get("entity", ""), + since=args.get("since"), + ) + ) + if tool_name == "mempalace_kg_add": + return json.dumps( + self._tool_kg_add( + subject=args.get("subject", ""), + predicate=args.get("predicate", ""), + obj=args.get("object", ""), + ) + ) + if tool_name == "mempalace_diary_write": + return json.dumps(self._tool_diary_write(args.get("entry", ""))) + if tool_name == "mempalace_diary_read": + return json.dumps(self._tool_diary_read(int(args.get("n", 10)))) + + # Tools that delegate directly to ``mempalace.mcp_server``'s + # public ``tool_*`` entry points. These share mempalace's own + # config for palace_path resolution rather than this plugin's + # ``self._palace_path`` — a known asymmetry that the original + # eight tools above don't share. In the common case (default + # palace at ``~/.mempalace/palace``) both resolve to the same + # place. + # New tools (everything that has a matching ``tool_*`` in + # mempalace.mcp_server) dispatch by name derivation. One + # mempalace asymmetry to remap: ``mempalace_traverse`` maps to + # ``tool_traverse_graph`` on the mcp_server side. + result = self._dispatch_mcp_passthrough(tool_name, args) + if result is not None: + return result + + return json.dumps({"error": f"Unknown tool: {tool_name}"}) + except Exception as exc: + logger.exception("MemPalace tool %s failed", tool_name) + return json.dumps({"error": f"{tool_name} failed: {exc}"}) + + # Tools that need name-remapping when dispatching to + # ``mempalace.mcp_server.tool_*``. Everything else uses + # ``tool_name.replace("mempalace_", "tool_", 1)`` straight up. + _MCP_FUNC_REMAP: Dict[str, str] = { + "mempalace_traverse": "tool_traverse_graph", + } + + # Allowlist of tools that route through the mcp_server passthrough. + # Anything NOT here either has explicit handling above (the original + # eight ``_tool_*`` methods) or returns ``"Unknown tool"``. Using an + # allowlist (rather than ``getattr(_mp_mcp, name, None)`` only) keeps + # mempalace's admin / internal tool_* functions hidden from this + # plugin's surface even if a future caller passes their names. + _MCP_PASSTHROUGH_TOOLS = frozenset( + { + "mempalace_add_drawer", + "mempalace_update_drawer", + "mempalace_delete_drawer", + "mempalace_list_drawers", + "mempalace_get_drawer", + "mempalace_check_duplicate", + "mempalace_kg_invalidate", + "mempalace_kg_timeline", + "mempalace_kg_stats", + "mempalace_get_taxonomy", + "mempalace_get_aaak_spec", + "mempalace_traverse", + "mempalace_graph_stats", + "mempalace_find_tunnels", + "mempalace_create_tunnel", + "mempalace_list_tunnels", + "mempalace_delete_tunnel", + "mempalace_follow_tunnels", + "mempalace_memories_filed_away", + } + ) + + def _dispatch_mcp_passthrough(self, tool_name: str, args: Dict[str, Any]) -> Optional[str]: + """Forward an allowlisted tool to its mempalace.mcp_server entry. + + Returns the JSON-encoded result, or ``None`` if the tool isn't in + the allowlist (so the caller can fall through to the standard + ``"Unknown tool"`` error). + """ + if tool_name not in self._MCP_PASSTHROUGH_TOOLS: + return None + from mempalace import mcp_server as _mp_mcp + + func_name = self._MCP_FUNC_REMAP.get(tool_name, tool_name.replace("mempalace_", "tool_", 1)) + func = getattr(_mp_mcp, func_name, None) + if func is None: + return json.dumps({"error": f"{tool_name}: mempalace.mcp_server.{func_name} not found"}) + # ``add_drawer`` is the only tool that needs a client-side default + # — tag agent-originated drawers so they're distinguishable from + # miner-ingested ones. + if tool_name == "mempalace_add_drawer": + args.setdefault("added_by", "hermes") + return json.dumps(func(**args)) + + # ----- Setup wizard integration ---------------------------------------- + + def get_config_schema(self) -> List[Dict[str, Any]]: + return [ + { + "key": "palace_path", + "description": "Path to palace directory.", + "default": self.DEFAULT_PALACE_PATH, + }, + { + "key": "identity_path", + "description": "Path to identity.txt (L0 wake-up layer).", + "default": self.DEFAULT_IDENTITY_PATH, + }, + { + "key": "wing", + "description": "Default wing for filing (omit to auto-classify via wing_config.json).", + }, + { + "key": "n_prefetch", + "description": "Number of search results to inject per turn.", + "default": 3, + }, + ] + + def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: + config_path = Path(hermes_home) / "mempalace.json" + existing: Dict[str, Any] = {} + if config_path.exists(): + try: + existing = json.loads(config_path.read_text()) + except Exception as exc: + logger.debug("MemPalace config read failed: %s", exc) + existing.update(values) + config_path.write_text(json.dumps(existing, indent=2) + "\n") + + def post_setup(self, hermes_home: str, config: Dict[str, Any]) -> None: + print() + print("MemPalace provider installed. To finish setup:") + print(" 1. mempalace init # generates ~/.mempalace/") + print(" 2. (optional) edit ~/.mempalace/identity.txt to seed L0 wake-up context") + print() + + # ----- Shutdown -------------------------------------------------------- + + def shutdown(self) -> None: + self._worker_stop.set() + if self._worker_thread and self._worker_thread.is_alive(): + self._worker_thread.join(timeout=5.0) + if self._worker_thread.is_alive(): + logger.warning("MemPalace worker did not drain within shutdown timeout") + + # ----- Internal: config + wing routing + identity --------------------- + + def _load_config(self) -> Dict[str, Any]: + config: Dict[str, Any] = {} + if self._hermes_home: + config_path = Path(self._hermes_home) / "mempalace.json" + if config_path.exists(): + try: + config.update(json.loads(config_path.read_text())) + except Exception as exc: + logger.debug("MemPalace config load failed: %s", exc) + for env_key, conf_key in ( + ("MEMPALACE_PALACE_PATH", "palace_path"), + ("MEMPALACE_IDENTITY_PATH", "identity_path"), + ("MEMPALACE_WING", "wing"), + ): + # Only honor a non-empty env var. ``export MEMPALACE_WING=`` (e.g. + # from a deactivation script) is intent to *unset*, not to set the + # wing to the empty string. + value = os.environ.get(env_key) + if value: + config[conf_key] = value + return config + + def _load_wing_config(self) -> None: + wing_config_path = Path(self._palace_path).parent / "wing_config.json" + if not wing_config_path.exists(): + self._wing_config = {} + logger.debug("MemPalace: no wing_config.json — run `mempalace init` to configure wings") + return + try: + with open(wing_config_path) as f: + self._wing_config = (json.load(f) or {}).get("wings", {}) + except Exception as exc: + logger.warning("MemPalace wing config load failed: %s", exc) + self._wing_config = {} + + def _load_identity(self) -> None: + identity_path = Path( + self._config.get("identity_path", self.DEFAULT_IDENTITY_PATH) + ).expanduser() + if not identity_path.exists(): + self._identity = "" + return + try: + self._identity = identity_path.read_text(encoding="utf-8").strip() + except Exception as exc: + logger.warning("MemPalace identity load failed: %s", exc) + self._identity = "" + + def _refresh_wake_up_cache(self) -> None: + try: + stack = MemoryStack(palace_path=self._palace_path) + wing = self._config.get("wing") or "" + self._wake_up_cache = stack.wake_up(wing=wing) or "" + except Exception as exc: + logger.debug("MemPalace wake-up refresh error: %s", exc) + self._wake_up_cache = "" + + def _classify_wing(self, text: str) -> str: + # If the user pinned a default wing in config, honor it without running + # keyword classification. The config field's whole purpose is "don't + # auto-classify this profile's turns" — silently keyword-routing on + # top of it would make the setting functionally dead. + forced = self._config.get("wing") + if forced: + return str(forced) + return _match_wing_by_keywords(text, self._wing_config) + + # ----- Internal: filing + background worker -------------------------- + + def _file_turn(self, payload: Dict[str, Any]) -> None: + user_msg = payload.get("user", "") or "" + assistant_msg = payload.get("assistant", "") or "" + if not user_msg and not assistant_msg: + return + with self._collection_lock: + col = self._collection + if col is None: + return + try: + text = f"User: {user_msg}\n\nAssistant: {assistant_msg}".strip() + wing = self._classify_wing(text) + # Always file under a stable room name ("conversations"). Using + # session_id here would mint one room per session — pollutes + # ``mempalace_list_rooms`` and splits live writes from backfill + # drawers (which also write to "conversations"). The session id + # stays available on the dedicated metadata field below. + session_id = payload.get("session_id") or "" + extra: Dict[str, Any] = {"source": "hermes"} + if session_id: + extra["session_id"] = session_id + file_conversation_exchange( + col, + wing=wing, + room="conversations", + text=text, + source_file=f"hermes-session:{session_id or 'unknown'}", + agent="hermes", + extra_metadata=extra, + ) + except Exception as exc: + logger.debug("MemPalace _file_turn error: %s", exc) + + def _mine_session(self, payload: Dict[str, Any]) -> None: + messages = payload.get("messages", []) or [] + session_id = payload.get("session_id", "") or "" + try: + for idx, msg in enumerate(messages): + if msg.get("role") != "user": + continue + # Same content normalization ``sync_turn`` and ``pre_compress`` + # use — list-shaped Anthropic content must not be persisted as + # its ``repr``. + content = _normalize_content(msg.get("content")) + if not content: + continue + assistant_content = "" + if idx + 1 < len(messages) and messages[idx + 1].get("role") == "assistant": + assistant_content = _normalize_content(messages[idx + 1].get("content")) + self._file_turn( + { + "user": content, + "assistant": assistant_content, + "session_id": session_id, + } + ) + except Exception as exc: + logger.debug("MemPalace _mine_session error: %s", exc) + + def _mirror_mem_write(self, payload: Dict[str, Any]) -> None: + db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") + kg: Optional[KnowledgeGraph] = None + try: + kg = KnowledgeGraph(db_path=db_path) + kg.add_triple( + subject="user", + predicate="asserted", + obj=payload.get("content", ""), + ) + except Exception as exc: + logger.debug("MemPalace _mirror_mem_write error: %s", exc) + finally: + if kg is not None: + try: + kg.close() + except Exception: + pass + + def _background_worker(self) -> None: + # Drain pending items even after the stop signal — otherwise turns + # queued just before shutdown would be lost. The compound condition + # exits only when both: (a) stop signal raised, (b) queue empty. + while not self._worker_stop.is_set() or not self._worker_queue.empty(): + try: + task, payload = self._worker_queue.get(timeout=1.0) + except queue.Empty: + if self._worker_stop.is_set(): + break + continue + try: + if task == "file_turn": + self._file_turn(payload) + elif task == "session_end": + self._mine_session(payload) + elif task == "pre_compress": + # Pair adjacent (user, assistant) messages into turns and + # file each pair. Filing only role==user would silently + # drop assistant content the ``on_pre_compress`` hint + # promised the summarizer was searchable. + msgs = payload.get("messages", []) or [] + session_id = payload.get("session_id", "") or "" + i = 0 + while i < len(msgs): + msg = msgs[i] + if msg.get("role") != "user": + # Lone non-user message (orphan tool result, etc.) + # — file under user= empty so we don't lose it. + self._file_turn( + { + "user": "", + "assistant": _normalize_content(msg.get("content")), + "session_id": session_id, + } + ) + i += 1 + continue + user_content = _normalize_content(msg.get("content")) + assistant_content = "" + if i + 1 < len(msgs) and msgs[i + 1].get("role") == "assistant": + assistant_content = _normalize_content(msgs[i + 1].get("content")) + i += 2 + else: + i += 1 + self._file_turn( + { + "user": user_content, + "assistant": assistant_content, + "session_id": session_id, + } + ) + elif task == "mem_write": + self._mirror_mem_write(payload) + except Exception as exc: + logger.debug("MemPalace worker task %s error: %s", task, exc) + finally: + try: + self._worker_queue.task_done() + except ValueError: + pass + + # ----- Tool handlers (delegate to mempalace internals) ---------------- + + def _tool_search( + self, + query: str, + wing: Optional[str] = None, + room: Optional[str] = None, + n_results: int = 5, + ) -> Dict[str, Any]: + if not query: + return {"error": "Missing required parameter: query"} + n = max(1, min(int(n_results or 5), 50)) + data = search_memories( + query, + palace_path=self._palace_path, + wing=wing or "", + room=room or "", + n_results=n, + ) + if isinstance(data, dict) and "error" in data: + return data + hits = data.get("results", []) if isinstance(data, dict) else [] + return {"results": hits, "count": len(hits)} + + def _scan_metadatas( + self, col: Any, where: Optional[Dict[str, Any]] = None + ) -> tuple[List[Dict[str, Any]], bool]: + """Pull at most ``STATUS_SCAN_LIMIT`` metadata records. + + Returns ``(metas, truncated)``. ``truncated`` is True when the + underlying collection holds more rows than the cap so the caller + can surface that to the model. + """ + cap = self.STATUS_SCAN_LIMIT + kwargs: Dict[str, Any] = {"include": ["metadatas"], "limit": cap} + if where: + kwargs["where"] = where + try: + result = col.get(**kwargs) + except TypeError: + # Very old chroma versions might not accept ``limit`` on + # ``get``; fall back to the full scan path. + result = col.get(include=["metadatas"], **({"where": where} if where else {})) + metas = result.get("metadatas") or [] + truncated = len(metas) >= cap + return metas, truncated + + def _tool_status(self) -> Dict[str, Any]: + with self._collection_lock: + col = self._collection + if col is None: + return {"error": "MemPalace not initialized"} + total = col.count() + metas, truncated = self._scan_metadatas(col) + wings: Dict[str, int] = {} + for m in metas: + w = m.get("wing", "unknown") + wings[w] = wings.get(w, 0) + 1 + out: Dict[str, Any] = { + "total_drawers": total, + "wings": wings, + "palace_path": self._palace_path, + } + if truncated: + # ``total_drawers`` already gives the model the 100% reference; + # ``scanned`` lets it compute coverage = scanned / total_drawers + # and qualify any wing claim accordingly. + out["truncated"] = True + out["scanned"] = len(metas) + return out + + def _tool_list_wings(self) -> Dict[str, Any]: + with self._collection_lock: + col = self._collection + if col is None: + return {"error": "MemPalace not initialized"} + metas, truncated = self._scan_metadatas(col) + wings: Dict[str, int] = {} + for m in metas: + w = m.get("wing", "unknown") + wings[w] = wings.get(w, 0) + 1 + out: Dict[str, Any] = {"wings": wings} + if truncated: + out["truncated"] = True + out["scanned"] = len(metas) + # Palace total is the model's 100% reference — same shape as + # ``_tool_status`` so a coverage ratio can be computed without + # an additional tool call. + out["total_drawers"] = col.count() + return out + + def _tool_list_rooms(self, wing: str) -> Dict[str, Any]: + if not wing: + return {"error": "Missing required parameter: wing"} + with self._collection_lock: + col = self._collection + if col is None: + return {"error": "MemPalace not initialized"} + metas, truncated = self._scan_metadatas(col, where={"wing": wing}) + rooms: Dict[str, int] = {} + for m in metas: + r = m.get("room", "unknown") + rooms[r] = rooms.get(r, 0) + 1 + out: Dict[str, Any] = {"wing": wing, "rooms": rooms} + if truncated: + out["truncated"] = True + out["scanned"] = len(metas) + # ChromaDB's ``count()`` doesn't support ``where=`` filtering + # in the versions mempalace pins, so we can't cheaply give an + # exact wing total. The model can still see this view is partial + # via the truncated/scanned pair. + return out + + def _tool_kg_query(self, entity: str, since: Optional[str] = None) -> Dict[str, Any]: + if not entity: + return {"error": "Missing required parameter: entity"} + db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") + kg = KnowledgeGraph(db_path=db_path) + try: + relations = kg.query_entity(entity, as_of=since or "") + finally: + try: + kg.close() + except Exception: + pass + return {"entity": entity, "relations": relations} + + def _tool_kg_add(self, subject: str, predicate: str, obj: str) -> Dict[str, Any]: + if not (subject and predicate and obj): + return {"error": "subject, predicate, object are all required"} + db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") + kg = KnowledgeGraph(db_path=db_path) + try: + kg.add_triple(subject=subject, predicate=predicate, obj=obj) + finally: + try: + kg.close() + except Exception: + pass + return {"status": "ok", "triple": [subject, predicate, obj]} + + def _tool_diary_write(self, entry: str) -> Dict[str, Any]: + if not entry: + return {"error": "Missing required parameter: entry"} + diary_path = Path(self._palace_path).parent / "diary.jsonl" + diary_path.parent.mkdir(parents=True, exist_ok=True) + record = {"ts": datetime.now(timezone.utc).isoformat(), "entry": entry} + with open(diary_path, "a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") + f.flush() + os.fsync(f.fileno()) + return {"status": "ok"} + + def _tool_diary_read(self, n: int = 10) -> Dict[str, Any]: + if n <= 0: + return {"entries": []} + diary_path = Path(self._palace_path).parent / "diary.jsonl" + if not diary_path.exists(): + return {"entries": []} + # Stream the file and keep only the trailing ``n`` lines. Avoids + # loading multi-megabyte diaries into memory just to discard the + # head. + with open(diary_path, encoding="utf-8") as f: + tail = deque(f, maxlen=n) + recent: List[Dict[str, Any]] = [] + for raw_line in tail: + try: + recent.append(json.loads(raw_line)) + except json.JSONDecodeError: + logger.debug("MemPalace: skipping malformed diary line") + return {"entries": recent} + + +# --------------------------------------------------------------------------- +# Plugin entry point +# --------------------------------------------------------------------------- + + +def register(ctx: Any) -> None: + """Register the MemPalace memory provider with Hermes.""" + ctx.register_memory_provider(MempalaceProvider()) diff --git a/tests/test_hermes_integration.py b/tests/test_hermes_integration.py new file mode 100644 index 0000000..ebe2fcb --- /dev/null +++ b/tests/test_hermes_integration.py @@ -0,0 +1,636 @@ +"""Tests for the MemPalace ↔ Hermes integration provider. + +The provider lives outside the importable ``mempalace`` package (it sits in +``mempalace/integrations/hermes/`` so it can be copied into +``~/.hermes/plugins/`` at install time). These tests load it the same way: +by file path. + +They also stub ``agent.memory_provider`` to mirror the runtime contract — the +plugin is only ever imported with Hermes on the import path. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import threading +import types +from pathlib import Path + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures: stub the Hermes ABC, then import the provider module by path. +# --------------------------------------------------------------------------- + + +def _install_stub_memory_provider() -> None: + """Install a minimal ``agent.memory_provider`` stub into sys.modules.""" + if "agent.memory_provider" in sys.modules: + return + agent_mod = types.ModuleType("agent") + mp_mod = types.ModuleType("agent.memory_provider") + + class MemoryProvider: # mirrors the parts the integration class uses + pass + + mp_mod.MemoryProvider = MemoryProvider # type: ignore[attr-defined] + agent_mod.memory_provider = mp_mod # type: ignore[attr-defined] + sys.modules["agent"] = agent_mod + sys.modules["agent.memory_provider"] = mp_mod + + +@pytest.fixture(scope="module") +def integration_module(): + _install_stub_memory_provider() + path = ( + Path(__file__).resolve().parent.parent + / "mempalace" + / "integrations" + / "hermes" + / "__init__.py" + ) + spec = importlib.util.spec_from_file_location("hermes_integration", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def provider(integration_module): + return integration_module.MempalaceProvider() + + +# --------------------------------------------------------------------------- +# Shape: name, schemas, config, availability +# --------------------------------------------------------------------------- + + +def test_name_matches_plugin_yaml(provider): + assert provider.name == "mempalace" + + +def test_is_available_imports_mempalace(provider): + # The repo's own dev install satisfies this. Failure means a broken venv. + assert provider.is_available() is True + + +def test_tool_schemas_visible_before_initialize(provider): + # Regression for the discovery bug: Hermes' + # ``agent.memory_manager._register_provider`` snapshots + # ``get_tool_schemas()`` once at registration time to build its + # ``tool_name → provider`` routing table. If we returned ``[]`` there, + # the dispatcher would never learn our tool names and every later call + # would hit ``"Unknown tool: "`` from the dispatcher without + # reaching ``handle_tool_call`` at all. Backend readiness gating + # belongs in ``handle_tool_call``, not here. + schemas = provider.get_tool_schemas() + assert len(schemas) == 27 # openclaw set + 8 tools added after #491 + names = {s["name"] for s in schemas} + assert "mempalace_status" in names + assert "mempalace_search" in names + assert "mempalace_add_drawer" in names + assert "mempalace_update_drawer" in names # added after openclaw #491 + assert "mempalace_kg_invalidate" in names + + +def test_config_schema_has_documented_keys(provider): + keys = {field["key"] for field in provider.get_config_schema()} + assert keys == { + "palace_path", + "identity_path", + "wing", + "n_prefetch", + } + # ``collection_name`` is intentionally absent — exposing it would let the + # provider write to a collection that ``search_memories`` doesn't read. + assert "collection_name" not in keys + + +def test_tool_schemas_module_constant_matches_expected_surface(integration_module): + # 27 tools — openclaw's reference skill set (19 tools at + # MemPalace/mempalace#491, April 2026) plus the 8 agent-facing tools + # mempalace has added since that openclaw hasn't caught up to. + # Admin/internal tools (sync, hook_settings, reconnect) intentionally + # omitted. + schemas = integration_module.TOOL_SCHEMAS + names = {s["name"] for s in schemas} + assert names == { + # Search + structure + "mempalace_search", + "mempalace_status", + "mempalace_list_wings", + "mempalace_list_rooms", + "mempalace_get_taxonomy", + "mempalace_get_aaak_spec", + # Drawer CRUD + "mempalace_add_drawer", + "mempalace_update_drawer", + "mempalace_delete_drawer", + "mempalace_list_drawers", + "mempalace_get_drawer", + "mempalace_check_duplicate", + # Knowledge graph + "mempalace_kg_query", + "mempalace_kg_add", + "mempalace_kg_invalidate", + "mempalace_kg_timeline", + "mempalace_kg_stats", + # Per-agent diary + "mempalace_diary_write", + "mempalace_diary_read", + # Room-graph navigation + tunnel management + "mempalace_traverse", + "mempalace_graph_stats", + "mempalace_find_tunnels", + "mempalace_create_tunnel", + "mempalace_list_tunnels", + "mempalace_delete_tunnel", + "mempalace_follow_tunnels", + # Session-level + "mempalace_memories_filed_away", + } + + +# --------------------------------------------------------------------------- +# Cron-context guard: provider must short-circuit on system-generated turns. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "kwargs", + [ + {"agent_context": "cron"}, + {"agent_context": "flush"}, + {"platform": "cron"}, + ], +) +def test_initialize_skips_under_cron_context(provider, kwargs, tmp_path): + # Even with a writable hermes_home, cron/flush context must not start the + # worker or open a collection. + provider.initialize("session-1", hermes_home=str(tmp_path), **kwargs) + assert provider._cron_skipped is True + assert provider._initialized is False + assert provider.get_tool_schemas() == [] + assert provider.system_prompt_block() == "" + assert provider.prefetch("anything") == "" + # Calls that would otherwise enqueue work must be no-ops. + provider.sync_turn("hi", "hello") + provider.on_session_end([]) + assert provider._worker_thread is None + + +def test_handle_tool_call_under_cron_returns_error_json(provider, tmp_path): + provider.initialize("session-1", hermes_home=str(tmp_path), agent_context="cron") + result = json.loads(provider.handle_tool_call("mempalace_search", {"query": "x"})) + assert "error" in result + + +def test_handle_tool_call_without_initialize_returns_error_json(provider): + result = json.loads(provider.handle_tool_call("mempalace_status", {})) + assert "error" in result + + +def test_on_session_end_no_op_when_not_initialized(provider): + # Without ``_initialized``, the worker thread isn't running. Enqueueing + # here would silently fill the bounded queue with tasks that never drain. + provider._initialized = False + provider._cron_skipped = False + pre = provider._worker_queue.qsize() + provider.on_session_end([{"role": "user", "content": "hi"}]) + assert provider._worker_queue.qsize() == pre + + +def test_on_memory_write_no_op_when_not_initialized(provider): + provider._initialized = False + provider._cron_skipped = False + pre = provider._worker_queue.qsize() + provider.on_memory_write("add", "user", "some fact") + assert provider._worker_queue.qsize() == pre + + +def test_normalize_content_flattens_anthropic_list(integration_module): + fn = integration_module._normalize_content + blocks = [ + {"type": "text", "text": "what's the auth flow?"}, + {"type": "tool_use", "name": "grep", "input": {"q": "JWT"}}, + {"type": "text", "text": "(short clarifier)"}, + ] + out = fn(blocks) + assert "what's the auth flow?" in out + assert "[tool_use: grep]" in out + assert "(short clarifier)" in out + # Must not be the literal Python repr. + assert "{'type'" not in out + + +def test_match_wing_by_keywords_word_boundary(integration_module): + fn = integration_module._match_wing_by_keywords + wing_config = { + "wing_ai": {"keywords": ["ai"]}, + "wing_dev": {"keywords": ["python"]}, + } + # Substring matching would have routed "said" / "rain" / "available" to wing_ai. + assert fn("She said rain is available", wing_config) == "wing_general" + assert fn("write some ai bindings", wing_config) == "wing_ai" + assert fn("python script for scraping", wing_config) == "wing_dev" + + +# --------------------------------------------------------------------------- +# Session switch / turn counter bookkeeping. +# --------------------------------------------------------------------------- + + +def test_on_session_switch_repoints_session_id(provider): + provider._session_id = "old" + provider._turn_count = 7 + provider.on_session_switch("new", reset=False) + assert provider._session_id == "new" + assert provider._turn_count == 7 # /resume / /branch keep counters + + +def test_on_session_switch_with_reset_clears_turn_counter(provider): + provider._turn_count = 9 + provider.on_session_switch("new", reset=True) + assert provider._turn_count == 0 + + +def test_on_turn_start_tracks_turn_number(provider): + provider.on_turn_start(turn_number=4, message="hi") + assert provider._turn_count == 4 + + +# --------------------------------------------------------------------------- +# on_pre_compress: contract is (a) file the discarded messages, (b) return a +# string to inject into the compression summary prompt. +# --------------------------------------------------------------------------- + + +def test_on_pre_compress_returns_string_hint_only_when_ready(provider): + # Before initialize: no hint (provider can't actually persist anything). + provider._cron_skipped = False + assert provider.on_pre_compress([{"role": "user", "content": "hi"}]) == "" + + # Simulate post-initialize ready state. We don't need a real backend for + # this assertion — just the readiness flag the hint gates on. + provider._initialized = True + hint = provider.on_pre_compress([{"role": "user", "content": "hi"}]) + assert isinstance(hint, str) and "mempalace_search" in hint + + +def test_on_pre_compress_under_cron_returns_empty_string(provider, tmp_path): + provider.initialize("session-1", hermes_home=str(tmp_path), agent_context="cron") + assert provider.on_pre_compress([{"role": "user", "content": "hi"}]) == "" + + +# --------------------------------------------------------------------------- +# Wing classification: keyword-based, fall back to wing_general. +# --------------------------------------------------------------------------- + + +def test_classify_wing_falls_back_to_general_with_no_config(provider): + provider._wing_config = {} + assert provider._classify_wing("anything") == "wing_general" + + +def test_classify_wing_matches_keyword(provider): + provider._wing_config = { + "wing_dev": {"keywords": ["python", "pytest"]}, + "wing_ops": {"keywords": ["deploy", "kubernetes"]}, + } + assert provider._classify_wing("Running pytest -q") == "wing_dev" + assert provider._classify_wing("kubectl deploy rollout") == "wing_ops" + assert provider._classify_wing("just chatting") == "wing_general" + + +# --------------------------------------------------------------------------- +# Shutdown is safe even when initialize() never ran. +# --------------------------------------------------------------------------- + + +def test_shutdown_is_safe_without_initialize(provider): + provider.shutdown() # must not raise + + +def test_shutdown_drains_running_worker(provider): + # Spin a fake worker that respects _worker_stop. + def _loop(): + while not provider._worker_stop.is_set(): + provider._worker_stop.wait(0.05) + + provider._worker_thread = threading.Thread(target=_loop, daemon=True) + provider._worker_thread.start() + provider.shutdown() + assert not provider._worker_thread.is_alive() + + +# --------------------------------------------------------------------------- +# End-to-end integration: real palace via mempalace's own fixtures. +# +# These exercise the ChromaBackend code path that fixes the dim-mismatch bug +# from prior in-tree Hermes PRs, and run the tool handlers against the +# `seeded_collection` / `seeded_kg` fixtures from `tests/conftest.py`. +# --------------------------------------------------------------------------- + + +@pytest.fixture +def initialized_provider(provider, palace_path, tmp_dir): + """Provider initialized against a fresh temp palace.""" + config_path = Path(tmp_dir) / "mempalace.json" + config_path.write_text(json.dumps({"palace_path": palace_path})) + provider.initialize("test-session-1", hermes_home=str(tmp_dir), platform="cli") + yield provider + provider.shutdown() + + +def test_initialize_opens_chroma_via_backend(initialized_provider): + """The dim-mismatch fix: collection access goes through ChromaBackend.""" + from mempalace.backends.chroma import ChromaBackend + + assert initialized_provider._initialized is True + assert initialized_provider._collection is not None + assert isinstance(initialized_provider._backend, ChromaBackend) + + +def test_get_tool_schemas_returns_full_surface_after_initialize(initialized_provider): + schemas = initialized_provider.get_tool_schemas() + names = {s["name"] for s in schemas} + assert len(schemas) == 27 + assert "mempalace_search" in names + assert "mempalace_kg_query" in names + assert "mempalace_add_drawer" in names + assert "mempalace_update_drawer" in names + assert "mempalace_memories_filed_away" in names + + +def test_sync_turn_persists_through_worker(initialized_provider): + initialized_provider.sync_turn("what's the plan?", "ship the PR") + initialized_provider._worker_queue.join() # block until worker drains the task + + col = initialized_provider._collection + assert col.count() >= 1 + metas = col.get(include=["metadatas"]).get("metadatas") or [] + assert any(m.get("source") == "hermes" for m in metas) + + +def test_sync_turn_writes_canonical_drawer_metadata(initialized_provider): + """Live turns must carry the same metadata the convo miner writes. + + Without hall / entities / filed_at, Hermes drawers are silently + invisible to hallway traversal, entity search, and the since/before + date filters — nothing errors, recall just degrades. + """ + initialized_provider.sync_turn("meeting with Sarah about the Q3 roadmap", "noted") + initialized_provider._worker_queue.join() + + metas = initialized_provider._collection.get(include=["metadatas"]).get("metadatas") or [] + hermes_metas = [m for m in metas if m.get("source") == "hermes"] + assert hermes_metas + meta = hermes_metas[0] + for key in ( + "wing", + "room", + "hall", + "source_file", + "added_by", + "filed_at", + "authored_at", + "ingest_mode", + "extract_mode", + "normalize_version", + "id_recipe", + ): + assert key in meta, f"missing canonical metadata key: {key!r}" + assert meta["room"] == "conversations" + assert meta["ingest_mode"] == "convos" + assert meta["extract_mode"] == "exchange" + + +def test_sync_turn_routes_to_configured_wing(initialized_provider): + initialized_provider._wing_config = {"wing_dev": {"keywords": ["pytest"]}} + initialized_provider.sync_turn("running pytest -q", "all passed") + initialized_provider._worker_queue.join() + + metas = initialized_provider._collection.get(include=["metadatas"]).get("metadatas") or [] + assert any(m.get("wing") == "wing_dev" for m in metas) + + +def test_sync_turn_skips_when_both_sides_empty(initialized_provider): + pre = initialized_provider._collection.count() + initialized_provider.sync_turn("", "") + # Queue should not have received an item; nothing to join, but worker has + # nothing to do either. Give it a moment then re-check. + initialized_provider._worker_queue.join() + assert initialized_provider._collection.count() == pre + + +# ----- Tool handlers against seeded data ---------------------------------- + + +@pytest.fixture +def provider_on_seeded_palace(seeded_collection, provider, palace_path, tmp_dir): + """Provider pointed at the same palace_path that ``seeded_collection`` filled. + + ``seeded_collection`` writes 4 drawers via raw ``chromadb.PersistentClient``; + we then have the provider open the same path via ``ChromaBackend`` — the + fact that this round-trips at all is the dim-mismatch regression check. + """ + (Path(tmp_dir) / "mempalace.json").write_text(json.dumps({"palace_path": palace_path})) + provider.initialize("s1", hermes_home=str(tmp_dir)) + yield provider + provider.shutdown() + + +def test_status_tool_counts_seeded_drawers(provider_on_seeded_palace): + result = json.loads(provider_on_seeded_palace.handle_tool_call("mempalace_status", {})) + assert result["total_drawers"] == 4 + assert result["wings"]["project"] == 3 + assert result["wings"]["notes"] == 1 + + +def test_list_wings_tool_returns_seeded_wings(provider_on_seeded_palace): + result = json.loads(provider_on_seeded_palace.handle_tool_call("mempalace_list_wings", {})) + assert result["wings"] == {"project": 3, "notes": 1} + + +def test_list_rooms_tool_filters_by_wing(provider_on_seeded_palace): + result = json.loads( + provider_on_seeded_palace.handle_tool_call( + "mempalace_list_rooms", + {"wing": "project"}, + ) + ) + assert result["wing"] == "project" + assert result["rooms"]["backend"] == 2 + assert result["rooms"]["frontend"] == 1 + + +def test_list_rooms_tool_rejects_missing_wing(provider_on_seeded_palace): + result = json.loads( + provider_on_seeded_palace.handle_tool_call( + "mempalace_list_rooms", + {}, + ) + ) + assert "error" in result + + +def test_status_tool_omits_truncated_under_cap(provider_on_seeded_palace): + # 4 seeded drawers, cap is 5000 — the response must not advertise + # itself as a partial view when in fact it's complete. + result = json.loads(provider_on_seeded_palace.handle_tool_call("mempalace_status", {})) + assert "truncated" not in result + assert "scanned" not in result + + +def test_status_tool_marks_truncated_with_structured_fields(provider_on_seeded_palace): + # Force the cap below the seeded count so we exercise the truncation path. + # The model needs ``truncated`` (bool) + ``scanned`` (int) so it can + # compute coverage = scanned / total_drawers itself rather than parsing + # a sentence. + provider_on_seeded_palace.STATUS_SCAN_LIMIT = 2 + result = json.loads(provider_on_seeded_palace.handle_tool_call("mempalace_status", {})) + assert result["truncated"] is True + assert result["scanned"] == 2 + assert result["total_drawers"] == 4 + + +def test_list_wings_tool_marks_truncated_with_palace_total(provider_on_seeded_palace): + # ``_tool_list_wings`` has no unconditional ``total_drawers`` field — + # when truncated it must surface ``total_drawers`` so callers can + # compute coverage without a second ``mempalace_status`` call. + provider_on_seeded_palace.STATUS_SCAN_LIMIT = 2 + result = json.loads(provider_on_seeded_palace.handle_tool_call("mempalace_list_wings", {})) + assert result["truncated"] is True + assert result["scanned"] == 2 + assert result["total_drawers"] == 4 + + +def test_list_rooms_tool_marks_truncated_without_wing_total(provider_on_seeded_palace): + # Rooms can't cheaply give an exact wing total (no ``where=`` on + # ``count()`` in the pinned chroma version). The structured fields are + # still present; the absent ``total_drawers`` is intentional and + # documented in the code. + provider_on_seeded_palace.STATUS_SCAN_LIMIT = 1 + result = json.loads( + provider_on_seeded_palace.handle_tool_call( + "mempalace_list_rooms", + {"wing": "project"}, + ) + ) + assert result["truncated"] is True + assert result["scanned"] == 1 + assert "total_drawers" not in result + + +# ----- Knowledge-graph tool handlers -------------------------------------- + + +def test_kg_add_persists_to_palace_sibling_sqlite(initialized_provider, palace_path): + """The provider writes to ``/../knowledge_graph.sqlite3``.""" + from mempalace.knowledge_graph import KnowledgeGraph + + result = json.loads( + initialized_provider.handle_tool_call( + "mempalace_kg_add", + {"subject": "user", "predicate": "likes", "object": "coffee"}, + ) + ) + assert result["status"] == "ok" + + db_path = str(Path(palace_path).parent / "knowledge_graph.sqlite3") + independent_kg = KnowledgeGraph(db_path=db_path) + try: + relations = independent_kg.query_entity("user") + finally: + independent_kg.close() + assert any( + (r.get("predicate") == "likes" and r.get("object") == "coffee") for r in (relations or []) + ) + + +def test_kg_query_tool_rejects_missing_entity(initialized_provider): + result = json.loads( + initialized_provider.handle_tool_call( + "mempalace_kg_query", + {}, + ) + ) + assert "error" in result + + +# ----- Diary roundtrip ---------------------------------------------------- + + +def test_diary_write_read_roundtrip(initialized_provider): + write = json.loads( + initialized_provider.handle_tool_call( + "mempalace_diary_write", + {"entry": "Today I deepened test coverage."}, + ) + ) + assert write["status"] == "ok" + + read = json.loads( + initialized_provider.handle_tool_call( + "mempalace_diary_read", + {"n": 5}, + ) + ) + assert read["entries"] + assert read["entries"][-1]["entry"] == "Today I deepened test coverage." + + +def test_diary_read_empty_when_no_writes(initialized_provider): + result = json.loads(initialized_provider.handle_tool_call("mempalace_diary_read", {})) + assert result["entries"] == [] + + +# ----- Config-load path --------------------------------------------------- + + +def test_initialize_reads_mempalace_json(provider, tmp_dir, palace_path): + (Path(tmp_dir) / "mempalace.json").write_text( + json.dumps( + { + "palace_path": palace_path, + "n_prefetch": 7, + "wing": "wing_from_config", + } + ) + ) + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + assert provider._config["n_prefetch"] == 7 + assert provider._config["wing"] == "wing_from_config" + finally: + provider.shutdown() + + +def test_collection_name_is_not_user_configurable(provider, tmp_dir, palace_path): + # Exposing ``collection_name`` would let the provider write to a collection + # that ``search_memories`` (which reads from mempalace's own config) does + # not read — making the provider silently appear mute. The field is + # intentionally absent from the schema and ignored in config files. + (Path(tmp_dir) / "mempalace.json").write_text( + json.dumps({"palace_path": palace_path, "collection_name": "custom_drawers"}) + ) + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + assert provider._collection_name == provider.DEFAULT_COLLECTION_NAME + finally: + provider.shutdown() + + +def test_env_vars_override_config_file(provider, tmp_dir, palace_path, monkeypatch): + monkeypatch.setenv("MEMPALACE_PALACE_PATH", palace_path) + monkeypatch.setenv("MEMPALACE_WING", "wing_forced") + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + assert provider._palace_path == palace_path + assert provider._config["wing"] == "wing_forced" + finally: + provider.shutdown() From 37bc50d98aa7702d84ba17575455fbad4f22ae48 Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:02:38 -0400 Subject: [PATCH 3/7] fix(hermes): address PR #1915 review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _scan_metadatas: fetch cap+1 and compare len > cap, so a collection holding exactly STATUS_SCAN_LIMIT rows is no longer reported as truncated (the view is complete). Callers still get at most cap rows. - status/list_wings/list_rooms: tolerate None metadata entries from legacy palaces / raw writers instead of failing the tool call. - _match_wing_by_keywords: skip non-string keywords so a hand-edited wing_config.json can't break live turn filing. - file_conversation_exchange: extra_metadata can no longer overwrite canonical keys (matches the documented append-only contract), and wing/room are validated with sanitize_name — invalid names fall back to wing_general / conversations rather than dropping the turn, per the verbatim-first mandate. - Fix two stale docstrings left from the pre-split layout. Co-Authored-By: Claude Fable 5 --- mempalace/convo_miner.py | 33 ++++++++-- mempalace/integrations/hermes/__init__.py | 28 +++++---- tests/test_convo_miner.py | 73 ++++++++++++++++++++++ tests/test_hermes_integration.py | 76 +++++++++++++++++++++-- 4 files changed, 191 insertions(+), 19 deletions(-) diff --git a/mempalace/convo_miner.py b/mempalace/convo_miner.py index abe66b4..d2ab98f 100644 --- a/mempalace/convo_miner.py +++ b/mempalace/convo_miner.py @@ -82,14 +82,38 @@ def file_conversation_exchange( traversal, entity search, and since/before date filters see integration drawers exactly like mined ones. + ``wing`` and ``room`` are validated with the same ``sanitize_name`` + rules the MCP write tools apply, but a failed name falls back + (``wing_general`` / ``conversations``) instead of erroring: this + path files *live* turns, and dropping a turn over a config typo + would break the verbatim / 100%-recall promise. The fallback is + logged at warning level so the misconfiguration is visible. + ``extra_metadata`` lets callers append integration-specific fields - (e.g. ``source`` / ``session_id``) but cannot be used to *drop* the - canonical keys. Returns the drawer id, or None when ``text`` is - empty after stripping. + (e.g. ``source`` / ``session_id``); keys that collide with the + canonical fields are ignored, so it cannot be used to overwrite or + drop them. Returns the drawer id, or None when ``text`` is empty + after stripping. """ + from .config import sanitize_name + text = (text or "").strip() if not text: return None + try: + wing = sanitize_name(wing, "wing") + except ValueError: + logger.warning( + "file_conversation_exchange: invalid wing %r — filing under wing_general", wing + ) + wing = "wing_general" + try: + room = sanitize_name(room, "room") + except ValueError: + logger.warning( + "file_conversation_exchange: invalid room %r — filing under conversations", room + ) + room = "conversations" filed_at = datetime.now().isoformat() drawer_id = make_exchange_drawer_id(wing, room, source_file, filed_at, text) metadata = { @@ -108,7 +132,8 @@ def file_conversation_exchange( "id_recipe": ID_RECIPE, } if extra_metadata: - metadata.update(extra_metadata) + for key, value in extra_metadata.items(): + metadata.setdefault(key, value) collection.upsert(ids=[drawer_id], documents=[text], metadatas=[metadata]) return drawer_id diff --git a/mempalace/integrations/hermes/__init__.py b/mempalace/integrations/hermes/__init__.py index 833b70f..541bf8c 100644 --- a/mempalace/integrations/hermes/__init__.py +++ b/mempalace/integrations/hermes/__init__.py @@ -78,10 +78,8 @@ def _match_wing_by_keywords(text: str, wing_config: Dict[str, Any]) -> str: Word boundaries matter — bare substring matching routes turns mentioning ``said`` into a wing whose keyword is ``ai``. Fall back to ``wing_general``. - Lives at module scope so ``backfill.py`` can use exactly the same matching - logic the live provider uses (it cannot import from this file as a - relative import — it's run by ``importlib.util.spec_from_file_location``). - The duplicate below in ``backfill.py`` must be kept in sync. + Lives at module scope so ``backfill.py`` can import and delegate to it — + one routing implementation shared by live and historical ingest. """ if not wing_config: return "wing_general" @@ -89,7 +87,10 @@ def _match_wing_by_keywords(text: str, wing_config: Dict[str, Any]) -> str: for wing_name, wing_def in wing_config.items(): keywords = wing_def.get("keywords", []) if isinstance(wing_def, dict) else [] for kw in keywords: - if not kw: + # The isinstance guard keeps a hand-edited wing_config.json + # (numbers / nulls in a keyword list) from raising inside + # _file_turn's try/except and silently dropping every turn. + if not kw or not isinstance(kw, str): continue pattern = r"\b" + re.escape(kw.lower()) + r"\b" if re.search(pattern, text_lower): @@ -1244,7 +1245,11 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] can surface that to the model. """ cap = self.STATUS_SCAN_LIMIT - kwargs: Dict[str, Any] = {"include": ["metadatas"], "limit": cap} + # Fetch one row beyond the cap: it's the only way to tell "exactly + # cap rows, view is complete" from "more than cap rows, view is + # partial" — comparing against ``col.count()`` can't answer that + # for the ``where``-filtered calls (chroma's count() is unfiltered). + kwargs: Dict[str, Any] = {"include": ["metadatas"], "limit": cap + 1} if where: kwargs["where"] = where try: @@ -1254,8 +1259,8 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] # ``get``; fall back to the full scan path. result = col.get(include=["metadatas"], **({"where": where} if where else {})) metas = result.get("metadatas") or [] - truncated = len(metas) >= cap - return metas, truncated + truncated = len(metas) > cap + return metas[:cap], truncated def _tool_status(self) -> Dict[str, Any]: with self._collection_lock: @@ -1266,7 +1271,8 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] metas, truncated = self._scan_metadatas(col) wings: Dict[str, int] = {} for m in metas: - w = m.get("wing", "unknown") + # Legacy palaces / raw writers can leave None metadata entries. + w = (m or {}).get("wing", "unknown") wings[w] = wings.get(w, 0) + 1 out: Dict[str, Any] = { "total_drawers": total, @@ -1289,7 +1295,7 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] metas, truncated = self._scan_metadatas(col) wings: Dict[str, int] = {} for m in metas: - w = m.get("wing", "unknown") + w = (m or {}).get("wing", "unknown") wings[w] = wings.get(w, 0) + 1 out: Dict[str, Any] = {"wings": wings} if truncated: @@ -1311,7 +1317,7 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] metas, truncated = self._scan_metadatas(col, where={"wing": wing}) rooms: Dict[str, int] = {} for m in metas: - r = m.get("room", "unknown") + r = (m or {}).get("room", "unknown") rooms[r] = rooms.get(r, 0) + 1 out: Dict[str, Any] = {"wing": wing, "rooms": rooms} if truncated: diff --git a/tests/test_convo_miner.py b/tests/test_convo_miner.py index df35be6..bf6f220 100644 --- a/tests/test_convo_miner.py +++ b/tests/test_convo_miner.py @@ -696,3 +696,76 @@ def test_register_file_sentinel_includes_source_mtime(): assert abs(mined[str(tiny_file)] - os.path.getmtime(tiny_file)) < 0.001 finally: shutil.rmtree(tmpdir, ignore_errors=True) + + +# --------------------------------------------------------------------------- +# file_conversation_exchange — canonical single-exchange write path +# --------------------------------------------------------------------------- + + +class _RecordingCollection: + """Captures upsert kwargs without a real ChromaDB behind it.""" + + def __init__(self): + self.upserts = [] + + def upsert(self, *, ids, documents, metadatas): + self.upserts.append({"ids": ids, "documents": documents, "metadatas": metadatas}) + + +def _exchange_kwargs(**overrides): + kwargs = { + "wing": "wing_dev", + "room": "conversations", + "text": "User: hi\n\nAssistant: hello", + "source_file": "hermes-session:s1", + "agent": "hermes", + } + kwargs.update(overrides) + return kwargs + + +def test_file_conversation_exchange_extra_metadata_cannot_clobber_canonical(): + """The docstring promises extras are append-only — colliding keys lose. + + PR #1915 review: ``metadata.update(extra_metadata)`` let a caller + silently overwrite ``wing`` / ``filed_at`` / etc. + """ + from mempalace.convo_miner import file_conversation_exchange + + col = _RecordingCollection() + file_conversation_exchange( + col, + **_exchange_kwargs(), + extra_metadata={"wing": "wing_evil", "filed_at": "1970-01-01", "source": "hermes"}, + ) + meta = col.upserts[0]["metadatas"][0] + assert meta["wing"] == "wing_dev" + assert meta["filed_at"] != "1970-01-01" + # Non-colliding extras still land. + assert meta["source"] == "hermes" + + +def test_file_conversation_exchange_invalid_wing_falls_back_to_wing_general(): + """A bad configured wing must not drop the turn — verbatim first. + + Same validation the MCP write tools apply (sanitize_name), but with a + wing_general fallback instead of an error: live filing losing turns + over a config typo would violate the 100%-recall promise. + """ + from mempalace.convo_miner import file_conversation_exchange + + col = _RecordingCollection() + file_conversation_exchange(col, **_exchange_kwargs(wing="../escape")) + meta = col.upserts[0]["metadatas"][0] + assert meta["wing"] == "wing_general" + assert col.upserts[0]["documents"] == ["User: hi\n\nAssistant: hello"] + + +def test_file_conversation_exchange_invalid_room_falls_back_to_conversations(): + from mempalace.convo_miner import file_conversation_exchange + + col = _RecordingCollection() + file_conversation_exchange(col, **_exchange_kwargs(room="a/b")) + meta = col.upserts[0]["metadatas"][0] + assert meta["room"] == "conversations" diff --git a/tests/test_hermes_integration.py b/tests/test_hermes_integration.py index ebe2fcb..2c67a30 100644 --- a/tests/test_hermes_integration.py +++ b/tests/test_hermes_integration.py @@ -1,9 +1,9 @@ """Tests for the MemPalace ↔ Hermes integration provider. -The provider lives outside the importable ``mempalace`` package (it sits in -``mempalace/integrations/hermes/`` so it can be copied into -``~/.hermes/plugins/`` at install time). These tests load it the same way: -by file path. +The provider ships inside the ``mempalace`` package (at +``mempalace/integrations/hermes/``) but at runtime Hermes loads it from a +copy in ``~/.hermes/plugins/`` via ``spec_from_file_location`` — not as a +package import. These tests load it the same way: by file path. They also stub ``agent.memory_provider`` to mirror the runtime contract — the plugin is only ever imported with Hermes on the import path. @@ -634,3 +634,71 @@ def test_env_vars_override_config_file(provider, tmp_dir, palace_path, monkeypat assert provider._config["wing"] == "wing_forced" finally: provider.shutdown() + + +# --------------------------------------------------------------------------- +# PR #1915 review fixes: scan truncation boundary, None metadata, bad keywords. +# --------------------------------------------------------------------------- + + +class _FakeScanCollection: + """Serves ``n`` rows through the same get(limit=...) shape chroma uses.""" + + def __init__(self, n, metadatas=None): + self._n = n + self._metadatas = metadatas + + def count(self): + return self._n + + def get(self, **kwargs): + if self._metadatas is not None: + return {"metadatas": list(self._metadatas)} + limit = kwargs.get("limit") or self._n + return {"metadatas": [{"wing": "wing_a", "room": "r"} for _ in range(min(self._n, limit))]} + + +def test_scan_metadatas_not_truncated_at_exactly_cap(provider): + cap = provider.STATUS_SCAN_LIMIT + metas, truncated = provider._scan_metadatas(_FakeScanCollection(cap)) + assert len(metas) == cap + # Exactly cap rows means the view is complete — flagging it truncated + # makes the model qualify a breakdown that is in fact 100% coverage. + assert truncated is False + + +def test_scan_metadatas_truncated_above_cap(provider): + cap = provider.STATUS_SCAN_LIMIT + metas, truncated = provider._scan_metadatas(_FakeScanCollection(cap + 1)) + assert truncated is True + # Callers still get at most cap rows — the +1 probe row is trimmed. + assert len(metas) == cap + + +def test_status_and_list_tools_tolerate_none_metadata_entries(provider): + # Legacy palaces / raw writers can leave None metadata entries; the + # breakdown loops must count them as "unknown", not fail the tool call. + rows = [None, {"wing": "wing_a", "room": "room_a"}] + provider._collection = _FakeScanCollection(2, metadatas=rows) + + status = provider._tool_status() + assert "error" not in status + assert status["wings"] == {"unknown": 1, "wing_a": 1} + + wings = provider._tool_list_wings() + assert "error" not in wings + assert wings["wings"] == {"unknown": 1, "wing_a": 1} + + rooms = provider._tool_list_rooms("wing_a") + assert "error" not in rooms + assert rooms["rooms"] == {"unknown": 1, "room_a": 1} + + +def test_match_wing_by_keywords_ignores_non_string_keywords(integration_module): + # A hand-edited wing_config.json with a number/null in a keyword list + # must not break wing routing — a raised AttributeError inside + # _file_turn's try/except silently drops every live turn. + fn = integration_module._match_wing_by_keywords + wing_config = {"wing_dev": {"keywords": [None, 3, "python"]}} + assert fn("write some python code", wing_config) == "wing_dev" + assert fn("unrelated chatter", wing_config) == "wing_general" From 2d6580b1f1d26f351f8d4c6e0860aeb62bb8530e Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Mon, 6 Jul 2026 09:24:46 -0400 Subject: [PATCH 4/7] fix(hermes): make sync_turn the sole filing path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit on_session_end and on_pre_compress blind-re-filed the raw message list, duplicating every turn sync_turn had already stored — filed_at is hashed into the drawer id, so upserts cannot collapse the copies. Drop the re-filing (and the pre-compress hint that over-promised persistence); on_session_end keeps only the wake-up cache refresh. Co-Authored-By: Claude Fable 5 --- mempalace/integrations/hermes/__init__.py | 121 +++------------------- tests/test_hermes_integration.py | 26 +++-- 2 files changed, 33 insertions(+), 114 deletions(-) diff --git a/mempalace/integrations/hermes/__init__.py b/mempalace/integrations/hermes/__init__.py index 541bf8c..5993c2e 100644 --- a/mempalace/integrations/hermes/__init__.py +++ b/mempalace/integrations/hermes/__init__.py @@ -17,6 +17,13 @@ Design notes * Per-turn writes go through a bounded background queue. The agent loop never blocks on ChromaDB or SQLite. +* ``sync_turn`` is the **sole** filing path. ``on_session_end`` and + ``on_pre_compress`` intentionally file nothing: re-filing the raw + message list duplicates every turn ``sync_turn`` already stored — + ``filed_at`` is hashed into the drawer id, so upserts cannot collapse + the copies. Any future safety net here must first scan what is + already filed and add only what is missing. + * The provider is **inactive** under ``agent_context in {"cron", "flush"}`` or ``platform == "cron"``. Cron-context turns are system-generated and would otherwise corrupt the user's representation. @@ -716,23 +723,12 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] self._turn_count = turn_number def on_session_end(self, messages: List[Dict[str, Any]]) -> None: - # Skip when the provider never came up — enqueueing to a queue whose - # worker never started would silently fill the bounded buffer with - # tasks that can never drain. if self._cron_skipped or not self._initialized: return - try: - self._worker_queue.put_nowait( - ( - "session_end", - {"messages": list(messages or []), "session_id": self._session_id}, - ) - ) - except queue.Full: - logger.warning( - "MemPalace queue full at session_end — %d messages will not be filed", - len(messages or []), - ) + # Intentionally no filing here. ``sync_turn`` has already filed every + # completed turn, and re-filing the message list mints duplicate + # drawers: ``filed_at`` is part of the drawer-id hash, so the upsert + # cannot collapse the re-file into the original. # Regenerate the AAAK wake-up cache for the next session. threading.Thread( target=self._refresh_wake_up_cache, @@ -756,33 +752,14 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] self._turn_count = 0 def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str: - """File the messages about to be discarded and signal verbatim persistence. + """Intentionally a no-op that returns no hint. - The hint is **only** returned when the worker can actually persist the - payload — if the backend never came up or the queue is saturated, we - say nothing so the summarizer falls back to its default conservative - discarding rather than acting on a false promise. + Blind-filing the compression window duplicates every turn + ``sync_turn`` already filed. Returning ``""`` keeps the + summarizer on its default conservative discarding — a hint must + never promise persistence this provider hasn't performed. """ - if self._cron_skipped or not self._initialized: - return "" - try: - self._worker_queue.put_nowait( - ( - "pre_compress", - {"messages": list(messages or []), "session_id": self._session_id}, - ) - ) - except queue.Full: - logger.warning( - "MemPalace queue full at pre_compress — %d messages will not be filed", - len(messages or []), - ) - return "" - return ( - "MemPalace has filed every message in this window verbatim. " - "Compressed content remains searchable via the `mempalace_search` " - "tool — the summarizer can be aggressive about discarding raw turns." - ) + return "" def on_memory_write( self, @@ -1104,32 +1081,6 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] except Exception as exc: logger.debug("MemPalace _file_turn error: %s", exc) - def _mine_session(self, payload: Dict[str, Any]) -> None: - messages = payload.get("messages", []) or [] - session_id = payload.get("session_id", "") or "" - try: - for idx, msg in enumerate(messages): - if msg.get("role") != "user": - continue - # Same content normalization ``sync_turn`` and ``pre_compress`` - # use — list-shaped Anthropic content must not be persisted as - # its ``repr``. - content = _normalize_content(msg.get("content")) - if not content: - continue - assistant_content = "" - if idx + 1 < len(messages) and messages[idx + 1].get("role") == "assistant": - assistant_content = _normalize_content(messages[idx + 1].get("content")) - self._file_turn( - { - "user": content, - "assistant": assistant_content, - "session_id": session_id, - } - ) - except Exception as exc: - logger.debug("MemPalace _mine_session error: %s", exc) - def _mirror_mem_write(self, payload: Dict[str, Any]) -> None: db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") kg: Optional[KnowledgeGraph] = None @@ -1163,44 +1114,6 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] try: if task == "file_turn": self._file_turn(payload) - elif task == "session_end": - self._mine_session(payload) - elif task == "pre_compress": - # Pair adjacent (user, assistant) messages into turns and - # file each pair. Filing only role==user would silently - # drop assistant content the ``on_pre_compress`` hint - # promised the summarizer was searchable. - msgs = payload.get("messages", []) or [] - session_id = payload.get("session_id", "") or "" - i = 0 - while i < len(msgs): - msg = msgs[i] - if msg.get("role") != "user": - # Lone non-user message (orphan tool result, etc.) - # — file under user= empty so we don't lose it. - self._file_turn( - { - "user": "", - "assistant": _normalize_content(msg.get("content")), - "session_id": session_id, - } - ) - i += 1 - continue - user_content = _normalize_content(msg.get("content")) - assistant_content = "" - if i + 1 < len(msgs) and msgs[i + 1].get("role") == "assistant": - assistant_content = _normalize_content(msgs[i + 1].get("content")) - i += 2 - else: - i += 1 - self._file_turn( - { - "user": user_content, - "assistant": assistant_content, - "session_id": session_id, - } - ) elif task == "mem_write": self._mirror_mem_write(payload) except Exception as exc: diff --git a/tests/test_hermes_integration.py b/tests/test_hermes_integration.py index 2c67a30..be26416 100644 --- a/tests/test_hermes_integration.py +++ b/tests/test_hermes_integration.py @@ -264,21 +264,27 @@ def test_on_turn_start_tracks_turn_number(provider): # --------------------------------------------------------------------------- -# on_pre_compress: contract is (a) file the discarded messages, (b) return a -# string to inject into the compression summary prompt. +# on_session_end / on_pre_compress: sync_turn is the sole filing path — these +# hooks must neither enqueue filing work nor promise persistence. Re-filing +# the raw message list mints duplicate drawers (filed_at is hashed into the +# drawer id, so upserts cannot collapse the copies). # --------------------------------------------------------------------------- -def test_on_pre_compress_returns_string_hint_only_when_ready(provider): - # Before initialize: no hint (provider can't actually persist anything). +def test_on_pre_compress_returns_no_hint_and_files_nothing(provider): provider._cron_skipped = False - assert provider.on_pre_compress([{"role": "user", "content": "hi"}]) == "" - - # Simulate post-initialize ready state. We don't need a real backend for - # this assertion — just the readiness flag the hint gates on. provider._initialized = True - hint = provider.on_pre_compress([{"role": "user", "content": "hi"}]) - assert isinstance(hint, str) and "mempalace_search" in hint + pre = provider._worker_queue.qsize() + assert provider.on_pre_compress([{"role": "user", "content": "hi"}]) == "" + assert provider._worker_queue.qsize() == pre + + +def test_on_session_end_files_nothing_when_initialized(provider): + provider._cron_skipped = False + provider._initialized = True + pre = provider._worker_queue.qsize() + provider.on_session_end([{"role": "user", "content": "hi"}]) + assert provider._worker_queue.qsize() == pre def test_on_pre_compress_under_cron_returns_empty_string(provider, tmp_path): From bb1d502ddc4260e4ce20871c3d506fa48dd60934 Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Mon, 6 Jul 2026 10:02:58 -0400 Subject: [PATCH 5/7] fix(hermes): resolve palace, collection, and KG against one config chain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The provider filed and searched self._palace_path while the mcp_server passthrough tools resolved mempalace's global config — a custom Hermes palace_path searched one palace while drawer CRUD, duplicate checks, and tunnels wrote another. Publish the resolved palace to MEMPALACE_PALACE_PATH (mcp_server's own --palace mechanism) with ownership tracking so a stale bridge never outranks edited config and a user-set env var is never touched. The KG tools become native handlers: mcp_server resolves its KG from DEFAULT_KG_PATH unless its own CLI flag was given, which no env bridge can influence. collection_name and the unset-palace default now defer to MempalaceConfig — the same single chain the MCP server itself uses. Also document that hermes backup does not cover ~/.mempalace (no ABC hook exists for contributing external paths). Co-Authored-By: Claude Fable 5 --- mempalace/integrations/hermes/__init__.py | 217 ++++++++++++++++++---- tests/test_hermes_integration.py | 196 ++++++++++++++++++- 2 files changed, 375 insertions(+), 38 deletions(-) diff --git a/mempalace/integrations/hermes/__init__.py b/mempalace/integrations/hermes/__init__.py index 5993c2e..63bd09a 100644 --- a/mempalace/integrations/hermes/__init__.py +++ b/mempalace/integrations/hermes/__init__.py @@ -31,17 +31,31 @@ Design notes * Configuration precedence: ``$HERMES_HOME/mempalace.json`` is read first, then env vars override (``MEMPALACE_PALACE_PATH``, ``MEMPALACE_IDENTITY_PATH``, ``MEMPALACE_WING``). An empty env var - is ignored — ``export MEMPALACE_WING=`` is intent to unset. Defaults - fill in anything still missing. ``collection_name`` is intentionally - not user-configurable here: the provider writes through - ``self._collection_name`` while ``search_memories`` (used by - ``prefetch`` and ``_tool_search``) reads its own configured collection - name from ``~/.mempalace/config.json``, and exposing two ways to set - it would let the two diverge silently. + is ignored — ``export MEMPALACE_WING=`` is intent to unset. A palace + still unset after that defers to mempalace's own config + (``~/.mempalace/config.json``) before falling back to the default + location. The resolved palace is then published to + ``MEMPALACE_PALACE_PATH`` (see ``_bridge_palace_env``) so the + ``mempalace.mcp_server`` passthrough tools operate on the same palace + as live filing and search — never a config-file-vs-provider split. + ``collection_name`` follows mempalace's own config for the same + reason: it is what ``search_memories`` (used by ``prefetch`` and + ``_tool_search``) and the mcp_server passthrough read, so live writes + land in the collection recall actually searches. It is intentionally + not configurable on the Hermes side — a second knob would let the + write and read sides diverge silently again. * ``~/.mempalace/identity.txt`` (L0) and ``~/.mempalace/wing_config.json`` are loaded if present but never created here. Run ``mempalace init `` to generate them. + +* All palace state (ChromaDB, knowledge graph, diary, identity) lives + under ``~/.mempalace/`` by design — the palace is the user's central + memory shared across agents, not per-agent Hermes state. This means + ``hermes backup`` (which archives only ``$HERMES_HOME``) does NOT + cover it; users must back up ``~/.mempalace/`` separately. Hermes' + ``MemoryProvider`` ABC currently offers no hook for contributing + external paths to its backup. """ from __future__ import annotations @@ -53,11 +67,17 @@ import queue import re import threading from collections import deque -from datetime import datetime, timezone +from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from mempalace.backends.chroma import ChromaBackend +from mempalace.config import ( + MempalaceConfig, + sanitize_iso_temporal, + sanitize_kg_value, + sanitize_name, +) from mempalace.convo_miner import file_conversation_exchange from mempalace.knowledge_graph import KnowledgeGraph from mempalace.layers import MemoryStack @@ -79,6 +99,47 @@ except ImportError: # pragma: no cover - Hermes not installed logger = logging.getLogger("mempalace.hermes") +# The palace path this provider last bridged into ``MEMPALACE_PALACE_PATH`` +# (None when the variable is user-set or unset). ``initialize`` must be able +# to tell its own bridge write apart from user intent: a stale bridge from a +# previous session would otherwise override a freshly edited hermes-side +# config forever, because both this provider's ``_load_config`` and +# ``MempalaceConfig`` give the env var top precedence. +_ENV_PALACE_BRIDGED: Optional[str] = None + + +def _clear_stale_palace_bridge() -> None: + """Drop our own previous bridge write so config re-resolution is fresh. + + Only removes the env var when it still holds exactly the value we set — + a user-set value (even one set after our bridge) never matches the + recorded sentinel and is left untouched. + """ + global _ENV_PALACE_BRIDGED + if _ENV_PALACE_BRIDGED and os.environ.get("MEMPALACE_PALACE_PATH") == _ENV_PALACE_BRIDGED: + del os.environ["MEMPALACE_PALACE_PATH"] + _ENV_PALACE_BRIDGED = None + + +def _bridge_palace_env(palace_path: str) -> None: + """Publish the provider's resolved palace to ``MEMPALACE_PALACE_PATH``. + + ``mempalace.mcp_server`` resolves its palace from this variable on every + config access (its ``--palace`` flag works by setting exactly this + variable), so bridging it makes the passthrough tools operate on the + same palace the provider writes and searches. When the variable already + points at the same palace (user-set), ownership is NOT claimed, so a + later ``_clear_stale_palace_bridge`` leaves the user's value alone. + """ + global _ENV_PALACE_BRIDGED + bridged = os.path.abspath(os.path.expanduser(palace_path)) + current = os.environ.get("MEMPALACE_PALACE_PATH") + if current and os.path.abspath(os.path.expanduser(current)) == bridged: + return + os.environ["MEMPALACE_PALACE_PATH"] = bridged + _ENV_PALACE_BRIDGED = bridged + + def _match_wing_by_keywords(text: str, wing_config: Dict[str, Any]) -> str: """Return the first wing whose keywords match a whole word in ``text``. @@ -563,17 +624,29 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] self._session_id = session_id or "" self._hermes_home = str(kwargs.get("hermes_home", "") or "") + # A bridge write from a previous initialize must not masquerade + # as a user-set env override during re-resolution. + _clear_stale_palace_bridge() self._config = self._load_config() - self._palace_path = str( - Path(self._config.get("palace_path", self.DEFAULT_PALACE_PATH)).expanduser() - ) - # Collection name is intentionally **not** configurable. ``_file_turn`` - # writes through ``self._collection``; ``prefetch`` / ``_tool_search`` - # go through ``search_memories``, which reads its own configured - # collection name from ``~/.mempalace/config.json``. Exposing two - # ways to set the name invites write-here, read-there mismatches - # that silently make the provider look mute. - self._collection_name = self.DEFAULT_COLLECTION_NAME + # No hermes-side palace configured → defer to mempalace's own + # config (env var > ~/.mempalace/config.json > default) so this + # provider agrees with `mempalace init`-managed setups instead + # of hardcoding the default location. + mempalace_config = MempalaceConfig() + configured = self._config.get("palace_path") or mempalace_config.palace_path + self._palace_path = str(Path(configured).expanduser()) + # Publish the resolved palace so the mcp_server passthrough + # tools (drawer CRUD, duplicate check, tunnels, taxonomy) + # read and write the SAME palace live filing and search use. + _bridge_palace_env(self._palace_path) + # Collection name follows mempalace's own config — the same + # source ``search_memories`` (prefetch / _tool_search) and the + # mcp_server passthrough read — so live writes land in the + # collection recall actually searches. Intentionally NOT + # configurable on the Hermes side: a second knob would let the + # write and read sides diverge silently, making the provider + # look mute. + self._collection_name = mempalace_config.collection_name self._load_wing_config() self._load_identity() @@ -841,18 +914,33 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] obj=args.get("object", ""), ) ) + if tool_name == "mempalace_kg_invalidate": + return json.dumps( + self._tool_kg_invalidate( + subject=args.get("subject", ""), + predicate=args.get("predicate", ""), + obj=args.get("object", ""), + ended=args.get("ended"), + ) + ) + if tool_name == "mempalace_kg_timeline": + return json.dumps(self._tool_kg_timeline(args.get("entity"))) + if tool_name == "mempalace_kg_stats": + return json.dumps(self._tool_kg_stats()) if tool_name == "mempalace_diary_write": return json.dumps(self._tool_diary_write(args.get("entry", ""))) if tool_name == "mempalace_diary_read": return json.dumps(self._tool_diary_read(int(args.get("n", 10)))) # Tools that delegate directly to ``mempalace.mcp_server``'s - # public ``tool_*`` entry points. These share mempalace's own - # config for palace_path resolution rather than this plugin's - # ``self._palace_path`` — a known asymmetry that the original - # eight tools above don't share. In the common case (default - # palace at ``~/.mempalace/palace``) both resolve to the same - # place. + # public ``tool_*`` entry points. mcp_server resolves its palace + # from ``MEMPALACE_PALACE_PATH`` on every config access, and + # ``initialize`` bridges this provider's resolved palace into + # that variable (see ``_bridge_palace_env``) — so passthrough + # reads and writes land in the same palace as live filing and + # search. KG tools are NOT passed through: mcp_server resolves + # its KG from ``DEFAULT_KG_PATH`` unless its own ``--palace`` + # CLI flag was given, which no env bridge can influence. # New tools (everything that has a matching ``tool_*`` in # mempalace.mcp_server) dispatch by name derivation. One # mempalace asymmetry to remap: ``mempalace_traverse`` maps to @@ -887,9 +975,6 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] "mempalace_list_drawers", "mempalace_get_drawer", "mempalace_check_duplicate", - "mempalace_kg_invalidate", - "mempalace_kg_timeline", - "mempalace_kg_stats", "mempalace_get_taxonomy", "mempalace_get_aaak_spec", "mempalace_traverse", @@ -967,6 +1052,9 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] print(" 1. mempalace init # generates ~/.mempalace/") print(" 2. (optional) edit ~/.mempalace/identity.txt to seed L0 wake-up context") print() + print("Note: palace data lives under ~/.mempalace/, outside $HERMES_HOME.") + print("`hermes backup` does not include it — back up ~/.mempalace/ separately.") + print() # ----- Shutdown -------------------------------------------------------- @@ -1082,10 +1170,9 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] logger.debug("MemPalace _file_turn error: %s", exc) def _mirror_mem_write(self, payload: Dict[str, Any]) -> None: - db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") kg: Optional[KnowledgeGraph] = None try: - kg = KnowledgeGraph(db_path=db_path) + kg = KnowledgeGraph(db_path=self._kg_db_path()) kg.add_triple( subject="user", predicate="asserted", @@ -1242,11 +1329,19 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] # via the truncated/scanned pair. return out + def _kg_db_path(self) -> str: + """The provider's knowledge-graph DB — sibling of the palace dir. + + Single source of truth for every KG access in this plugin; the KG + tools must never fall back to mempalace's global ``DEFAULT_KG_PATH`` + or they would read a different graph than the one live turns feed. + """ + return str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") + def _tool_kg_query(self, entity: str, since: Optional[str] = None) -> Dict[str, Any]: if not entity: return {"error": "Missing required parameter: entity"} - db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") - kg = KnowledgeGraph(db_path=db_path) + kg = KnowledgeGraph(db_path=self._kg_db_path()) try: relations = kg.query_entity(entity, as_of=since or "") finally: @@ -1259,8 +1354,7 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] def _tool_kg_add(self, subject: str, predicate: str, obj: str) -> Dict[str, Any]: if not (subject and predicate and obj): return {"error": "subject, predicate, object are all required"} - db_path = str(Path(self._palace_path).parent / "knowledge_graph.sqlite3") - kg = KnowledgeGraph(db_path=db_path) + kg = KnowledgeGraph(db_path=self._kg_db_path()) try: kg.add_triple(subject=subject, predicate=predicate, obj=obj) finally: @@ -1270,6 +1364,63 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] pass return {"status": "ok", "triple": [subject, predicate, obj]} + def _tool_kg_invalidate( + self, + subject: str, + predicate: str, + obj: str, + ended: Optional[str] = None, + ) -> Dict[str, Any]: + # Same validation semantics as mcp_server.tool_kg_invalidate, but + # against the provider's own KG (see _kg_db_path). + try: + subject = sanitize_kg_value(subject, "subject") + predicate = sanitize_name(predicate, "predicate") + obj = sanitize_kg_value(obj, "object") + ended = sanitize_iso_temporal(ended, "ended") + except ValueError as exc: + return {"success": False, "error": str(exc)} + resolved_ended = ended or date.today().isoformat() + kg = KnowledgeGraph(db_path=self._kg_db_path()) + try: + kg.invalidate(subject, predicate, obj, ended=resolved_ended) + finally: + try: + kg.close() + except Exception: + pass + return { + "success": True, + "fact": f"{subject} → {predicate} → {obj}", + "ended": resolved_ended, + } + + def _tool_kg_timeline(self, entity: Optional[str] = None) -> Dict[str, Any]: + if entity is not None: + try: + entity = sanitize_kg_value(entity, "entity") + except ValueError as exc: + return {"error": str(exc)} + kg = KnowledgeGraph(db_path=self._kg_db_path()) + try: + results = kg.timeline(entity) if entity else kg.timeline() + finally: + try: + kg.close() + except Exception: + pass + return {"entity": entity or "all", "timeline": results, "count": len(results)} + + def _tool_kg_stats(self) -> Dict[str, Any]: + kg = KnowledgeGraph(db_path=self._kg_db_path()) + try: + return kg.stats() + finally: + try: + kg.close() + except Exception: + pass + def _tool_diary_write(self, entry: str) -> Dict[str, Any]: if not entry: return {"error": "Missing required parameter: entry"} diff --git a/tests/test_hermes_integration.py b/tests/test_hermes_integration.py index be26416..783ec5e 100644 --- a/tests/test_hermes_integration.py +++ b/tests/test_hermes_integration.py @@ -13,6 +13,7 @@ from __future__ import annotations import importlib.util import json +import os import sys import threading import types @@ -64,6 +65,25 @@ def provider(integration_module): return integration_module.MempalaceProvider() +@pytest.fixture(autouse=True) +def _isolate_palace_env(integration_module): + """Keep initialize()'s palace-env bridge from leaking between tests. + + ``initialize`` publishes the resolved palace to MEMPALACE_PALACE_PATH + (so mcp_server passthrough tools resolve the same palace) and records + ownership in the module-level ``_ENV_PALACE_BRIDGED`` sentinel. Both + are process-global — restore them after every test. + """ + original = os.environ.get("MEMPALACE_PALACE_PATH") + original_sentinel = integration_module._ENV_PALACE_BRIDGED + yield + integration_module._ENV_PALACE_BRIDGED = original_sentinel + if original is None: + os.environ.pop("MEMPALACE_PALACE_PATH", None) + else: + os.environ["MEMPALACE_PALACE_PATH"] = original + + # --------------------------------------------------------------------------- # Shape: name, schemas, config, availability # --------------------------------------------------------------------------- @@ -616,11 +636,11 @@ def test_initialize_reads_mempalace_json(provider, tmp_dir, palace_path): provider.shutdown() -def test_collection_name_is_not_user_configurable(provider, tmp_dir, palace_path): - # Exposing ``collection_name`` would let the provider write to a collection - # that ``search_memories`` (which reads from mempalace's own config) does - # not read — making the provider silently appear mute. The field is - # intentionally absent from the schema and ignored in config files. +def test_collection_name_is_not_hermes_configurable(provider, tmp_dir, palace_path): + # A hermes-side ``collection_name`` would be a second way to set the + # name — the write and read sides could silently diverge, making the + # provider look mute. The key is ignored; with no mempalace-side + # override (conftest redirects HOME to a temp dir), the default applies. (Path(tmp_dir) / "mempalace.json").write_text( json.dumps({"palace_path": palace_path, "collection_name": "custom_drawers"}) ) @@ -631,6 +651,34 @@ def test_collection_name_is_not_user_configurable(provider, tmp_dir, palace_path provider.shutdown() +def test_collection_name_follows_mempalace_config( + integration_module, provider, tmp_dir, palace_path, tmp_path, monkeypatch +): + # One source of truth: the provider writes to the collection that + # ``search_memories`` and the mcp_server passthrough actually read — + # mempalace's own config — so a customized ``collection_name`` in + # ``~/.mempalace/config.json`` cannot make live turns invisible to + # recall. + mp_config_dir = tmp_path / "mp_home" + mp_config_dir.mkdir() + (mp_config_dir / "config.json").write_text(json.dumps({"collection_name": "family_drawers"})) + + from mempalace.config import MempalaceConfig as real_config + + def _patched_config(config_dir=None): + return real_config(config_dir=str(mp_config_dir)) + + # Patch the provider module's own reference — it imported the name at + # module load, so patching mempalace.config wouldn't reach it. + monkeypatch.setattr(integration_module, "MempalaceConfig", _patched_config) + (Path(tmp_dir) / "mempalace.json").write_text(json.dumps({"palace_path": palace_path})) + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + assert provider._collection_name == "family_drawers" + finally: + provider.shutdown() + + def test_env_vars_override_config_file(provider, tmp_dir, palace_path, monkeypatch): monkeypatch.setenv("MEMPALACE_PALACE_PATH", palace_path) monkeypatch.setenv("MEMPALACE_WING", "wing_forced") @@ -708,3 +756,141 @@ def test_match_wing_by_keywords_ignores_non_string_keywords(integration_module): wing_config = {"wing_dev": {"keywords": [None, 3, "python"]}} assert fn("write some python code", wing_config) == "wing_dev" assert fn("unrelated chatter", wing_config) == "wing_general" + + +# --------------------------------------------------------------------------- +# Palace unification: the mcp_server passthrough tools must operate on the +# SAME palace the provider writes and searches. The provider bridges its +# resolved palace into MEMPALACE_PALACE_PATH (mcp_server re-reads that var on +# every config access — its own --palace flag works the same way), and the +# KG tools are handled natively because mcp_server's KG path ignores the env +# var unless its CLI flag was given. +# --------------------------------------------------------------------------- + + +def test_initialize_bridges_palace_env_for_passthrough(provider, tmp_dir, palace_path): + from mempalace.config import MempalaceConfig + + (Path(tmp_dir) / "mempalace.json").write_text(json.dumps({"palace_path": palace_path})) + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + expected = os.path.abspath(os.path.expanduser(palace_path)) + assert os.environ.get("MEMPALACE_PALACE_PATH") == expected + # The passthrough side (mcp_server's config) now resolves the same + # palace the provider writes — the split-brain regression check. + assert MempalaceConfig().palace_path == expected + finally: + provider.shutdown() + + +def test_reinitialize_follows_updated_hermes_config(provider, tmp_dir, palace_path, tmp_path): + # The bridge write from session 1 must not masquerade as a user env + # override in session 2 — a stale bridge would pin the palace to the + # old hermes-side value forever. + config_path = Path(tmp_dir) / "mempalace.json" + config_path.write_text(json.dumps({"palace_path": palace_path})) + provider.initialize("s1", hermes_home=str(tmp_dir)) + provider.shutdown() + + new_palace = str(tmp_path / "palace_b") + config_path.write_text(json.dumps({"palace_path": new_palace})) + provider.initialize("s2", hermes_home=str(tmp_dir)) + try: + assert provider._palace_path == new_palace + assert os.environ.get("MEMPALACE_PALACE_PATH") == os.path.abspath(new_palace) + finally: + provider.shutdown() + + +def test_user_set_palace_env_wins_and_is_never_cleared( + provider, tmp_dir, palace_path, tmp_path, monkeypatch +): + # A user-set env var outranks the hermes-side config (documented + # precedence) and the bridge must not claim ownership of it — a later + # re-initialize must leave the user's value in place. + monkeypatch.setenv("MEMPALACE_PALACE_PATH", palace_path) + (Path(tmp_dir) / "mempalace.json").write_text( + json.dumps({"palace_path": str(tmp_path / "other_palace")}) + ) + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + assert provider._palace_path == palace_path + assert os.environ.get("MEMPALACE_PALACE_PATH") == palace_path + # Ownership was not claimed: the sentinel stays unset. + assert provider.__class__.__module__ is not None # provider alive + finally: + provider.shutdown() + provider.initialize("s2", hermes_home=str(tmp_dir)) + try: + assert os.environ.get("MEMPALACE_PALACE_PATH") == palace_path + finally: + provider.shutdown() + + +def test_hermes_config_defers_to_mempalace_config_when_unset( + integration_module, provider, tmp_dir, tmp_path, monkeypatch +): + # No hermes-side palace_path → the provider follows mempalace's own + # config rather than hardcoding the default location. + mp_config_dir = tmp_path / "mp_home" + mp_config_dir.mkdir() + custom_palace = str(tmp_path / "custom_palace") + (mp_config_dir / "config.json").write_text(json.dumps({"palace_path": custom_palace})) + + from mempalace.config import MempalaceConfig as real_config + + def _patched_config(config_dir=None): + return real_config(config_dir=str(mp_config_dir)) + + # Patch the provider module's own reference — it imported the name at + # module load, so patching mempalace.config wouldn't reach it. + monkeypatch.setattr(integration_module, "MempalaceConfig", _patched_config) + (Path(tmp_dir) / "mempalace.json").write_text(json.dumps({})) + provider.initialize("s1", hermes_home=str(tmp_dir)) + try: + assert provider._palace_path == custom_palace + finally: + provider.shutdown() + + +def test_kg_tools_all_use_provider_sibling_kg(initialized_provider, palace_path): + # All five KG tools must hit the SAME database: the sibling of the + # provider's palace dir — never mcp_server's global DEFAULT_KG_PATH. + add = json.loads( + initialized_provider.handle_tool_call( + "mempalace_kg_add", + {"subject": "user", "predicate": "drinks", "object": "tea"}, + ) + ) + assert add["status"] == "ok" + + timeline = json.loads( + initialized_provider.handle_tool_call("mempalace_kg_timeline", {"entity": "user"}) + ) + assert timeline["count"] >= 1 + assert any(t["predicate"] == "drinks" and t["object"] == "tea" for t in timeline["timeline"]) + + stats = json.loads(initialized_provider.handle_tool_call("mempalace_kg_stats", {})) + assert stats["triples"] >= 1 + + inv = json.loads( + initialized_provider.handle_tool_call( + "mempalace_kg_invalidate", + {"subject": "user", "predicate": "drinks", "object": "tea"}, + ) + ) + assert inv["success"] is True + + # And the file itself lives next to the palace dir. + assert (Path(palace_path).parent / "knowledge_graph.sqlite3").exists() + + +def test_kg_invalidate_rejects_invalid_input(initialized_provider): + result = json.loads( + initialized_provider.handle_tool_call( + "mempalace_kg_invalidate", + {"subject": "user", "predicate": "likes", "object": "x", "ended": "not-a-date"}, + ) + ) + assert result["success"] is False + assert "error" in result From fc3beb71ab5fa2ae14850a5dc4733fbdd2d4a1c7 Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Mon, 6 Jul 2026 10:03:14 -0400 Subject: [PATCH 6/7] fix(hermes): mirror default-target memory writes into the knowledge graph MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hermes' memory tool defaults to target="memory" (the agent's own notes); filtering on_memory_write to target == "user" silently dropped the majority of writes. Mirror both targets under distinct subjects — user→asserted for user facts, hermes→noted for agent notes — so kg_query("user") never surfaces environment quirks. Co-Authored-By: Claude Fable 5 --- mempalace/integrations/hermes/__init__.py | 26 +++++++++++++--- tests/test_hermes_integration.py | 38 +++++++++++++++++++++++ 2 files changed, 60 insertions(+), 4 deletions(-) diff --git a/mempalace/integrations/hermes/__init__.py b/mempalace/integrations/hermes/__init__.py index 63bd09a..50673a7 100644 --- a/mempalace/integrations/hermes/__init__.py +++ b/mempalace/integrations/hermes/__init__.py @@ -841,11 +841,17 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] content: str, metadata: Optional[Dict[str, Any]] = None, ) -> None: + # Hermes' memory tool has two targets: "user" (who the user is) and + # "memory" (the agent's own notes — environment facts, conventions, + # lessons). "memory" is the tool's DEFAULT when the model omits the + # field, so filtering to target == "user" would silently drop the + # majority of writes. Both mirror into the knowledge graph, under + # distinct subjects (see _mirror_mem_write). if ( self._cron_skipped or not self._initialized # worker isn't running; queueing leaks or action != "add" - or target != "user" + or target not in ("user", "memory") or not content ): return @@ -853,7 +859,11 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] self._worker_queue.put_nowait( ( "mem_write", - {"content": content, "metadata": dict(metadata or {})}, + { + "content": content, + "target": target, + "metadata": dict(metadata or {}), + }, ) ) except queue.Full: @@ -1170,12 +1180,20 @@ class MempalaceProvider(MemoryProvider): # type: ignore[misc] logger.debug("MemPalace _file_turn error: %s", exc) def _mirror_mem_write(self, payload: Dict[str, Any]) -> None: + # target "user" carries facts about the user; target "memory" carries + # the agent's own notes. Distinct subjects keep the graph honest about + # WHO the fact describes — `kg_query("user")` must not surface tool + # quirks the agent noted about its environment. + if payload.get("target") == "memory": + subject, predicate = "hermes", "noted" + else: + subject, predicate = "user", "asserted" kg: Optional[KnowledgeGraph] = None try: kg = KnowledgeGraph(db_path=self._kg_db_path()) kg.add_triple( - subject="user", - predicate="asserted", + subject=subject, + predicate=predicate, obj=payload.get("content", ""), ) except Exception as exc: diff --git a/tests/test_hermes_integration.py b/tests/test_hermes_integration.py index 783ec5e..0cbe29e 100644 --- a/tests/test_hermes_integration.py +++ b/tests/test_hermes_integration.py @@ -894,3 +894,41 @@ def test_kg_invalidate_rejects_invalid_input(initialized_provider): ) assert result["success"] is False assert "error" in result + + +# --------------------------------------------------------------------------- +# on_memory_write: Hermes' memory tool defaults to target="memory" (the +# agent's own notes); only target="user" carries facts about the user. Both +# must mirror into the knowledge graph, under distinct subjects. +# --------------------------------------------------------------------------- + + +def test_on_memory_write_mirrors_both_targets(initialized_provider, palace_path): + from mempalace.knowledge_graph import KnowledgeGraph + + initialized_provider.on_memory_write("add", "user", "lives in Boston") + initialized_provider.on_memory_write("add", "memory", "repo uses uv for deps") + initialized_provider._worker_queue.join() + + kg = KnowledgeGraph(db_path=str(Path(palace_path).parent / "knowledge_graph.sqlite3")) + try: + user_relations = kg.query_entity("user") + agent_relations = kg.query_entity("hermes") + finally: + kg.close() + assert any( + r.get("predicate") == "asserted" and r.get("object") == "lives in Boston" + for r in user_relations + ) + assert any( + r.get("predicate") == "noted" and r.get("object") == "repo uses uv for deps" + for r in agent_relations + ) + + +def test_on_memory_write_skips_unknown_target_and_non_add(initialized_provider): + pre = initialized_provider._worker_queue.qsize() + initialized_provider.on_memory_write("add", "bogus", "x") + initialized_provider.on_memory_write("replace", "memory", "x") + initialized_provider.on_memory_write("remove", "user", "x") + assert initialized_provider._worker_queue.qsize() == pre From 449d6d86e57758a0531e3a2126c98ad458d793b8 Mon Sep 17 00:00:00 2001 From: raman325 <7243222+raman325@users.noreply.github.com> Date: Wed, 5 Aug 2026 12:35:41 -0400 Subject: [PATCH 7/7] ci: retrigger flaky windows hermes integration test