Merge pull request #1422 from YuanYiZheXue/fix/lone-surrogate-clean

fix(mcp): sanitize lone surrogates before ChromaDB ops (#1235)

mempalace/config.py

@@ -19,6 +19,17 @@ from pathlib import Path
 MAX_NAME_LENGTH = 128
 _SAFE_NAME_RE = re.compile(r"^(?:[^\W_]|[^\W_][\w .'-]{0,126}[^\W_])$")
 
+# MCP clients (e.g. Claude Desktop, WorkBuddy) occasionally relay lone UTF-16
+# surrogates (U+D800–U+DFFF) when proxying binary-in-Unicode or corrupted
+# clipboard input. Python's ``str.encode('utf-8')`` raises on these, which
+# crashes ChromaDB add/upsert with -32000. See issue #1235.
+_LONE_SURROGATE_RE = re.compile(r"[\ud800-\udfff]")
+
+
+def strip_lone_surrogates(text: str) -> str:
+    """Replace lone UTF-16 surrogates with U+FFFD so the string is legal UTF-8 (#1235)."""
+    return _LONE_SURROGATE_RE.sub("�", text)
+
 
 def normalize_wing_name(name: str) -> str:
     """Lower-case + collapse separators (`-`, ` `) to `_` for wing slugs.
@@ -79,7 +90,7 @@ def sanitize_kg_value(value: str, field_name: str = "value") -> str:
     if "\x00" in value:
         raise ValueError(f"{field_name} contains null bytes")
 
-    return value
+    return strip_lone_surrogates(value)
 
 
 # ISO-8601 temporal validator for knowledge-graph temporal parameters
@@ -175,7 +186,7 @@ def sanitize_content(value: str, max_length: int = 100_000) -> str:
         raise ValueError(f"content exceeds maximum length of {max_length} characters")
     if "\x00" in value:
         raise ValueError("content contains null bytes")
-    return value
+    return strip_lone_surrogates(value)
 
 
 DEFAULT_PALACE_PATH = os.path.expanduser("~/.mempalace/palace")

mempalace/mcp_server.py

@@ -60,6 +60,7 @@ from .config import (  # noqa: E402
     sanitize_name,
     sanitize_content,
     sanitize_iso_temporal,
+    strip_lone_surrogates,
 )
 from .version import __version__  # noqa: E402
 from chromadb.errors import NotFoundError as _ChromaNotFoundError  # noqa: E402
@@ -964,6 +965,7 @@ def tool_check_duplicate(content: str, threshold: float = 0.9):
     if not col:
         return _no_palace()
     try:
+        content = strip_lone_surrogates(content)
         results = col.query(
             query_texts=[content],
             n_results=5,
@@ -1120,6 +1122,9 @@ def tool_add_drawer(
         wing = sanitize_name(wing, "wing")
         room = sanitize_name(room, "room")
         content = sanitize_content(content)
+        if source_file:
+            source_file = strip_lone_surrogates(source_file)
+        added_by = strip_lone_surrogates(added_by)
     except ValueError as e:
         return {"success": False, "error": str(e)}
 

mempalace/query_sanitizer.py

@@ -21,6 +21,8 @@ Expected recovery:
 import re
 import logging
 
+from mempalace.config import strip_lone_surrogates
+
 logger = logging.getLogger("mempalace_mcp")
 
 # --- Constants ---
@@ -65,7 +67,9 @@ def sanitize_query(raw_query: str) -> dict:
             "method": "passthrough",
         }
 
-    raw_query = raw_query.strip()
+    # Strip lone surrogates before any further processing so downstream embed
+    # calls don't crash on UTF-8 encode (#1235).
+    raw_query = strip_lone_surrogates(raw_query.strip())
     original_length = len(raw_query)
 
     def _strip_wrapping_quotes(candidate: str) -> str:

tests/conftest.py

@@ -33,6 +33,23 @@ import pytest  # noqa: E402
 from mempalace.config import MempalaceConfig  # noqa: E402
 from mempalace.knowledge_graph import KnowledgeGraph  # noqa: E402
 
+# Redirect ChromaDB's ONNX model cache back to the real user's cache so tests
+# don't re-download the 79 MB model on every run. The HOME redirect above
+# would otherwise point ONNXMiniLM_L6_V2.DOWNLOAD_PATH at the empty temp dir.
+try:
+    from pathlib import Path  # noqa: E402
+    from chromadb.utils.embedding_functions.onnx_mini_lm_l6_v2 import (  # noqa: E402
+        ONNXMiniLM_L6_V2,
+    )
+
+    _real_home = _original_env.get("USERPROFILE") or _original_env.get("HOME")
+    if _real_home:
+        _real_cache = Path(_real_home) / ".cache" / "chroma" / "onnx_models" / "all-MiniLM-L6-v2"
+        if _real_cache.exists():
+            ONNXMiniLM_L6_V2.DOWNLOAD_PATH = _real_cache
+except ImportError:
+    pass
+
 
 @pytest.fixture(autouse=True)
 def _reset_mcp_cache():

tests/test_clean_lone_surrogates.py

@@ -0,0 +1,179 @@
+"""Tests for lone-surrogate sanitisation (issue #1235).
+
+Covers:
+- Unit: ``strip_lone_surrogates()`` edge cases (lone surrogates, real emoji,
+  empty input, hashing).
+- Integration: every MCP tool that takes a user string and reaches ChromaDB
+  must survive lone surrogates in its payload.
+
+MCP clients (Claude Desktop, WorkBuddy) occasionally relay lone UTF-16
+surrogates (U+D800–U+DFFF) when proxying binary-in-Unicode or corrupted
+clipboard input. Python's ``str.encode('utf-8')`` raises on these, which
+crashes ChromaDB add/upsert with -32000 Internal Error.
+
+Note on surrogate pairs in Python source: writing ``"\\ud83d\\ude00"`` in
+Python creates a string with *two* lone surrogates, not the astral emoji
+U+1F600. Python 3 does not silently merge them. Use ``"\\U0001f600"`` to
+embed a real emoji as a single code point.
+"""
+
+import hashlib
+
+from mempalace.config import strip_lone_surrogates
+
+
+# ── Unit tests ─────────────────────────────────────────────────────────────
+
+
+class TestStripLoneSurrogates:
+    def test_passthrough_normal(self):
+        assert strip_lone_surrogates("hello world") == "hello world"
+        assert strip_lone_surrogates("你好世界") == "你好世界"
+        assert strip_lone_surrogates("mixed 中 English 文") == "mixed 中 English 文"
+
+    def test_empty_string(self):
+        assert strip_lone_surrogates("") == ""
+
+    def test_replaces_high_surrogate(self):
+        assert strip_lone_surrogates("hello\udc95world") == "hello�world"
+        assert strip_lone_surrogates("\udcff\udc00\udcaf") == "�" * 3
+
+    def test_replaces_low_surrogate(self):
+        assert strip_lone_surrogates("test\ud800more") == "test�more"
+        assert strip_lone_surrogates("\ud800\udbff") == "�" * 2
+
+    def test_replaces_multiple_at_different_positions(self):
+        assert strip_lone_surrogates("a\udca1b\udcffc") == "a�b�c"
+
+    def test_preserves_real_emoji(self):
+        """Astral code points written with \\U have no surrogates and pass through."""
+        assert strip_lone_surrogates("\U0001f600") == "\U0001f600"
+        assert strip_lone_surrogates("\U0001f680") == "\U0001f680"
+        assert strip_lone_surrogates("hello \U0001f600 world") == "hello \U0001f600 world"
+
+    def test_real_emoji_with_adjacent_lone_surrogate(self):
+        assert strip_lone_surrogates("\U0001f600\udc95") == "\U0001f600�"
+        assert strip_lone_surrogates("\udc95\U0001f600") == "�\U0001f600"
+
+    def test_only_surrogates(self):
+        assert strip_lone_surrogates("\udc95\udcff") == "��"
+        assert strip_lone_surrogates("\ud800" + "\udbff" + "\udc00" + "\udfff") == "�" * 4
+
+    def test_cleaned_string_is_utf8_encodable(self):
+        """The actual crash path: encode/hash must not raise after cleaning."""
+        dirty = "content\udc95with\ud800surrogate"
+        clean = strip_lone_surrogates(dirty)
+        h = hashlib.sha256(clean.encode("utf-8")).hexdigest()
+        assert len(h) == 64
+
+    def test_workbuddy_injected_surrogate(self):
+        """The specific surrogate observed in WorkBuddy production logs."""
+        result = strip_lone_surrogates("2026-04-27\udcadworkBuddy relay")
+        assert result == "2026-04-27�workBuddy relay"
+        # Must not raise — was the original failure mode.
+        hashlib.sha256(result.encode("utf-8")).hexdigest()
+
+
+# ── Sanitizer integration ──────────────────────────────────────────────────
+
+
+class TestSanitizersStripSurrogates:
+    """sanitize_content and sanitize_kg_value clean inline — verifies the
+    architectural fix so all MCP tools route through and inherit it for free."""
+
+    def test_sanitize_content_strips_surrogates(self):
+        from mempalace.config import sanitize_content
+
+        assert sanitize_content("hello\udc95world") == "hello�world"
+
+    def test_sanitize_kg_value_strips_surrogates(self):
+        from mempalace.config import sanitize_kg_value
+
+        assert sanitize_kg_value("Alice\udc95") == "Alice�"
+
+    def test_sanitize_query_strips_surrogates(self):
+        from mempalace.query_sanitizer import sanitize_query
+
+        result = sanitize_query("search\udc95term")
+        assert "\udc95" not in result["clean_query"]
+        assert "�" in result["clean_query"]
+
+
+# ── End-to-end tool integration ────────────────────────────────────────────
+
+
+def _patch_mcp_server(monkeypatch, config, kg):
+    from mempalace import mcp_server
+
+    monkeypatch.setattr(mcp_server, "_config", config)
+    monkeypatch.setattr(mcp_server, "_get_kg", lambda: kg)
+
+
+class TestToolsAcceptSurrogates:
+    """End-to-end: every MCP tool that takes user text must not crash on
+    lone surrogates. Failures here surface as ChromaDB UnicodeEncodeError."""
+
+    def test_add_drawer_content(self, monkeypatch, collection, config, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_add_drawer
+
+        result = tool_add_drawer(
+            wing="test",
+            room="surrogate",
+            content="drawer content with \udc95 surrogate",
+        )
+        assert result["success"] is True
+        assert result["drawer_id"].startswith("drawer_test_surrogate_")
+
+    def test_add_drawer_metadata(self, monkeypatch, collection, config, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_add_drawer
+
+        result = tool_add_drawer(
+            wing="test",
+            room="meta",
+            content="content here",
+            source_file="path/to/\udcadfile.txt",
+            added_by="user\udc95agent",
+        )
+        assert result["success"] is True
+
+    def test_check_duplicate(self, monkeypatch, collection, config, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_check_duplicate
+
+        result = tool_check_duplicate(content="exact\udc95match")
+        assert isinstance(result, dict)
+        assert "is_duplicate" in result
+
+    def test_search_query(self, monkeypatch, collection, config, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_search
+
+        result = tool_search(query="search\udc95term")
+        assert isinstance(result, dict)
+        assert "error" not in result or "results" in result
+
+    def test_update_drawer(self, monkeypatch, collection, config, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_add_drawer, tool_update_drawer
+
+        add_result = tool_add_drawer(wing="test", room="update", content="original content")
+        assert add_result["success"] is True
+
+        update_result = tool_update_drawer(
+            drawer_id=add_result["drawer_id"],
+            content="updated\udc95content",
+        )
+        assert update_result["success"] is True
+
+    def test_diary_write(self, monkeypatch, collection, config, kg):
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace.mcp_server import tool_diary_write
+
+        result = tool_diary_write(
+            agent_name="数数",
+            entry="今日工作\udc95完成了，修复了Chromadb crash",
+            topic="log",
+        )
+        assert result["success"] is True