feat: detect embedding model mismatch on collection open

Stores the embedding model identity in ChromaDB collection metadata
at creation time. On every collection open, compares stored model to
current config. Raises EmbeddingModelMismatchError on mismatch with
recovery instructions (mempalace re-mine or MEMPALACE_FORCE_EMBEDDING).

- get_embedding_model_name() returns canonical model identity string
- _resolve_model_and_device() shared by both name and function resolvers
- force_embedding config (env var / config.json) to bypass check
- Legacy palaces silently stamped on first open
- All 7 direct ChromaDB callers routed through palace.get_collection()
- Fixed pre-existing test env isolation bug (monkeypatch.delenv)

Author: FabioLissi

mempalace/cli.py

@@ -33,7 +33,8 @@
 import argparse
 from pathlib import Path
 
-from .config import MempalaceConfig, get_embedding_function
+from .config import MempalaceConfig
+from .palace import get_collection as _palace_get_collection
 
 
 def cmd_init(args):
@@ -183,9 +184,7 @@ def cmd_repair(args):
 
     # Try to read existing drawers
     try:
-        ef = get_embedding_function()
-        client = chromadb.PersistentClient(path=palace_path)
-        col = client.get_collection("mempalace_drawers", embedding_function=ef)
+        col = _palace_get_collection(palace_path, force=True)
         total = col.count()
         print(f"  Drawers found: {total}")
     except Exception as e:
@@ -221,8 +220,9 @@ def cmd_repair(args):
     shutil.copytree(palace_path, backup_path)
 
     print("  Rebuilding collection...")
+    client = chromadb.PersistentClient(path=palace_path)
     client.delete_collection("mempalace_drawers")
-    new_col = client.create_collection("mempalace_drawers", embedding_function=ef)
+    new_col = _palace_get_collection(palace_path)
 
     filed = 0
     for i in range(0, len(all_ids), batch_size):
@@ -275,7 +275,6 @@ def cmd_mcp(args):
 
 def cmd_compress(args):
     """Compress drawers in a wing using AAAK Dialect."""
-    import chromadb
     from .dialect import Dialect
 
     palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
@@ -296,9 +295,7 @@ def cmd_compress(args):
 
     # Connect to palace
     try:
-        ef = get_embedding_function()
-        client = chromadb.PersistentClient(path=palace_path)
-        col = client.get_collection("mempalace_drawers", embedding_function=ef)
+        col = _palace_get_collection(palace_path)
     except Exception:
         print(f"\n  No palace found at {palace_path}")
         print("  Run: mempalace init <dir> then mempalace mine <dir>")
@@ -369,9 +366,7 @@ def cmd_compress(args):
     # Store compressed versions (unless dry-run)
     if not args.dry_run:
         try:
-            comp_col = client.get_or_create_collection(
-                "mempalace_compressed", embedding_function=ef
-            )
+            comp_col = _palace_get_collection(palace_path, "mempalace_compressed")
             for doc_id, compressed, meta, stats in compressed_entries:
                 comp_meta = dict(meta)
                 comp_meta["compression_ratio"] = round(stats["ratio"], 1)

mempalace/config.py

@@ -222,6 +222,20 @@ def embedding_device(self):
             return env_val
         return self._file_config.get("embedding_device", None)
 
+    @property
+    def force_embedding(self):
+        """Whether to bypass embedding model mismatch checks.
+
+        Priority: ``MEMPALACE_FORCE_EMBEDDING`` env var > ``config.json``
+        ``"force_embedding"`` key > ``False`` default.
+
+        Env var: ``"true"`` (case-insensitive) to enable, anything else is false.
+        """
+        env_val = os.environ.get("MEMPALACE_FORCE_EMBEDDING")
+        if env_val is not None:
+            return env_val.lower() == "true"
+        return bool(self._file_config.get("force_embedding", False))
+
     def save_people_map(self, people_map):
         """Write people_map.json to config directory.
 
@@ -245,6 +259,57 @@ def save_people_map(self, people_map):
 _DEFAULT_MODEL_FOR_DEVICE = "sentence-transformers/all-MiniLM-L6-v2"
 
 
+class EmbeddingModelMismatchError(Exception):
+    """Raised when palace was created with a different embedding model."""
+
+    def __init__(self, stored_model: str, current_model: str):
+        self.stored_model = stored_model
+        self.current_model = current_model
+        super().__init__(
+            f"Embedding model mismatch.\n"
+            f"Palace was created with: {stored_model}\n"
+            f"Currently configured:    {current_model}\n\n"
+            f"To switch models, re-mine your palace:\n"
+            f"  mempalace re-mine\n\n"
+            f"Or set MEMPALACE_FORCE_EMBEDDING=true to bypass this check."
+        )
+
+
+def _resolve_model_and_device(config=None):
+    """Resolve the effective model name and device from config.
+
+    Returns (model_name, device) where model_name is None when no
+    explicit model is configured (i.e. ChromaDB default should be used).
+    """
+    cfg = config or MempalaceConfig()
+    model_name = cfg.embedding_model
+    device = cfg.embedding_device
+
+    # Ergonomic default: if the user asked for a device but didn't pick a
+    # model, use the same model ChromaDB uses by default so vectors stay
+    # compatible with existing palaces.
+    if not model_name and device:
+        model_name = _DEFAULT_MODEL_FOR_DEVICE
+
+    return model_name, device
+
+
+def get_embedding_model_name(config=None):
+    """Return the canonical identity string for the active embedding model.
+
+    Does **not** instantiate the model or touch the singleton cache.
+
+    Returns:
+        ``"chromadb-default"`` when no model/device is configured,
+        ``_DEFAULT_MODEL_FOR_DEVICE`` when only a device is set,
+        or the explicit model name string.
+    """
+    model_name, _ = _resolve_model_and_device(config)
+    if not model_name:
+        return "chromadb-default"
+    return model_name
+
+
 def get_embedding_function(config=None):
     """Return the configured ChromaDB embedding function.
 
