Merge pull request #131 from igorls/test/expand-coverage-and-uv-migration

test: expand coverage from 20 to 92 tests, migrate to uv

tests/conftest.py

@@ -0,0 +1,169 @@
+"""
+conftest.py — Shared fixtures for MemPalace tests.
+
+Provides isolated palace and knowledge graph instances so tests never
+touch the user's real data or leak temp files on failure.
+
+HOME is redirected to a temp directory at module load time — before any
+mempalace imports — so that module-level initialisations (e.g.
+``_kg = KnowledgeGraph()`` in mcp_server) write to a throwaway location
+instead of the real user profile.
+"""
+
+import os
+import shutil
+import tempfile
+
+# ── Isolate HOME before any mempalace imports ──────────────────────────
+_original_env = {}
+_session_tmp = tempfile.mkdtemp(prefix="mempalace_session_")
+
+for _var in ("HOME", "USERPROFILE", "HOMEDRIVE", "HOMEPATH"):
+    _original_env[_var] = os.environ.get(_var)
+
+os.environ["HOME"] = _session_tmp
+os.environ["USERPROFILE"] = _session_tmp
+os.environ["HOMEDRIVE"] = os.path.splitdrive(_session_tmp)[0] or "C:"
+os.environ["HOMEPATH"] = os.path.splitdrive(_session_tmp)[1] or _session_tmp
+
+# Now it is safe to import mempalace modules that trigger initialisation.
+import chromadb  # noqa: E402
+import pytest  # noqa: E402
+
+from mempalace.config import MempalaceConfig  # noqa: E402
+from mempalace.knowledge_graph import KnowledgeGraph  # noqa: E402
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _isolate_home(tmp_path_factory):
+    """Ensure HOME points to a temp dir for the entire test session.
+
+    The env vars were already set at module level (above) so that
+    module-level initialisations are captured.  This fixture simply
+    restores the originals on teardown and cleans up the temp dir.
+    """
+    yield
+    for var, orig in _original_env.items():
+        if orig is None:
+            os.environ.pop(var, None)
+        else:
+            os.environ[var] = orig
+    shutil.rmtree(_session_tmp, ignore_errors=True)
+
+
+@pytest.fixture
+def tmp_dir():
+    """Create and auto-cleanup a temporary directory."""
+    d = tempfile.mkdtemp(prefix="mempalace_test_")
+    yield d
+    shutil.rmtree(d, ignore_errors=True)
+
+
+@pytest.fixture
+def palace_path(tmp_dir):
+    """Path to an empty palace directory inside tmp_dir."""
+    p = os.path.join(tmp_dir, "palace")
+    os.makedirs(p)
+    return p
+
+
+@pytest.fixture
+def config(tmp_dir, palace_path):
+    """A MempalaceConfig pointing at the temp palace."""
+    cfg_dir = os.path.join(tmp_dir, "config")
+    os.makedirs(cfg_dir)
+    import json
+
+    with open(os.path.join(cfg_dir, "config.json"), "w") as f:
+        json.dump({"palace_path": palace_path}, f)
+    return MempalaceConfig(config_dir=cfg_dir)
+
+
+@pytest.fixture
+def collection(palace_path):
+    """A ChromaDB collection pre-seeded in the temp palace."""
+    client = chromadb.PersistentClient(path=palace_path)
+    col = client.get_or_create_collection("mempalace_drawers")
+    return col
+
+
+@pytest.fixture
+def seeded_collection(collection):
+    """Collection with a handful of representative drawers."""
+    collection.add(
+        ids=[
+            "drawer_proj_backend_aaa",
+            "drawer_proj_backend_bbb",
+            "drawer_proj_frontend_ccc",
+            "drawer_notes_planning_ddd",
+        ],
+        documents=[
+            "The authentication module uses JWT tokens for session management. "
+            "Tokens expire after 24 hours. Refresh tokens are stored in HttpOnly cookies.",
+            "Database migrations are handled by Alembic. We use PostgreSQL 15 "
+            "with connection pooling via pgbouncer.",
+            "The React frontend uses TanStack Query for server state management. "
+            "All API calls go through a centralized fetch wrapper.",
+            "Sprint planning: migrate auth to passkeys by Q3. "
+            "Evaluate ChromaDB alternatives for vector search.",
+        ],
+        metadatas=[
+            {
+                "wing": "project",
+                "room": "backend",
+                "source_file": "auth.py",
+                "chunk_index": 0,
+                "added_by": "miner",
+                "filed_at": "2026-01-01T00:00:00",
+            },
+            {
+                "wing": "project",
+                "room": "backend",
+                "source_file": "db.py",
+                "chunk_index": 0,
+                "added_by": "miner",
+                "filed_at": "2026-01-02T00:00:00",
+            },
+            {
+                "wing": "project",
+                "room": "frontend",
+                "source_file": "App.tsx",
+                "chunk_index": 0,
+                "added_by": "miner",
+                "filed_at": "2026-01-03T00:00:00",
+            },
+            {
+                "wing": "notes",
+                "room": "planning",
+                "source_file": "sprint.md",
+                "chunk_index": 0,
+                "added_by": "miner",
+                "filed_at": "2026-01-04T00:00:00",
+            },
+        ],
+    )
+    return collection
+
+
+@pytest.fixture
+def kg(tmp_dir):
+    """An isolated KnowledgeGraph using a temp SQLite file."""
+    db_path = os.path.join(tmp_dir, "test_kg.sqlite3")
+    return KnowledgeGraph(db_path=db_path)
+
+
+@pytest.fixture
+def seeded_kg(kg):
+    """KnowledgeGraph pre-loaded with sample triples."""
+    kg.add_entity("Alice", entity_type="person")
+    kg.add_entity("Max", entity_type="person")
+    kg.add_entity("swimming", entity_type="activity")
+    kg.add_entity("chess", entity_type="activity")
+
+    kg.add_triple("Alice", "parent_of", "Max", valid_from="2015-04-01")
+    kg.add_triple("Max", "does", "swimming", valid_from="2025-01-01")
+    kg.add_triple("Max", "does", "chess", valid_from="2024-06-01")
+    kg.add_triple("Alice", "works_at", "Acme Corp", valid_from="2020-01-01", valid_to="2024-12-31")
+    kg.add_triple("Alice", "works_at", "NewCo", valid_from="2025-01-01")
+
+    return kg

tests/test_dialect.py

@@ -0,0 +1,157 @@
+"""
+test_dialect.py — Tests for the AAAK Dialect compression system.
+
+Covers plain text compression, entity detection, emotion detection,
+topic extraction, key sentence extraction, zettel encoding, and stats.
+"""
+
+from mempalace.dialect import Dialect
+
+
+class TestPlainTextCompression:
+    def test_compress_basic(self):
+        d = Dialect()
+        result = d.compress("We decided to use GraphQL instead of REST for the API layer.")
+        assert isinstance(result, str)
+        assert len(result) > 0
+        # AAAK format uses pipe-separated fields
+        assert "|" in result
+
+    def test_compress_with_metadata(self):
+        d = Dialect()
+        result = d.compress(
+            "Authentication now uses JWT tokens.",
+            metadata={"wing": "project", "room": "backend", "source_file": "auth.py"},
+        )
+        assert "project" in result
+        assert "backend" in result
+
+    def test_compress_produces_entity_codes(self):
+        d = Dialect(entities={"Alice": "ALC", "Bob": "BOB"})
+        result = d.compress("Alice told Bob about the new deployment strategy.")
+        assert "ALC" in result or "BOB" in result
+
+    def test_compress_empty_text(self):
+        d = Dialect()
+        result = d.compress("")
+        assert isinstance(result, str)
+
+
+class TestEntityDetection:
+    def test_known_entities(self):
+        d = Dialect(entities={"Alice": "ALC"})
+        found = d._detect_entities_in_text("Alice went to the store.")
+        assert "ALC" in found
+
+    def test_auto_code_unknown_entities(self):
+        d = Dialect()
+        found = d._detect_entities_in_text("I spoke with Bernardo about the project today.")
+        assert any(code for code in found if len(code) == 3)
+
+    def test_skip_names(self):
+        d = Dialect(entities={"Gandalf": "GAN"}, skip_names=["Gandalf"])
+        code = d.encode_entity("Gandalf")
+        assert code is None
+
+
+class TestEmotionDetection:
+    def test_detect_emotions(self):
+        d = Dialect()
+        emotions = d._detect_emotions("I'm really excited and happy about this breakthrough!")
+        assert len(emotions) > 0
+
+    def test_max_three_emotions(self):
+        d = Dialect()
+        text = "I feel scared, happy, angry, surprised, disgusted, and confused."
+        emotions = d._detect_emotions(text)
+        assert len(emotions) <= 3
+
+
+class TestTopicExtraction:
+    def test_extract_topics(self):
+        d = Dialect()
+        topics = d._extract_topics(
+            "The Python authentication server uses PostgreSQL for storage "
+            "and Redis for caching sessions."
+        )
+        assert len(topics) > 0
+        assert len(topics) <= 3
+
+    def test_boosts_technical_terms(self):
+        d = Dialect()
+        topics = d._extract_topics("GraphQL vs REST: we chose GraphQL for the new API endpoint.")
+        # "graphql" should appear since it's mentioned twice + capitalized
+        topic_lower = [t.lower() for t in topics]
+        assert "graphql" in topic_lower
+
+
+class TestKeySentenceExtraction:
+    def test_extract_key_sentence(self):
+        d = Dialect()
+        text = (
+            "The server runs on port 3000. "
+            "We decided to use PostgreSQL instead of MongoDB. "
+            "The config file needs updating."
+        )
+        key = d._extract_key_sentence(text)
+        assert "decided" in key.lower() or "instead" in key.lower()
+
+    def test_truncates_long_sentences(self):
+        d = Dialect()
+        text = "a " * 100  # very long
+        key = d._extract_key_sentence(text)
+        assert len(key) <= 55
+
+
+class TestCompressionStats:
+    def test_stats(self):
+        d = Dialect()
+        original = "We decided to use GraphQL instead of REST. " * 10
+        compressed = d.compress(original)
+        stats = d.compression_stats(original, compressed)
+        assert stats["ratio"] > 1
+        assert stats["original_chars"] > stats["compressed_chars"]
+
+    def test_count_tokens(self):
+        assert Dialect.count_tokens("hello world") == len("hello world") // 3
+
+
+class TestZettelEncoding:
+    def test_encode_zettel(self):
+        d = Dialect(entities={"Alice": "ALC"})
+        zettel = {
+            "id": "zettel-001",
+            "people": ["Alice"],
+            "topics": ["memory", "ai"],
+            "content": 'She said "I want to remember everything"',
+            "emotional_weight": 0.9,
+            "emotional_tone": ["joy"],
+            "origin_moment": False,
+            "sensitivity": "",
+            "notes": "",
+            "origin_label": "",
+            "title": "Test - Memory Discussion",
+        }
+        result = d.encode_zettel(zettel)
+        assert "ALC" in result
+        assert "memory" in result
+
+    def test_encode_tunnel(self):
+        d = Dialect()
+        tunnel = {"from": "zettel-001", "to": "zettel-002", "label": "follows: temporal"}
+        result = d.encode_tunnel(tunnel)
+        assert "T:" in result
+        assert "001" in result
+        assert "002" in result
+
+
+class TestDecode:
+    def test_decode_roundtrip(self):
+        d = Dialect()
+        encoded = (
+            '001|ALC+BOB|2025-01-01|test_title\nARC:journey\n001:ALC|memory_ai|"test quote"|0.9|joy'
+        )
+        decoded = d.decode(encoded)
+        assert decoded["header"]["file"] == "001"
+        assert decoded["arc"] == "journey"
+        assert len(decoded["zettels"]) == 1

tests/test_knowledge_graph.py

@@ -0,0 +1,122 @@
+"""
+test_knowledge_graph.py — Tests for the temporal knowledge graph.
+
+Covers: entity CRUD, triple CRUD, temporal queries, invalidation,
+timeline, stats, and edge cases (duplicate triples, ID collisions).
+"""
+
+
+class TestEntityOperations:
+    def test_add_entity(self, kg):
+        eid = kg.add_entity("Alice", entity_type="person")
+        assert eid == "alice"
+
+    def test_add_entity_normalizes_id(self, kg):
+        eid = kg.add_entity("Dr. Chen", entity_type="person")
+        assert eid == "dr._chen"
+
+    def test_add_entity_upsert(self, kg):
+        kg.add_entity("Alice", entity_type="person")
+        kg.add_entity("Alice", entity_type="engineer")
+        # Should not raise — INSERT OR REPLACE
+        stats = kg.stats()
+        assert stats["entities"] == 1
+
+
+class TestTripleOperations:
+    def test_add_triple_creates_entities(self, kg):
+        tid = kg.add_triple("Alice", "knows", "Bob")
+        assert tid.startswith("t_alice_knows_bob_")
+        stats = kg.stats()
+        assert stats["entities"] == 2  # auto-created
+
+    def test_add_triple_with_dates(self, kg):
+        tid = kg.add_triple("Max", "does", "swimming", valid_from="2025-01-01")
+        assert tid.startswith("t_max_does_swimming_")
+
+    def test_duplicate_triple_returns_existing_id(self, kg):
+        tid1 = kg.add_triple("Alice", "knows", "Bob")
+        tid2 = kg.add_triple("Alice", "knows", "Bob")
+        assert tid1 == tid2
+
+    def test_invalidated_triple_allows_re_add(self, kg):
+        tid1 = kg.add_triple("Alice", "works_at", "Acme")
+        kg.invalidate("Alice", "works_at", "Acme", ended="2025-01-01")
+        tid2 = kg.add_triple("Alice", "works_at", "Acme")
+        assert tid1 != tid2  # new triple since old one was closed
+
+
+class TestQueries:
+    def test_query_outgoing(self, seeded_kg):
+        results = seeded_kg.query_entity("Alice", direction="outgoing")
+        predicates = {r["predicate"] for r in results}
+        assert "parent_of" in predicates
+        assert "works_at" in predicates
+
+    def test_query_incoming(self, seeded_kg):
+        results = seeded_kg.query_entity("Max", direction="incoming")
+        assert any(r["subject"] == "Alice" and r["predicate"] == "parent_of" for r in results)
+
+    def test_query_both_directions(self, seeded_kg):
+        results = seeded_kg.query_entity("Max", direction="both")
+        directions = {r["direction"] for r in results}
+        assert "outgoing" in directions
+        assert "incoming" in directions
+
+    def test_query_as_of_filters_expired(self, seeded_kg):
+        results = seeded_kg.query_entity("Alice", as_of="2023-06-01", direction="outgoing")
+        employers = [r["object"] for r in results if r["predicate"] == "works_at"]
+        assert "Acme Corp" in employers
+        assert "NewCo" not in employers
+
+    def test_query_as_of_shows_current(self, seeded_kg):
+        results = seeded_kg.query_entity("Alice", as_of="2025-06-01", direction="outgoing")
+        employers = [r["object"] for r in results if r["predicate"] == "works_at"]
+        assert "NewCo" in employers
+        assert "Acme Corp" not in employers
+
+    def test_query_relationship(self, seeded_kg):
+        results = seeded_kg.query_relationship("does")
+        assert len(results) == 2  # swimming + chess
+
+
+class TestInvalidation:
+    def test_invalidate_sets_valid_to(self, seeded_kg):
+        seeded_kg.invalidate("Max", "does", "chess", ended="2026-01-01")
+        results = seeded_kg.query_entity("Max", direction="outgoing")
+        chess = [r for r in results if r["object"] == "chess"]
+        assert len(chess) == 1
+        assert chess[0]["valid_to"] == "2026-01-01"
+        assert chess[0]["current"] is False
+
+
+class TestTimeline:
+    def test_timeline_all(self, seeded_kg):
+        tl = seeded_kg.timeline()
+        assert len(tl) >= 4
+
+    def test_timeline_entity(self, seeded_kg):
+        tl = seeded_kg.timeline("Max")
+        subjects_and_objects = {t["subject"] for t in tl} | {t["object"] for t in tl}
+        assert "Max" in subjects_and_objects
+
+    def test_timeline_global_has_limit(self, kg):
+        # Add > 100 triples
+        for i in range(105):
+            kg.add_triple(f"entity_{i}", "relates_to", f"entity_{i + 1}")
+        tl = kg.timeline()
+        assert len(tl) == 100  # LIMIT 100
+
+
+class TestStats:
+    def test_stats_empty(self, kg):
+        stats = kg.stats()
+        assert stats["entities"] == 0
+        assert stats["triples"] == 0
+
+    def test_stats_seeded(self, seeded_kg):
+        stats = seeded_kg.stats()
+        assert stats["entities"] >= 4
+        assert stats["triples"] == 5
+        assert stats["current_facts"] == 4  # 1 expired (Acme Corp)
+        assert stats["expired_facts"] == 1

tests/test_mcp_server.py

@@ -0,0 +1,334 @@
+"""
+test_mcp_server.py — Tests for the MCP server tool handlers and dispatch.
+
+Tests each tool handler directly (unit-level) and the handle_request
+dispatch layer (integration-level). Uses isolated palace + KG fixtures
+via monkeypatch to avoid touching real data.
+"""
+
+import json
+
+
+def _patch_mcp_server(monkeypatch, config, palace_path, kg):
+    """Patch the mcp_server module globals to use test fixtures."""
+    from mempalace import mcp_server
+
+    monkeypatch.setattr(mcp_server, "_config", config)
+    monkeypatch.setattr(mcp_server, "_kg", kg)
+
+
+def _get_collection(palace_path, create=False):
+    """Helper to get collection from test palace."""
+    import chromadb
+
+    client = chromadb.PersistentClient(path=palace_path)
+    if create:
+        return client.get_or_create_collection("mempalace_drawers")
+    return client.get_collection("mempalace_drawers")
+
+
+# ── Protocol Layer ──────────────────────────────────────────────────────
+
+
+class TestHandleRequest:
+    def test_initialize(self):
+        from mempalace.mcp_server import handle_request
+
+        resp = handle_request({"method": "initialize", "id": 1, "params": {}})
+        assert resp["result"]["serverInfo"]["name"] == "mempalace"
+        assert resp["id"] == 1
+
+    def test_notifications_initialized_returns_none(self):
+        from mempalace.mcp_server import handle_request
+
+        resp = handle_request({"method": "notifications/initialized", "id": None, "params": {}})
+        assert resp is None
+
+    def test_tools_list(self):
+        from mempalace.mcp_server import handle_request
+
+        resp = handle_request({"method": "tools/list", "id": 2, "params": {}})
+        tools = resp["result"]["tools"]
+        names = {t["name"] for t in tools}
+        assert "mempalace_status" in names
+        assert "mempalace_search" in names
+        assert "mempalace_add_drawer" in names
+        assert "mempalace_kg_add" in names
+
+    def test_unknown_tool(self):
+        from mempalace.mcp_server import handle_request
+
+        resp = handle_request(
+            {
+                "method": "tools/call",
+                "id": 3,
+                "params": {"name": "nonexistent_tool", "arguments": {}},
+            }
+        )
+        assert resp["error"]["code"] == -32601
+
+    def test_unknown_method(self):
+        from mempalace.mcp_server import handle_request
+
+        resp = handle_request({"method": "unknown/method", "id": 4, "params": {}})
+        assert resp["error"]["code"] == -32601
+
+    def test_tools_call_dispatches(self, monkeypatch, config, palace_path, seeded_kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, seeded_kg)
+        from mempalace.mcp_server import handle_request
+
+        # Create a collection so status works
+        _get_collection(palace_path, create=True)
+
+        resp = handle_request(
+            {
+                "method": "tools/call",
+                "id": 5,
+                "params": {"name": "mempalace_status", "arguments": {}},
+            }
+        )
+        assert "result" in resp
+        content = json.loads(resp["result"]["content"][0]["text"])
+        assert "total_drawers" in content
+
+
+# ── Read Tools ──────────────────────────────────────────────────────────
+
+
+class TestReadTools:
+    def test_status_empty_palace(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        _get_collection(palace_path, create=True)
+        from mempalace.mcp_server import tool_status
+
+        result = tool_status()
+        assert result["total_drawers"] == 0
+        assert result["wings"] == {}
+
+    def test_status_with_data(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_status
+
+        result = tool_status()
+        assert result["total_drawers"] == 4
+        assert "project" in result["wings"]
+        assert "notes" in result["wings"]
+
+    def test_list_wings(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_list_wings
+
+        result = tool_list_wings()
+        assert result["wings"]["project"] == 3
+        assert result["wings"]["notes"] == 1
+
+    def test_list_rooms_all(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_list_rooms
+
+        result = tool_list_rooms()
+        assert "backend" in result["rooms"]
+        assert "frontend" in result["rooms"]
+        assert "planning" in result["rooms"]
+
+    def test_list_rooms_filtered(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_list_rooms
+
+        result = tool_list_rooms(wing="project")
+        assert "backend" in result["rooms"]
+        assert "planning" not in result["rooms"]
+
+    def test_get_taxonomy(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_get_taxonomy
+
+        result = tool_get_taxonomy()
+        assert result["taxonomy"]["project"]["backend"] == 2
+        assert result["taxonomy"]["project"]["frontend"] == 1
+        assert result["taxonomy"]["notes"]["planning"] == 1
+
+    def test_no_palace_returns_error(self, monkeypatch, config, kg):
+        _patch_mcp_server(monkeypatch, config, "/nonexistent/path", kg)
+        from mempalace.mcp_server import tool_status
+
+        result = tool_status()
+        assert "error" in result
+
+
+# ── Search Tool ─────────────────────────────────────────────────────────
+
+
+class TestSearchTool:
+    def test_search_basic(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_search
+
+        result = tool_search(query="JWT authentication tokens")
+        assert "results" in result
+        assert len(result["results"]) > 0
+        # Top result should be the auth drawer
+        top = result["results"][0]
+        assert "JWT" in top["text"] or "authentication" in top["text"].lower()
+
+    def test_search_with_wing_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_search
+
+        result = tool_search(query="planning", wing="notes")
+        assert all(r["wing"] == "notes" for r in result["results"])
+
+    def test_search_with_room_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_search
+
+        result = tool_search(query="database", room="backend")
+        assert all(r["room"] == "backend" for r in result["results"])
+
+
+# ── Write Tools ─────────────────────────────────────────────────────────
+
+
+class TestWriteTools:
+    def test_add_drawer(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        _get_collection(palace_path, create=True)
+        from mempalace.mcp_server import tool_add_drawer
+
+        result = tool_add_drawer(
+            wing="test_wing",
+            room="test_room",
+            content="This is a test memory about Python decorators and metaclasses.",
+        )
+        assert result["success"] is True
+        assert result["wing"] == "test_wing"
+        assert result["room"] == "test_room"
+        assert result["drawer_id"].startswith("drawer_test_wing_test_room_")
+
+    def test_add_drawer_duplicate_detection(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        _get_collection(palace_path, create=True)
+        from mempalace.mcp_server import tool_add_drawer
+
+        content = "This is a unique test memory about Rust ownership and borrowing."
+        result1 = tool_add_drawer(wing="w", room="r", content=content)
+        assert result1["success"] is True
+
+        result2 = tool_add_drawer(wing="w", room="r", content=content)
+        assert result2["success"] is False
+        assert result2["reason"] == "duplicate"
+
+    def test_delete_drawer(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_delete_drawer
+
+        result = tool_delete_drawer("drawer_proj_backend_aaa")
+        assert result["success"] is True
+        assert seeded_collection.count() == 3
+
+    def test_delete_drawer_not_found(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_delete_drawer
+
+        result = tool_delete_drawer("nonexistent_drawer")
+        assert result["success"] is False
+
+    def test_check_duplicate(self, monkeypatch, config, palace_path, seeded_collection, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_check_duplicate
+
+        # Exact match text from seeded_collection should be flagged
+        result = tool_check_duplicate(
+            "The authentication module uses JWT tokens for session management. "
+            "Tokens expire after 24 hours. Refresh tokens are stored in HttpOnly cookies.",
+            threshold=0.5,
+        )
+        assert result["is_duplicate"] is True
+
+        # Unrelated content should not be flagged
+        result = tool_check_duplicate(
+            "Black holes emit Hawking radiation at the event horizon.",
+            threshold=0.99,
+        )
+        assert result["is_duplicate"] is False
+
+
+# ── KG Tools ────────────────────────────────────────────────────────────
+
+
+class TestKGTools:
+    def test_kg_add(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        from mempalace.mcp_server import tool_kg_add
+
+        result = tool_kg_add(
+            subject="Alice",
+            predicate="likes",
+            object="coffee",
+            valid_from="2025-01-01",
+        )
+        assert result["success"] is True
+
+    def test_kg_query(self, monkeypatch, config, palace_path, seeded_kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, seeded_kg)
+        from mempalace.mcp_server import tool_kg_query
+
+        result = tool_kg_query(entity="Max")
+        assert result["count"] > 0
+
+    def test_kg_invalidate(self, monkeypatch, config, palace_path, seeded_kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, seeded_kg)
+        from mempalace.mcp_server import tool_kg_invalidate
+
+        result = tool_kg_invalidate(
+            subject="Max",
+            predicate="does",
+            object="chess",
+            ended="2026-03-01",
+        )
+        assert result["success"] is True
+
+    def test_kg_timeline(self, monkeypatch, config, palace_path, seeded_kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, seeded_kg)
+        from mempalace.mcp_server import tool_kg_timeline
+
+        result = tool_kg_timeline(entity="Alice")
+        assert result["count"] > 0
+
+    def test_kg_stats(self, monkeypatch, config, palace_path, seeded_kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, seeded_kg)
+        from mempalace.mcp_server import tool_kg_stats
+
+        result = tool_kg_stats()
+        assert result["entities"] >= 4
+
+
+# ── Diary Tools ─────────────────────────────────────────────────────────
+
+
+class TestDiaryTools:
+    def test_diary_write_and_read(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        _get_collection(palace_path, create=True)
+        from mempalace.mcp_server import tool_diary_write, tool_diary_read
+
+        w = tool_diary_write(
+            agent_name="TestAgent",
+            entry="Today we discussed authentication patterns.",
+            topic="architecture",
+        )
+        assert w["success"] is True
+        assert w["agent"] == "TestAgent"
+
+        r = tool_diary_read(agent_name="TestAgent")
+        assert r["total"] == 1
+        assert r["entries"][0]["topic"] == "architecture"
+        assert "authentication" in r["entries"][0]["content"]
+
+    def test_diary_read_empty(self, monkeypatch, config, palace_path, kg):
+        _patch_mcp_server(monkeypatch, config, palace_path, kg)
+        _get_collection(palace_path, create=True)
+        from mempalace.mcp_server import tool_diary_read
+
+        r = tool_diary_read(agent_name="Nobody")
+        assert r["entries"] == []

tests/test_searcher.py

@@ -0,0 +1,45 @@
+"""
+test_searcher.py — Tests for the programmatic search_memories API.
+
+Tests the library-facing search interface (not the CLI print variant).
+"""
+
+from mempalace.searcher import search_memories
+
+
+class TestSearchMemories:
+    def test_basic_search(self, palace_path, seeded_collection):
+        result = search_memories("JWT authentication", palace_path)
+        assert "results" in result
+        assert len(result["results"]) > 0
+        assert result["query"] == "JWT authentication"
+
+    def test_wing_filter(self, palace_path, seeded_collection):
+        result = search_memories("planning", palace_path, wing="notes")
+        assert all(r["wing"] == "notes" for r in result["results"])
+
+    def test_room_filter(self, palace_path, seeded_collection):
+        result = search_memories("database", palace_path, room="backend")
+        assert all(r["room"] == "backend" for r in result["results"])
+
+    def test_wing_and_room_filter(self, palace_path, seeded_collection):
+        result = search_memories("code", palace_path, wing="project", room="frontend")
+        assert all(r["wing"] == "project" and r["room"] == "frontend" for r in result["results"])
+
+    def test_n_results_limit(self, palace_path, seeded_collection):
+        result = search_memories("code", palace_path, n_results=2)
+        assert len(result["results"]) <= 2
+
+    def test_no_palace_returns_error(self):
+        result = search_memories("anything", "/nonexistent/path")
+        assert "error" in result
+
+    def test_result_fields(self, palace_path, seeded_collection):
+        result = search_memories("authentication", palace_path)
+        hit = result["results"][0]
+        assert "text" in hit
+        assert "wing" in hit
+        assert "room" in hit
+        assert "source_file" in hit
+        assert "similarity" in hit
+        assert isinstance(hit["similarity"], float)