@@ -275,15 +340,7 @@ def get_embedding_function(config=None):
 
     _embedding_function_resolved = True
 
-    cfg = config or MempalaceConfig()
-    model_name = cfg.embedding_model
-    device = cfg.embedding_device
-
-    # Ergonomic default: if the user asked for a device but didn't pick a
-    # model, use the same model ChromaDB uses by default so vectors stay
-    # compatible with existing palaces.
-    if not model_name and device:
-        model_name = _DEFAULT_MODEL_FOR_DEVICE
+    model_name, device = _resolve_model_and_device(config)
 
     if not model_name:
         # No explicit configuration — use ChromaDB's default embedder.

mempalace/layers.py

@@ -21,9 +21,8 @@
 from pathlib import Path
 from collections import defaultdict
 
-import chromadb
-
-from .config import MempalaceConfig, get_embedding_function
+from .config import MempalaceConfig
+from .palace import get_collection as _palace_get_collection
 
 
 # ---------------------------------------------------------------------------
@@ -91,10 +90,7 @@ def __init__(self, palace_path: str = None, wing: str = None):
     def generate(self) -> str:
         """Pull top drawers from ChromaDB and format as compact L1 text."""
         try:
-            client = chromadb.PersistentClient(path=self.palace_path)
-            col = client.get_collection(
-                "mempalace_drawers", embedding_function=get_embedding_function()
-            )
+            col = _palace_get_collection(self.palace_path)
         except Exception:
             return "## L1 — No palace found. Run: mempalace mine <dir>"
 
@@ -198,10 +194,7 @@ def __init__(self, palace_path: str = None):
     def retrieve(self, wing: str = None, room: str = None, n_results: int = 10) -> str:
         """Retrieve drawers filtered by wing and/or room."""
         try:
-            client = chromadb.PersistentClient(path=self.palace_path)
-            col = client.get_collection(
-                "mempalace_drawers", embedding_function=get_embedding_function()
-            )
+            col = _palace_get_collection(self.palace_path)
         except Exception:
             return "No palace found."
 
@@ -264,10 +257,7 @@ def __init__(self, palace_path: str = None):
     def search(self, query: str, wing: str = None, room: str = None, n_results: int = 5) -> str:
         """Semantic search, returns compact result text."""
         try:
-            client = chromadb.PersistentClient(path=self.palace_path)
-            col = client.get_collection(
-                "mempalace_drawers", embedding_function=get_embedding_function()
-            )
+            col = _palace_get_collection(self.palace_path)
         except Exception:
             return "No palace found."
 
@@ -322,10 +312,7 @@ def search_raw(
     ) -> list:
         """Return raw dicts instead of formatted text."""
         try:
-            client = chromadb.PersistentClient(path=self.palace_path)
-            col = client.get_collection(
-                "mempalace_drawers", embedding_function=get_embedding_function()
-            )
+            col = _palace_get_collection(self.palace_path)
         except Exception:
             return []
 
@@ -445,10 +432,7 @@ def status(self) -> dict:
 
         # Count drawers
         try:
-            client = chromadb.PersistentClient(path=self.palace_path)
-            col = client.get_collection(
-                "mempalace_drawers", embedding_function=get_embedding_function()
-            )
+            col = _palace_get_collection(self.palace_path)
             count = col.count()
             result["total_drawers"] = count
         except Exception:

mempalace/mcp_server.py

@@ -26,11 +26,11 @@
 from datetime import datetime
 from pathlib import Path
 
-from .config import MempalaceConfig, sanitize_name, sanitize_content, get_embedding_function
+from .config import MempalaceConfig, sanitize_name, sanitize_content
 from .version import __version__
+from .palace import iter_all_metadatas, get_collection as _palace_get_collection
 from .searcher import search_memories
 from .palace_graph import traverse, find_tunnels, graph_stats
-import chromadb
 
 from .knowledge_graph import KnowledgeGraph
 
@@ -100,31 +100,16 @@ def _wal_log(operation: str, params: dict, result: dict = None):
         logger.error(f"WAL write failed: {e}")
 
 
-_client_cache = None
 _collection_cache = None
 
 
-def _get_client():
-    """Return a singleton ChromaDB PersistentClient."""
-    global _client_cache
-    if _client_cache is None:
-        _client_cache = chromadb.PersistentClient(path=_config.palace_path)
-    return _client_cache
-
-
 def _get_collection(create=False):
     """Return the ChromaDB collection, caching the client between calls."""
     global _collection_cache
     try:
-        ef = get_embedding_function()
-        client = _get_client()
-        if create:
-            _collection_cache = client.get_or_create_collection(
-                _config.collection_name, embedding_function=ef
-            )
-        elif _collection_cache is None:
-            _collection_cache = client.get_collection(
-                _config.collection_name, embedding_function=ef
+        if create or _collection_cache is None:
+            _collection_cache = _palace_get_collection(
+                _config.palace_path, _config.collection_name
             )
         return _collection_cache
     except Exception:

mempalace/miner.py

@@ -15,9 +15,6 @@
 from datetime import datetime
 from collections import defaultdict
 
-import chromadb
-
-from .config import get_embedding_function
 from .palace import SKIP_DIRS, get_collection, file_already_mined
 
 READABLE_EXTENSIONS = {
@@ -625,10 +622,10 @@ def mine(
 
 def status(palace_path: str):
     """Show what's been filed in the palace."""
+    from .palace import iter_all_metadatas, get_collection as _palace_get_collection
+
     try:
-        ef = get_embedding_function()
-        client = chromadb.PersistentClient(path=palace_path)
-        col = client.get_collection("mempalace_drawers", embedding_function=ef)
+        col = _palace_get_collection(palace_path)
     except Exception:
         print(f"\n  No palace found at {palace_path}")
         print("  Run: mempalace init <dir> then mempalace mine <dir>")

mempalace/palace.py

@@ -4,10 +4,19 @@
 Consolidates ChromaDB access patterns used by both miners and the MCP server.
 """
 
+import logging
 import os
+
 import chromadb
 
-from .config import get_embedding_function
+from .config import (
+    EmbeddingModelMismatchError,
+    MempalaceConfig,
+    get_embedding_function,
+    get_embedding_model_name,
+)
+
+logger = logging.getLogger("mempalace")
 
 SKIP_DIRS = {
     ".git",
@@ -36,19 +45,51 @@
 }
 
 
-def get_collection(palace_path: str, collection_name: str = "mempalace_drawers"):
-    """Get or create the palace ChromaDB collection."""
+def get_collection(palace_path: str, collection_name: str = "mempalace_drawers", force: bool = None):
+    """Get or create the palace ChromaDB collection.
+
+    Verifies that the collection's embedding model matches the currently
+    configured model. Raises EmbeddingModelMismatchError on mismatch
+    unless force=True or MEMPALACE_FORCE_EMBEDDING=true.
+    """
     os.makedirs(palace_path, exist_ok=True)
     try:
         os.chmod(palace_path, 0o700)
     except (OSError, NotImplementedError):
         pass
+
     ef = get_embedding_function()
+    current_model = get_embedding_model_name()
+    if force is None:
+        force = MempalaceConfig().force_embedding
+
     client = chromadb.PersistentClient(path=palace_path)
     try:
-        return client.get_collection(collection_name, embedding_function=ef)
+        col = client.get_collection(collection_name, embedding_function=ef)
+        stored_model = (col.metadata or {}).get("embedding_model")
+
+        if stored_model is None:
+            # Legacy palace — silent stamp
+            col.modify(metadata={**(col.metadata or {}), "embedding_model": current_model})
+        elif stored_model != current_model:
+            if force:
+                logger.warning(
+                    "Embedding model mismatch (forced): %s -> %s",
+                    stored_model, current_model,
+                )
+                col.modify(metadata={**(col.metadata or {}), "embedding_model": current_model})
+            else:
+                raise EmbeddingModelMismatchError(stored_model, current_model)
+
+        return col
+    except EmbeddingModelMismatchError:
+        raise
     except Exception:
-        return client.create_collection(collection_name, embedding_function=ef)
+        return client.create_collection(
+            collection_name,
+            embedding_function=ef,
+            metadata={"embedding_model": current_model},
+        )
 
 
 def file_already_mined(collection, source_file: str, check_mtime: bool = False) -> bool: