diff --git a/ROADMAP.md b/ROADMAP.md
index 0546ed4a..c155aa1c 100644
--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -31,7 +31,7 @@
 
 ## 4. Wiki Restructuring
 
-- [ ] **Unify wiki system** — merge `file_wiki.py`, `user_wiki.py`, `agent_wiki.py` into one `WikiManager` with layer-based separation. Current 3 separate classes duplicate logic. Goal: single FTS5 index, shared sync logic, configurable per-layer behavior.
+- [x] **Unify wiki system** — merge `file_wiki.py`, `user_wiki.py`, `agent_wiki.py` into one `WikiManager` with layer-based separation. Current 3 separate classes duplicate logic. Goal: single FTS5 index, shared sync logic, configurable per-layer behavior.
 
 ## 4. Bug Fixes & Cleanup
 
@@ -161,5 +161,5 @@
 
 ---
 
-**Completed:** 35/65 items
-**Last updated:** 2026-07-03
+**Completed:** 36/65 items
+**Last updated:** 2026-07-04
diff --git a/features/backup.py b/features/backup.py
index 96552a55..add62147 100644
--- a/features/backup.py
+++ b/features/backup.py
@@ -9,6 +9,7 @@
 from typing import Any, Optional
 
 from config import config
+from shared.path_safety import safe_resolve
 
 
 class BackupManager:
@@ -50,6 +51,7 @@ async def restore(self, backup_name: str) -> dict[str, Any]:
 
         restored = []
         for db_file in manifest.get("files", []):
+            safe_resolve(self.base_dir, db_file)  # raises ValueError if traversal
             backup_file = src / db_file
             if backup_file.exists():
                 shutil.copy2(backup_file, self.base_dir / db_file)
diff --git a/features/backup_cron.py b/features/backup_cron.py
index 4647bc84..6318a721 100644
--- a/features/backup_cron.py
+++ b/features/backup_cron.py
@@ -13,6 +13,7 @@
 from typing import Any, Optional
 
 from config import config
+from shared.path_safety import safe_resolve
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +37,9 @@ def __init__(self, base_dir: Optional[str] = None):
     def _load_state(self):
         if self._state_file.exists():
             try:
-                state = json.loads(self._state_file.read_text(encoding="utf-8"))
+                from shared.saga_crypto import read_state_legacy_or_encrypted
+
+                state = read_state_legacy_or_encrypted(self._state_file)
                 self._last_backup = state.get("last_backup", 0.0)
                 self._last_wiki_sync = state.get("last_wiki_sync", 0.0)
             except Exception:
@@ -131,10 +134,10 @@ def _cleanup_old(self):
     def _sync_wiki(self):
         """Synchronize wiki files with disk."""
         try:
-            from wiki.file_wiki import FileWiki
+            from wiki.manager import WikiManager
 
             for layer in ["user", "agent"]:
-                fw = FileWiki(layer=layer)
+                fw = WikiManager(layer=layer)
                 raw = fw.reindex_all()
                 result: dict[str, Any] = asyncio.run(raw) if asyncio.iscoroutine(raw) else raw
                 if isinstance(result, dict) and result.get("indexed", 0) > 0:
@@ -162,6 +165,7 @@ def restore(self, backup_name: str) -> dict[str, Any]:
 
         restored = []
         for db_file in manifest.get("files", []):
+            safe_resolve(self.base_dir, db_file)  # raises ValueError if traversal
             if db_file.endswith("/"):
                 # Restore wiki directory
                 src_wiki = src / db_file
diff --git a/features/dashboard.py b/features/dashboard.py
index 8595e04a..4efde35d 100644
--- a/features/dashboard.py
+++ b/features/dashboard.py
@@ -171,10 +171,10 @@ def __init__(self, mm=None, data_dir: Optional[str] = None):
 
     async def get_stats(self, user_id: str = "default") -> dict[str, Any]:
         from graph.epistemic import EpistemicGraph
-        from wiki.file_wiki import FileWiki
+        from wiki.manager import WikiManager
 
-        uw = FileWiki(layer="user")
-        aw = FileWiki(layer="agent")
+        uw = WikiManager(layer="user")
+        aw = WikiManager(layer="agent")
         ug = EpistemicGraph(layer="user")
 
         um = self.mm.user_memory(user_id)
diff --git a/features/import_export.py b/features/import_export.py
index 48ea1fd2..98114c9a 100644
--- a/features/import_export.py
+++ b/features/import_export.py
@@ -8,6 +8,7 @@
 from typing import Any, Optional
 
 from shared.connection import AsyncConnectionManager, connection_manager
+from shared.path_safety import safe_resolve
 
 
 class ImportExport:
@@ -58,6 +59,7 @@ async def export_user(self, user_id: str) -> str:
         return str(filepath)
 
     async def import_user(self, filepath: str, target_user_id: Optional[str] = None) -> dict[str, int]:
+        safe_resolve(self.export_dir, filepath)  # raises ValueError if traversal
         data = json.loads(Path(filepath).read_text(encoding="utf-8"))
         user_id = target_user_id or data.get("user_id", "default")
         imported = {"core_memory": 0, "episodes": 0}
diff --git a/lifecycle/emotion_trigger.py b/lifecycle/emotion_trigger.py
index c2ffd381..90b8b5f2 100644
--- a/lifecycle/emotion_trigger.py
+++ b/lifecycle/emotion_trigger.py
@@ -158,47 +158,70 @@ class EmotionTrigger:
     def should_save(self, message: str, emotional_state: Optional[dict] = None, state_delta: Optional[dict] = None) -> tuple[bool, str, float]:
         msg_lower = message.lower()
 
-        # 1. Phrase patterns (high priority)
-        all_patterns = PHRASE_PATTERNS + PHRASE_PATTERNS_EN
-        for pattern, emotion, weight in all_patterns:
+        result = self._check_phrase_patterns(msg_lower)
+        if result:
+            return result
+
+        result = self._check_emotion_markers(msg_lower)
+        if result:
+            return result
+
+        result = self._check_emoji(message)
+        if result:
+            return result
+
+        result = self._check_emotional_state(emotional_state)
+        if result:
+            return result
+
+        result = self._check_state_shift(state_delta)
+        if result:
+            return result
+
+        if len(message) > 300:
+            return True, "long_message", 0.3
+
+        if message.count("?") >= 3:
+            return True, "complex_question", 0.4
+
+        if message.count("!") >= 2:
+            return True, "exclamation", 0.3
+
+        return False, "", 0.0
+
+    def _check_phrase_patterns(self, msg_lower: str) -> tuple[bool, str, float] | None:
+        for pattern, emotion, weight in PHRASE_PATTERNS + PHRASE_PATTERNS_EN:
             if re.search(pattern, msg_lower):
                 return True, f"emotion_{emotion}", weight
+        return None
 
-        # 2. Emotion markers
+    def _check_emotion_markers(self, msg_lower: str) -> tuple[bool, str, float] | None:
+        high_weight = ("love", "fear", "anger")
         for emotion, markers in EMOTION_MARKERS.items():
             for marker in markers:
                 if marker in msg_lower:
-                    weight = 0.7 if emotion in ("love", "fear", "anger") else 0.5
+                    weight = 0.7 if emotion in high_weight else 0.5
                     return True, f"emotion_{emotion}", weight
+        return None
 
-        # 3. Emoji
+    def _check_emoji(self, message: str) -> tuple[bool, str, float] | None:
+        high_weight = ("love", "fear", "anger")
         for emotion, emojis in EMOJI_MARKERS.items():
             for emoji in emojis:
                 if emoji in message:
-                    weight = 0.7 if emotion in ("love", "fear", "anger") else 0.5
+                    weight = 0.7 if emotion in high_weight else 0.5
                     return True, f"emotion_{emotion}", weight
+        return None
 
-        # 4. Emotional state from context
+    def _check_emotional_state(self, emotional_state: Optional[dict]) -> tuple[bool, str, float] | None:
         if emotional_state:
             if emotional_state.get("joy", 0) > 0.8 or emotional_state.get("interest", 0) > 0.8:
                 return True, "high_emotion", 0.6
+        return None
 
-        # 5. State shift
+    def _check_state_shift(self, state_delta: Optional[dict]) -> tuple[bool, str, float] | None:
         if state_delta:
             for key, delta in state_delta.items():
                 if abs(delta) > STATE_SHIFT_THRESHOLD:
                     return True, f"state_shift_{key}", 0.4
-
-        # 6. Long message
-        if len(message) > 300:
-            return True, "long_message", 0.3
-
-        # 7. Multiple questions
-        if message.count("?") >= 3:
-            return True, "complex_question", 0.4
-
-        # 8. Exclamation marks
-        if message.count("!") >= 2:
-            return True, "exclamation", 0.3
-
-        return False, "", 0.0
+        return None
diff --git a/mcp_server/server.py b/mcp_server/server.py
index ca7a0ddf..f84aab18 100644
--- a/mcp_server/server.py
+++ b/mcp_server/server.py
@@ -35,15 +35,15 @@
 from rag.multi_source import MultiSourceRAG
 from shared.cache import MemoryCache
 from shared.read_only import read_only_replica
-from wiki.file_wiki import FileWiki
+from wiki.manager import WikiManager
 
 
 class AppContext:
     def __init__(self):
         self.cache = MemoryCache()
         self.mm = MemoryManager(cache=self.cache)
-        self.user_wiki = FileWiki(layer="user")
-        self.agent_wiki = FileWiki(layer="agent")
+        self.user_wiki = WikiManager(layer="user")
+        self.agent_wiki = WikiManager(layer="agent")
         self.user_rag = RAGEngine(layer="user")
         self.agent_rag = RAGEngine(layer="agent")
         self.user_multi = MultiSourceRAG(self.user_rag, self.user_wiki)
diff --git a/shared/path_safety.py b/shared/path_safety.py
new file mode 100644
index 00000000..3e7ff000
--- /dev/null
+++ b/shared/path_safety.py
@@ -0,0 +1,18 @@
+"""Path traversal prevention — shared guard for all file-accepting functions."""
+
+import os
+from pathlib import Path
+
+
+def safe_resolve(base: Path, user_input: str) -> Path:
+    """Resolve user_input relative to base, raising ValueError if it escapes.
+
+    Checks both the base-relative resolution and the real path (follows symlinks).
+    """
+    base_resolved = base.resolve()
+    target = (base / user_input).resolve()
+
+    if not str(target).startswith(str(base_resolved) + os.sep) and target != base_resolved:
+        raise ValueError(f"Path escapes base directory: {user_input!r}")
+
+    return target
diff --git a/shared/saga.py b/shared/saga.py
index 7f801ab2..2f7dc94c 100644
--- a/shared/saga.py
+++ b/shared/saga.py
@@ -130,16 +130,15 @@ def _save_state(self):
 
     def _load_state(self, saga_id: str) -> dict | None:
         """Load state from disk (supports encrypted and legacy plain JSON)."""
+        from shared.saga_crypto import read_state_legacy_or_encrypted
+
         state_file = SAGA_DIR / (saga_id + ".json")
-        if state_file.exists():
-            try:
-                blob = state_file.read_bytes()
-                if _HAS_ENCRYPTION and is_encrypted_blob(state_file):
-                    return decrypt_json(blob)
-                return json.loads(blob.decode("utf-8"))
-            except Exception:
-                pass
-        return None
+        if not state_file.exists():
+            return None
+        try:
+            return read_state_legacy_or_encrypted(state_file)
+        except Exception:
+            return None
 
     def _cleanup_state(self):
         """Delete state file after completion."""
@@ -339,23 +338,32 @@ async def _compensate(self, failed_step: int) -> None:
                 continue
 
             if isinstance(step.action, Saga):
-                inner = step.action
-                for j in range(len(inner._steps) - 1, -1, -1):
-                    inner_step = inner._steps[j]
-                    if inner_step.status == SagaStatus.COMPLETED and inner_step.compensation:
-                        try:
-                            await inner_step.compensation(inner_step.data)
-                            logger.info("Saga '%s' compensated inner step '%s'" % (self.name, inner_step.name))
-                        except Exception as e:
-                            logger.error("Saga '%s' inner compensation failed for '%s': %s" % (self.name, inner_step.name, e))
+                await self._compensate_inner_saga(step.action)
             elif step.compensation:
+                await self._compensate_step(step)
+
+        self._status = SagaStatus.COMPENSATED
+
+    async def _compensate_inner_saga(self, inner: "Saga") -> None:
+        """Compensate all completed steps of a nested saga in reverse order."""
+        for j in range(len(inner._steps) - 1, -1, -1):
+            inner_step = inner._steps[j]
+            if inner_step.status == SagaStatus.COMPLETED and inner_step.compensation:
                 try:
-                    await step.compensation(step.data)
-                    logger.info("Saga '%s' compensated step '%s'" % (self.name, step.name))
+                    await inner_step.compensation(inner_step.data)
+                    logger.info("Saga '%s' compensated inner step '%s'" % (self.name, inner_step.name))
                 except Exception as e:
-                    logger.error("Saga '%s' compensation failed for '%s': %s" % (self.name, step.name, e))
+                    logger.error("Saga '%s' inner compensation failed for '%s': %s" % (self.name, inner_step.name, e))
 
-        self._status = SagaStatus.COMPENSATED
+    async def _compensate_step(self, step: SagaStep) -> None:
+        """Run compensation for a single step, logging success or failure."""
+        if not step.compensation:
+            return
+        try:
+            await step.compensation(step.data)
+            logger.info("Saga '%s' compensated step '%s'" % (self.name, step.name))
+        except Exception as e:
+            logger.error("Saga '%s' compensation failed for '%s': %s" % (self.name, step.name, e))
 
     def get_state(self) -> dict:
         return {
diff --git a/tests/test_all.py b/tests/test_all.py
index a12c88d6..c76fe81b 100644
--- a/tests/test_all.py
+++ b/tests/test_all.py
@@ -91,10 +91,10 @@ async def t():
 
 
 def test_user_wiki():
-    from wiki.file_wiki import FileWiki
+    from wiki.manager import WikiManager
 
     async def t():
-        w = FileWiki(layer="user")
+        w = WikiManager(layer="user")
         path = await w.add("work_notes", "Day 1", "Started project")
         assert path is not None
         results = await w.search("project")
diff --git a/tests/test_features/test_backup_path_safety.py b/tests/test_features/test_backup_path_safety.py
new file mode 100644
index 00000000..3b274b19
--- /dev/null
+++ b/tests/test_features/test_backup_path_safety.py
@@ -0,0 +1,63 @@
+"""Tests for backup path traversal prevention."""
+
+import json
+
+import pytest
+
+from features.backup import BackupManager
+
+
+@pytest.fixture
+def bm(tmp_path):
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    return BackupManager(base_dir=str(data_dir))
+
+
+def test_restore_rejects_traversal_in_manifest(bm):
+    """Crafted manifest with ../../ in filenames should be rejected."""
+    # Create a malicious backup directory with crafted manifest
+    backup_dir = bm.backup_dir / "malicious"
+    backup_dir.mkdir()
+    manifest = {
+        "files": ["../../etc/crontab", "memory.db"],
+        "created_at": "2026-01-01T00:00:00",
+    }
+    (backup_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+
+    import asyncio
+
+    with pytest.raises(ValueError, match="escapes base directory"):
+        asyncio.run(bm.restore("malicious"))
+
+
+def test_restore_rejects_absolute_path(bm):
+    """Manifest with absolute path should be rejected."""
+    backup_dir = bm.backup_dir / "absolute"
+    backup_dir.mkdir()
+    manifest = {
+        "files": ["/etc/passwd"],
+        "created_at": "2026-01-01T00:00:00",
+    }
+    (backup_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+
+    import asyncio
+
+    with pytest.raises(ValueError, match="escapes base directory"):
+        asyncio.run(bm.restore("absolute"))
+
+
+def test_restore_accepts_valid_files(bm):
+    """Valid manifest with normal files should work."""
+    # Create a real db file
+    db_file = bm.base_dir / "memory.db"
+    db_file.write_bytes(b"fake db")
+
+    # Create a valid backup
+    import asyncio
+
+    asyncio.run(bm.backup("test_backup"))
+
+    # Restore should work
+    result = asyncio.run(bm.restore("test_backup"))
+    assert "restored" in result
diff --git a/tests/test_features/test_import_export_path_safety.py b/tests/test_features/test_import_export_path_safety.py
new file mode 100644
index 00000000..ce1d4bb6
--- /dev/null
+++ b/tests/test_features/test_import_export_path_safety.py
@@ -0,0 +1,64 @@
+"""Tests for import_export path traversal prevention."""
+
+import json
+
+import pytest
+
+from features.import_export import ImportExport
+
+
+@pytest.fixture
+def ie(tmp_path):
+    """Create ImportExport with controlled export_dir."""
+    export_dir = tmp_path / "exports"
+    export_dir.mkdir()
+
+    class FakeCM:
+        def __init__(self, base):
+            self._base = base
+
+        @property
+        def base_dir(self):
+            return self._base
+
+    obj = ImportExport.__new__(ImportExport)
+    obj._cm = FakeCM(tmp_path)
+    obj.export_dir = export_dir
+    return obj
+
+
+def test_import_rejects_traversal(ie):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        import asyncio
+
+        asyncio.run(ie.import_user("../../etc/passwd"))
+
+
+def test_import_rejects_absolute_path(ie):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        import asyncio
+
+        asyncio.run(ie.import_user("/etc/passwd"))
+
+
+def test_import_accepts_valid_file(ie):
+    """Valid file in export_dir should be accepted (may fail on DB, but path check passes)."""
+    export_file = ie.export_dir / "valid_export.json"
+    export_file.write_text(
+        json.dumps(
+            {
+                "user_id": "test_user",
+                "core_memory": [],
+                "episodes": [],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    import asyncio
+
+    # This will fail at the DB level (FakeCM has no get), but path validation passes
+    try:
+        asyncio.run(ie.import_user(str(export_file)))
+    except AttributeError:
+        pass  # Expected — FakeCM doesn't have get()
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 43116d31..c5b19d14 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -361,10 +361,10 @@ async def test_agent_hooks():
 @pytest.mark.asyncio
 async def test_file_wiki():
     from shared.connection import connection_manager
-    from wiki.file_wiki import FileWiki
+    from wiki.manager import WikiManager
 
     tmpdir = tempfile.mkdtemp()
-    fw = FileWiki(layer="user", base_dir=tmpdir, cm=connection_manager)
+    fw = WikiManager(layer="user", base_dir=tmpdir, cm=connection_manager)
     await fw.init_db()
 
     path = await fw.add("diary", "Day 1", "Started project", tags=["work"])
@@ -381,25 +381,25 @@ async def test_file_wiki():
 @pytest.mark.asyncio
 async def test_user_wiki():
     from shared.connection import connection_manager
-    from wiki.user_wiki import UserWiki
+    from wiki.manager import WikiManager
 
-    uw = UserWiki(cm=connection_manager)
+    uw = WikiManager(layer="user", cm=connection_manager)
     await uw.init_db()
 
-    entry_id = await uw.add("test_integ", "diary", "Day 1", "Content", ["work"])
-    assert entry_id > 0
+    path = await uw.add("diary", "Day 1", "Content", ["work"])
+    assert path is not None
 
 
 @pytest.mark.asyncio
 async def test_agent_wiki():
     from shared.connection import connection_manager
-    from wiki.agent_wiki import AgentWiki
+    from wiki.manager import WikiManager
 
-    aw = AgentWiki(cm=connection_manager)
+    aw = WikiManager(layer="agent", cm=connection_manager)
     await aw.init_db()
 
-    entry_id = await aw.add("test_integ", "decision_log", "Choice A", "Chose A", [])
-    assert entry_id > 0
+    path = await aw.add("decision_log", "Choice A", "Chose A", [])
+    assert path is not None
 
 
 # ═══════════════════════════════════════════════════════════════
diff --git a/tests/test_shared/test_path_safety.py b/tests/test_shared/test_path_safety.py
new file mode 100644
index 00000000..f13d92e1
--- /dev/null
+++ b/tests/test_shared/test_path_safety.py
@@ -0,0 +1,46 @@
+"""Tests for shared.path_safety — path traversal prevention."""
+
+import pytest
+
+from shared.path_safety import safe_resolve
+
+
+def test_safe_resolve_within_base(tmp_path):
+    result = safe_resolve(tmp_path, "subdir/file.txt")
+    assert result == tmp_path / "subdir" / "file.txt"
+
+
+def test_safe_resolve_realpath_within_base(tmp_path):
+    sub = tmp_path / "allowed"
+    sub.mkdir()
+    result = safe_resolve(tmp_path, "allowed")
+    assert result.resolve() == sub.resolve()
+
+
+def test_safe_resolve_traversal_raises(tmp_path):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        safe_resolve(tmp_path, "../../etc/passwd")
+
+
+def test_safe_resolve_absolute_escape_raises(tmp_path):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        safe_resolve(tmp_path, "/etc/passwd")
+
+
+def test_safe_resolve_symlink_escape_raises(tmp_path):
+    link = tmp_path / "escape"
+    link.symlink_to("/etc")
+    with pytest.raises(ValueError, match="escapes base directory"):
+        safe_resolve(tmp_path, "escape/passwd")
+
+
+def test_safe_resolve_dot_dot_within_base(tmp_path):
+    sub = tmp_path / "a" / "b"
+    sub.mkdir(parents=True)
+    result = safe_resolve(tmp_path, "a/b/../b/file.txt")
+    assert result.resolve() == (tmp_path / "a" / "b" / "file.txt").resolve()
+
+
+def test_safe_resolve_empty_string(tmp_path):
+    result = safe_resolve(tmp_path, "")
+    assert result.resolve() == tmp_path.resolve()
diff --git a/tests/test_shared/test_saga_crypto_wiring.py b/tests/test_shared/test_saga_crypto_wiring.py
new file mode 100644
index 00000000..1951fbe3
--- /dev/null
+++ b/tests/test_shared/test_saga_crypto_wiring.py
@@ -0,0 +1,17 @@
+"""Verify that saga._load_state and backup_cron._load_state use saga_crypto functions."""
+
+import inspect
+from shared import saga
+from features import backup_cron
+
+
+def test_saga_load_state_uses_read_state():
+    """saga._load_state should call read_state or read_state_legacy_or_encrypted."""
+    source = inspect.getsource(saga.Saga._load_state)
+    assert "read_state" in source, "saga._load_state doesn't use saga_crypto.read_state"
+
+
+def test_backup_cron_load_state_uses_read_state():
+    """backup_cron._load_state should call read_state or read_state_legacy_or_encrypted."""
+    source = inspect.getsource(backup_cron.BackupCron._load_state)
+    assert "read_state" in source, "backup_cron._load_state doesn't use saga_crypto.read_state"
diff --git a/tests/test_wiki/test_manager.py b/tests/test_wiki/test_manager.py
new file mode 100644
index 00000000..d62316f6
--- /dev/null
+++ b/tests/test_wiki/test_manager.py
@@ -0,0 +1,127 @@
+"""Tests for WikiManager — unified wiki with layer separation."""
+
+import pytest
+from wiki.manager import WikiManager
+
+
+@pytest.fixture
+async def user_wiki(tmp_path):
+    wm = WikiManager(layer="user", base_dir=str(tmp_path / "wiki"))
+    await wm.init_db()
+    # Clean entries from previous tests (FTS5 content tables auto-sync via triggers)
+    conn = await wm._cm.get("memory.db")
+    await conn.execute("DELETE FROM wiki_index WHERE layer='user'")
+    await conn.commit()
+    return wm
+
+
+@pytest.fixture
+async def agent_wiki(tmp_path):
+    wm = WikiManager(layer="agent", base_dir=str(tmp_path / "wiki"))
+    await wm.init_db()
+    # Clean entries from previous tests
+    conn = await wm._cm.get("memory.db")
+    await conn.execute("DELETE FROM wiki_index WHERE layer='agent'")
+    await conn.commit()
+    return wm
+
+
+@pytest.mark.asyncio
+async def test_user_wiki_add_and_get(user_wiki):
+    path = await user_wiki.add("diary", "Test Entry", "Some content", tags=["test"])
+    assert path.endswith(".md")
+    entry = await user_wiki.get(path)
+    assert entry is not None
+    assert entry.title == "Test Entry"
+
+
+@pytest.mark.asyncio
+async def test_agent_wiki_add_and_get(agent_wiki):
+    path = await agent_wiki.add("decision_log", "Decision", "Chose X", tags=["arch"])
+    entry = await agent_wiki.get(path)
+    assert entry is not None
+    assert entry.title == "Decision"
+
+
+@pytest.mark.asyncio
+async def test_wiki_type_isolation(user_wiki, agent_wiki):
+    """User and agent wikis use different layers."""
+    await user_wiki.add("diary", "User Note", "content")
+    await agent_wiki.add("decision_log", "Agent Note", "content")
+    assert await user_wiki.count() == 1
+    assert await agent_wiki.count() == 1
+
+
+@pytest.mark.asyncio
+async def test_wiki_disabled_type_raises(user_wiki):
+    with pytest.raises(ValueError, match="disabled"):
+        await user_wiki.add("nonexistent_type", "Title", "Content")
+
+
+@pytest.mark.asyncio
+async def test_wiki_update(user_wiki):
+    path = await user_wiki.add("diary", "Original", "content")
+    await user_wiki.update(path, title="Updated")
+    entry = await user_wiki.get(path)
+    assert entry.title == "Updated"
+
+
+@pytest.mark.asyncio
+async def test_wiki_delete(user_wiki):
+    path = await user_wiki.add("diary", "ToDelete", "content")
+    result = await user_wiki.delete(path)
+    assert result is True
+    entry = await user_wiki.get(path)
+    assert entry is None
+
+
+@pytest.mark.asyncio
+async def test_wiki_search(user_wiki):
+    await user_wiki.add("diary", "Python Learning", "Learned async/await today")
+    await user_wiki.add("diary", "Grocery Shopping", "Bought milk and eggs")
+    results = await user_wiki.search("Python")
+    assert len(results) >= 1
+    assert any("Python" in r["title"] for r in results)
+
+
+@pytest.mark.asyncio
+async def test_wiki_list_by_type(user_wiki):
+    await user_wiki.add("diary", "Entry 1", "content")
+    await user_wiki.add("diary", "Entry 2", "content")
+    entries = await user_wiki.list_by_type("diary")
+    assert len(entries) == 2
+
+
+@pytest.mark.asyncio
+async def test_wiki_list_all(user_wiki):
+    await user_wiki.add("diary", "Entry 1", "content")
+    await user_wiki.add("relationships", "Friend", "Best friend")
+    entries = await user_wiki.list_all()
+    assert len(entries) == 2
+
+
+@pytest.mark.asyncio
+async def test_wiki_count(user_wiki):
+    await user_wiki.add("diary", "Entry 1", "content")
+    await user_wiki.add("diary", "Entry 2", "content")
+    assert await user_wiki.count() == 2
+    assert await user_wiki.count("diary") == 2
+    assert await user_wiki.count("relationships") == 0
+
+
+@pytest.mark.asyncio
+async def test_wiki_rejects_traversal(user_wiki):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        await user_wiki.update("../../etc/passwd", content="pwned")
+
+
+def test_wiki_enabled_types(user_wiki):
+    types = user_wiki.get_enabled_types()
+    assert "diary" in types
+    assert "relationships" in types
+
+
+def test_wiki_agent_enabled_types(agent_wiki):
+    types = agent_wiki.get_enabled_types()
+    assert "decision_log" in types
+    assert "error_analysis" in types
diff --git a/tests/test_wiki/test_path_safety.py b/tests/test_wiki/test_path_safety.py
new file mode 100644
index 00000000..97f50cd9
--- /dev/null
+++ b/tests/test_wiki/test_path_safety.py
@@ -0,0 +1,45 @@
+"""Tests for FileWiki path traversal prevention."""
+
+import pytest
+from wiki.manager import WikiManager
+
+
+@pytest.fixture
+def wiki(tmp_path):
+    return WikiManager(layer="user", base_dir=str(tmp_path / "wiki"))
+
+
+@pytest.mark.asyncio
+async def test_update_rejects_traversal(wiki):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        await wiki.update("../../etc/passwd", content="pwned")
+
+
+@pytest.mark.asyncio
+async def test_get_rejects_traversal(wiki):
+    result = await wiki.get("../../etc/passwd")
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_delete_rejects_traversal(wiki):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        await wiki.delete("../../etc/passwd")
+
+
+@pytest.mark.asyncio
+async def test_update_rejects_absolute_path(wiki):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        await wiki.update("/etc/passwd", content="pwned")
+
+
+@pytest.mark.asyncio
+async def test_get_rejects_absolute_path(wiki):
+    result = await wiki.get("/etc/passwd")
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_delete_rejects_absolute_path(wiki):
+    with pytest.raises(ValueError, match="escapes base directory"):
+        await wiki.delete("/etc/passwd")
diff --git a/tests/test_wiki/test_wiki.py b/tests/test_wiki/test_wiki.py
index 77d42434..7eabe31f 100644
--- a/tests/test_wiki/test_wiki.py
+++ b/tests/test_wiki/test_wiki.py
@@ -1,4 +1,4 @@
-"""Tests for wiki/ module (FileWiki) — async."""
+"""Tests for wiki/ module (WikiManager) — async."""
 
 import asyncio
 import sys
@@ -8,10 +8,10 @@
 
 
 def test_file_wiki_add_search():
-    from wiki.file_wiki import FileWiki
+    from wiki.manager import WikiManager
 
     async def t():
-        w = FileWiki(layer="user")
+        w = WikiManager(layer="user")
         path = await w.add("work_notes", "Test Entry 2026", "Test content here", tags=["test"])
         assert path is not None
         assert Path(path).exists()
@@ -22,19 +22,19 @@ async def t():
 
 
 def test_file_wiki_enabled_types():
-    from wiki.file_wiki import FileWiki
+    from wiki.manager import WikiManager
 
-    w = FileWiki(layer="user")
+    w = WikiManager(layer="user")
     types = w.get_enabled_types()
     assert len(types) > 0
     assert "diary" in types
 
 
 def test_file_wiki_count():
-    from wiki.file_wiki import FileWiki
+    from wiki.manager import WikiManager
 
     async def t():
-        w = FileWiki(layer="user")
+        w = WikiManager(layer="user")
         await w.add("work_notes", "Count Test 2026", "content")
         assert await w.count() >= 1
 
diff --git a/wiki/__init__.py b/wiki/__init__.py
index 2b3cef6d..82892eac 100644
--- a/wiki/__init__.py
+++ b/wiki/__init__.py
@@ -2,8 +2,11 @@
 Wiki Module — .md files as source of truth + SQLite FTS5 index
 """
 
-from .agent_wiki import AgentWiki
-from .file_wiki import ALL_AGENT_TYPES, ALL_USER_TYPES, FileWiki
-from .user_wiki import UserWiki
+from .manager import ALL_AGENT_TYPES, ALL_USER_TYPES, WikiManager
 
-__all__ = ["FileWiki", "UserWiki", "AgentWiki", "ALL_USER_TYPES", "ALL_AGENT_TYPES"]
+# Backward-compatible aliases
+FileWiki = WikiManager
+UserWiki = WikiManager
+AgentWiki = WikiManager
+
+__all__ = ["WikiManager", "FileWiki", "UserWiki", "AgentWiki", "ALL_USER_TYPES", "ALL_AGENT_TYPES"]
diff --git a/wiki/agent_wiki.py b/wiki/agent_wiki.py
deleted file mode 100644
index 1227bce7..00000000
--- a/wiki/agent_wiki.py
+++ /dev/null
@@ -1,261 +0,0 @@
-"""
-Agent Wiki — 7 types of agent identity knowledge
-Supports external folders for lore, knowledge bases, style guides
-"""
-
-import json
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Optional
-
-from shared.connection import AsyncConnectionManager, connection_manager
-from wiki.shared import (
-    build_count_query,
-    build_update_clause,
-    find_by_source,
-    format_search_result,
-    get_enabled_types,
-    get_external_dirs,
-    parse_tags,
-)
-
-
-@dataclass
-class AgentWikiEntry:
-    entry_id: int
-    user_id: str
-    wiki_type: str
-    title: str
-    content: str
-    tags: list[str]
-    importance: float
-    created_at: float
-    updated_at: float
-
-
-ALL_WIKI_TYPES = [
-    "decision_log",
-    "error_analysis",
-    "personality_evolution",
-    "emotional_context",
-    "wiki_agent",
-    "learning_journal",
-    "principle_log",
-]
-
-
-def _get_enabled_types() -> list[str]:
-    return get_enabled_types("agent", ALL_WIKI_TYPES)
-
-
-def _get_external_dirs() -> list[str]:
-    return get_external_dirs("agent")
-
-
-class AgentWiki:
-    def __init__(self, cm: Optional[AsyncConnectionManager] = None):
-        self._cm = cm or connection_manager
-
-    async def init_db(self):
-        await self._cm.execute_script(
-            "memory.db",
-            """
-            CREATE TABLE IF NOT EXISTS agent_wiki (
-                entry_id INTEGER PRIMARY KEY AUTOINCREMENT,
-                user_id TEXT NOT NULL,
-                wiki_type TEXT NOT NULL,
-                title TEXT NOT NULL,
-                content TEXT NOT NULL,
-                tags TEXT,
-                importance REAL DEFAULT 0.5,
-                source TEXT DEFAULT 'manual',
-                created_at REAL NOT NULL,
-                updated_at REAL NOT NULL
-            );
-            CREATE INDEX IF NOT EXISTS idx_awiki_user ON agent_wiki(user_id);
-            CREATE INDEX IF NOT EXISTS idx_awiki_type ON agent_wiki(wiki_type);
-            CREATE INDEX IF NOT EXISTS idx_awiki_user_type ON agent_wiki(user_id, wiki_type);
-            CREATE INDEX IF NOT EXISTS idx_awiki_source ON agent_wiki(source);
-            CREATE INDEX IF NOT EXISTS idx_awiki_updated ON agent_wiki(updated_at);
-        """,
-        )
-        conn = await self._cm.get("memory.db")
-        try:
-            await conn.execute("ALTER TABLE agent_wiki ADD COLUMN source TEXT DEFAULT 'manual'")
-        except Exception:
-            pass
-        await conn.execute("""
-            CREATE VIRTUAL TABLE IF NOT EXISTS agent_wiki_fts USING fts5(
-                title, content, wiki_type,
-                content=agent_wiki,
-                content_rowid=entry_id
-            )
-        """)
-        await conn.commit()
-
-    async def add(
-        self,
-        user_id: str,
-        wiki_type: str,
-        title: str,
-        content: str,
-        tags: Optional[list[str]] = None,
-        importance: float = 0.5,
-        source: str = "manual",
-    ) -> int:
-        enabled = _get_enabled_types()
-        if enabled and wiki_type not in enabled:
-            raise ValueError(f"Wiki type '{wiki_type}' is disabled. Enabled: {enabled}")
-
-        conn = await self._cm.get("memory.db")
-        now = time.time()
-        cur = await conn.execute(
-            "INSERT INTO agent_wiki (user_id, wiki_type, title, content, tags, importance, source, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-            (user_id, wiki_type, title, content, json.dumps(tags or []), importance, source, now, now),
-        )
-        entry_id = cur.lastrowid
-        await conn.execute(
-            "INSERT INTO agent_wiki_fts(rowid, title, content, wiki_type) VALUES (?, ?, ?, ?)",
-            (entry_id, title, content, wiki_type),
-        )
-        await conn.commit()
-        return entry_id
-
-    async def update(
-        self,
-        entry_id: int,
-        title: Optional[str] = None,
-        content: Optional[str] = None,
-        tags: Optional[list[str]] = None,
-        importance: Optional[float] = None,
-    ):
-        conn = await self._cm.get("memory.db")
-        updates, params = build_update_clause({"title": title, "content": content, "tags": tags, "importance": importance})
-        params.append(entry_id)
-        await conn.execute(f"UPDATE agent_wiki SET {', '.join(updates)} WHERE entry_id=?", params)
-        await conn.commit()
-
-    async def get(self, entry_id: int) -> AgentWikiEntry | None:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute("SELECT * FROM agent_wiki WHERE entry_id=?", (entry_id,))
-        row = await cur.fetchone()
-        return self._row_to_entry(row) if row else None
-
-    async def search(self, user_id: str, query: str, limit: int = 10) -> list[dict[str, Any]]:
-        try:
-            conn = await self._cm.get("memory.db")
-            cur = await conn.execute(
-                """SELECT aw.entry_id, aw.title, aw.content, aw.wiki_type, aw.tags, aw.importance, fts.rank
-                   FROM agent_wiki_fts fts JOIN agent_wiki aw ON fts.rowid = aw.entry_id
-                   WHERE agent_wiki_fts MATCH ? AND aw.user_id = ?
-                   ORDER BY fts.rank DESC LIMIT ?""",
-                (query, user_id, limit),
-            )
-            rows = await cur.fetchall()
-            return [format_search_result(r) for r in rows]
-        except Exception:
-            return []
-
-    async def list_by_type(self, user_id: str, wiki_type: str, limit: int = 20) -> list[AgentWikiEntry]:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute(
-            "SELECT * FROM agent_wiki WHERE user_id=? AND wiki_type=? ORDER BY updated_at DESC LIMIT ?",
-            (user_id, wiki_type, limit),
-        )
-        rows = await cur.fetchall()
-        return [self._row_to_entry(r) for r in rows]
-
-    async def list_all(self, user_id: str, limit: int = 50) -> list[AgentWikiEntry]:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute("SELECT * FROM agent_wiki WHERE user_id=? ORDER BY updated_at DESC LIMIT ?", (user_id, limit))
-        rows = await cur.fetchall()
-        return [self._row_to_entry(r) for r in rows]
-
-    async def delete(self, entry_id: int) -> bool:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute("DELETE FROM agent_wiki WHERE entry_id=?", (entry_id,))
-        await conn.commit()
-        return cur.rowcount > 0
-
-    async def count(self, user_id: Optional[str] = None, wiki_type: Optional[str] = None) -> int:
-        conn = await self._cm.get("memory.db")
-        query, params = build_count_query("agent_wiki", user_id, wiki_type)
-        cur = await conn.execute(query, params)
-        row = await cur.fetchone()
-        return row[0] if row else 0
-
-    def get_enabled_types(self) -> list[str]:
-        return _get_enabled_types()
-
-    def get_external_dirs(self) -> list[str]:
-        return _get_external_dirs()
-
-    async def sync_external(self, user_id: str) -> dict[str, int]:
-        """Sync external .md files (lore, knowledge bases, style guides) into wiki."""
-        results = {"imported": 0, "skipped": 0, "errors": 0}
-        for dir_path in _get_external_dirs():
-            p = Path(dir_path)
-            if not p.exists():
-                continue
-            for md_file in p.glob("**/*.md"):
-                try:
-                    content = md_file.read_text(encoding="utf-8")
-                    title = md_file.stem
-                    wiki_type = self._guess_type(md_file, content)
-                    if wiki_type not in _get_enabled_types():
-                        results["skipped"] += 1
-                        continue
-                    existing = await self._find_by_source(user_id, str(md_file))
-                    if existing:
-                        await self.update(existing, title=title, content=content)
-                    else:
-                        await self.add(user_id, wiki_type, title, content, tags=[md_file.parent.name], source=str(md_file))
-                    results["imported"] += 1
-                except Exception:
-                    results["errors"] += 1
-        return results
-
-    async def _find_by_source(self, user_id: str, source: str) -> int | None:
-        return await find_by_source(self._cm, "agent_wiki", user_id, source)
-
-    def _guess_type(self, path: Path, content: str) -> str:
-        name = path.stem.lower()
-        parent = path.parent.name.lower()
-
-        if any(w in name or w in parent for w in ["lore", "лор", "world", "мир"]):
-            return "wiki_agent"
-        if any(w in name or w in parent for w in ["knowledge", "знани", "reference", "справочник"]):
-            return "wiki_agent"
-        if any(w in name or w in parent for w in ["style", "стиль", "guide", "гайд"]):
-            return "personality_evolution"
-        if any(w in name or w in parent for w in ["error", "ошибк", "bug"]):
-            return "error_analysis"
-        if any(w in name or w in parent for w in ["decision", "решени", "choice"]):
-            return "decision_log"
-        if any(w in name or w in parent for w in ["learning", "обучен", "learn"]):
-            return "learning_journal"
-        if any(w in name or w in parent for w in ["principle", "принцип", "rule", "правило"]):
-            return "principle_log"
-
-        if any(w in content.lower() for w in ["решение", "decided", "chose"]):
-            return "decision_log"
-        if any(w in content.lower() for w in ["ошибка", "error", "bug", "исправил"]):
-            return "error_analysis"
-        if any(w in content.lower() for w in ["принцип", "principle", "всегда", "никогда"]):
-            return "principle_log"
-
-        return "wiki_agent"
-
-    def _row_to_entry(self, row) -> AgentWikiEntry:
-        return AgentWikiEntry(
-            entry_id=row["entry_id"],
-            user_id=row["user_id"],
-            wiki_type=row["wiki_type"],
-            title=row["title"],
-            content=row["content"],
-            tags=parse_tags(row["tags"]),
-            importance=row["importance"],
-            created_at=row["created_at"],
-            updated_at=row["updated_at"],
-        )
diff --git a/wiki/file_wiki.py b/wiki/manager.py
similarity index 95%
rename from wiki/file_wiki.py
rename to wiki/manager.py
index 0463a3d2..3221b183 100644
--- a/wiki/file_wiki.py
+++ b/wiki/manager.py
@@ -1,6 +1,7 @@
 """
-File-based Wiki — .md files as source of truth + SQLite index for search.
-Architecture: files on disk = primary, DB = index/cache.
+WikiManager — unified wiki system with layer-based separation.
+Architecture: .md files on disk = primary, SQLite FTS5 = search index.
+Layers: user, agent, shared.
 """
 
 import json
@@ -10,6 +11,7 @@
 from typing import Any, Optional
 
 from shared.connection import AsyncConnectionManager, connection_manager
+from shared.path_safety import safe_resolve
 from wiki.shared import (
     get_enabled_types,
     get_external_dirs,
@@ -41,9 +43,15 @@ class WikiEntry:
     "principle_log",
 ]
 
+LAYER_TYPES = {
+    "user": ALL_USER_TYPES,
+    "agent": ALL_AGENT_TYPES,
+    "shared": ALL_USER_TYPES + ALL_AGENT_TYPES,
+}
 
-class FileWiki:
-    """Wiki where .md files are source of truth, SQLite is search index."""
+
+class WikiManager:
+    """Unified wiki: .md files on disk + SQLite FTS5 index, parameterized by layer."""
 
     def __init__(self, layer: str = "user", base_dir: Optional[str] = None, cm: Optional[AsyncConnectionManager] = None):
         self.layer = layer
@@ -85,7 +93,7 @@ async def init_db(self):
         await conn.commit()
 
     def _get_enabled_types(self) -> list[str]:
-        all_types = ALL_USER_TYPES if "user" in self.layer else ALL_AGENT_TYPES
+        all_types = LAYER_TYPES.get(self.layer, ALL_USER_TYPES)
         return get_enabled_types(self.layer, all_types)
 
     def _type_dir(self, wiki_type: str) -> Path:
@@ -117,7 +125,7 @@ async def update(
         importance: Optional[float] = None,
     ):
         """Update .md file and re-index."""
-        p = Path(file_path)
+        p = safe_resolve(self.base_dir, file_path)
         if not p.exists():
             return
 
@@ -134,7 +142,10 @@ async def update(
         await self._index_file(p, wiki_type, new_title, new_content, new_tags, new_importance)
 
     async def get(self, file_path: str) -> WikiEntry | None:
-        p = Path(file_path)
+        try:
+            p = safe_resolve(self.base_dir, file_path)
+        except ValueError:
+            return None
         if not p.exists():
             return None
         parsed = self._parse_md(p.read_text(encoding="utf-8"))
@@ -240,7 +251,7 @@ async def list_all(self, limit: int = 50) -> list[WikiEntry]:
         return entries
 
     async def delete(self, file_path: str) -> bool:
-        p = Path(file_path)
+        p = safe_resolve(self.base_dir, file_path)
         if p.exists():
             p.unlink()
         conn = await self._cm.get("memory.db")
@@ -408,7 +419,7 @@ def _to_md(self, title: str, content: str, tags: Optional[list[str]] = None, imp
     def _guess_type(self, path: Path, content: str) -> str:
         name = path.stem.lower()
         parent = path.parent.name.lower()
-        all_types = ALL_USER_TYPES if self.layer == "user" else ALL_AGENT_TYPES
+        all_types = LAYER_TYPES.get(self.layer, ALL_USER_TYPES)
 
         for t in all_types:
             if t in name or t in parent:
diff --git a/wiki/user_wiki.py b/wiki/user_wiki.py
deleted file mode 100644
index 83ec6a40..00000000
--- a/wiki/user_wiki.py
+++ /dev/null
@@ -1,238 +0,0 @@
-"""
-User Wiki — 7 types of user knowledge
-Supports external folders for auto-sync
-"""
-
-import json
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Optional
-
-from shared.connection import AsyncConnectionManager, connection_manager
-from wiki.shared import (
-    build_count_query,
-    build_update_clause,
-    find_by_source,
-    format_search_result,
-    get_enabled_types,
-    get_external_dirs,
-    parse_tags,
-)
-
-
-@dataclass
-class WikiEntry:
-    entry_id: int
-    user_id: str
-    wiki_type: str
-    title: str
-    content: str
-    tags: list[str]
-    importance: float
-    created_at: float
-    updated_at: float
-
-
-ALL_WIKI_TYPES = ["diary", "relationships", "desires", "aspirations", "work_notes", "preferences", "retrospective"]
-
-
-def _get_enabled_types() -> list[str]:
-    """Get enabled wiki types from config."""
-    return get_enabled_types("user", ALL_WIKI_TYPES)
-
-
-def _get_external_dirs() -> list[str]:
-    """Get external directories from config."""
-    return get_external_dirs("user")
-
-
-class UserWiki:
-    def __init__(self, cm: Optional[AsyncConnectionManager] = None):
-        self._cm = cm or connection_manager
-
-    async def init_db(self):
-        await self._cm.execute_script(
-            "memory.db",
-            """
-            CREATE TABLE IF NOT EXISTS user_wiki (
-                entry_id INTEGER PRIMARY KEY AUTOINCREMENT,
-                user_id TEXT NOT NULL,
-                wiki_type TEXT NOT NULL,
-                title TEXT NOT NULL,
-                content TEXT NOT NULL,
-                tags TEXT,
-                importance REAL DEFAULT 0.5,
-                source TEXT DEFAULT 'manual',
-                created_at REAL NOT NULL,
-                updated_at REAL NOT NULL
-            );
-            CREATE INDEX IF NOT EXISTS idx_uwiki_user ON user_wiki(user_id);
-            CREATE INDEX IF NOT EXISTS idx_uwiki_type ON user_wiki(wiki_type);
-            CREATE INDEX IF NOT EXISTS idx_uwiki_user_type ON user_wiki(user_id, wiki_type);
-            CREATE INDEX IF NOT EXISTS idx_uwiki_source ON user_wiki(source);
-            CREATE INDEX IF NOT EXISTS idx_uwiki_updated ON user_wiki(updated_at);
-        """,
-        )
-        conn = await self._cm.get("memory.db")
-        try:
-            await conn.execute("ALTER TABLE user_wiki ADD COLUMN source TEXT DEFAULT 'manual'")
-        except Exception:
-            pass
-        await conn.execute("""
-            CREATE VIRTUAL TABLE IF NOT EXISTS user_wiki_fts USING fts5(
-                title, content, wiki_type,
-                content=user_wiki,
-                content_rowid=entry_id
-            )
-        """)
-        await conn.commit()
-
-    async def add(
-        self,
-        user_id: str,
-        wiki_type: str,
-        title: str,
-        content: str,
-        tags: Optional[list[str]] = None,
-        importance: float = 0.5,
-        source: str = "manual",
-    ) -> int:
-        enabled = _get_enabled_types()
-        if enabled and wiki_type not in enabled:
-            raise ValueError(f"Wiki type '{wiki_type}' is disabled. Enabled: {enabled}")
-
-        conn = await self._cm.get("memory.db")
-        now = time.time()
-        cur = await conn.execute(
-            "INSERT INTO user_wiki (user_id, wiki_type, title, content, tags, importance, source, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-            (user_id, wiki_type, title, content, json.dumps(tags or []), importance, source, now, now),
-        )
-        entry_id = cur.lastrowid
-        await conn.execute(
-            "INSERT INTO user_wiki_fts(rowid, title, content, wiki_type) VALUES (?, ?, ?, ?)",
-            (entry_id, title, content, wiki_type),
-        )
-        await conn.commit()
-        return entry_id
-
-    async def update(
-        self,
-        entry_id: int,
-        title: Optional[str] = None,
-        content: Optional[str] = None,
-        tags: Optional[list[str]] = None,
-        importance: Optional[float] = None,
-    ):
-        conn = await self._cm.get("memory.db")
-        updates, params = build_update_clause({"title": title, "content": content, "tags": tags, "importance": importance})
-        params.append(entry_id)
-        await conn.execute(f"UPDATE user_wiki SET {', '.join(updates)} WHERE entry_id=?", params)
-        await conn.commit()
-
-    async def get(self, entry_id: int) -> WikiEntry | None:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute("SELECT * FROM user_wiki WHERE entry_id=?", (entry_id,))
-        row = await cur.fetchone()
-        return self._row_to_entry(row) if row else None
-
-    async def search(self, user_id: str, query: str, limit: int = 10) -> list[dict[str, Any]]:
-        try:
-            conn = await self._cm.get("memory.db")
-            cur = await conn.execute(
-                """SELECT uw.entry_id, uw.title, uw.content, uw.wiki_type, uw.tags, uw.importance, fts.rank
-                   FROM user_wiki_fts fts JOIN user_wiki uw ON fts.rowid = uw.entry_id
-                   WHERE user_wiki_fts MATCH ? AND uw.user_id = ?
-                   ORDER BY fts.rank DESC LIMIT ?""",
-                (query, user_id, limit),
-            )
-            rows = await cur.fetchall()
-            return [format_search_result(r) for r in rows]
-        except Exception:
-            return []
-
-    async def list_by_type(self, user_id: str, wiki_type: str, limit: int = 20) -> list[WikiEntry]:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute(
-            "SELECT * FROM user_wiki WHERE user_id=? AND wiki_type=? ORDER BY updated_at DESC LIMIT ?",
-            (user_id, wiki_type, limit),
-        )
-        rows = await cur.fetchall()
-        return [self._row_to_entry(r) for r in rows]
-
-    async def list_all(self, user_id: str, limit: int = 50) -> list[WikiEntry]:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute("SELECT * FROM user_wiki WHERE user_id=? ORDER BY updated_at DESC LIMIT ?", (user_id, limit))
-        rows = await cur.fetchall()
-        return [self._row_to_entry(r) for r in rows]
-
-    async def delete(self, entry_id: int) -> bool:
-        conn = await self._cm.get("memory.db")
-        cur = await conn.execute("DELETE FROM user_wiki WHERE entry_id=?", (entry_id,))
-        await conn.commit()
-        return cur.rowcount > 0
-
-    async def count(self, user_id: Optional[str] = None, wiki_type: Optional[str] = None) -> int:
-        conn = await self._cm.get("memory.db")
-        query, params = build_count_query("user_wiki", user_id, wiki_type)
-        cur = await conn.execute(query, params)
-        row = await cur.fetchone()
-        return row[0] if row else 0
-
-    def get_enabled_types(self) -> list[str]:
-        return _get_enabled_types()
-
-    def get_external_dirs(self) -> list[str]:
-        return _get_external_dirs()
-
-    async def sync_external(self, user_id: str) -> dict[str, int]:
-        """Sync external .md files into wiki."""
-        results = {"imported": 0, "skipped": 0, "errors": 0}
-        for dir_path in _get_external_dirs():
-            p = Path(dir_path)
-            if not p.exists():
-                continue
-            for md_file in p.glob("**/*.md"):
-                try:
-                    content = md_file.read_text(encoding="utf-8")
-                    title = md_file.stem
-                    wiki_type = self._guess_type(md_file, content)
-                    if wiki_type not in _get_enabled_types():
-                        results["skipped"] += 1
-                        continue
-                    existing = await self._find_by_source(user_id, str(md_file))
-                    if existing:
-                        await self.update(existing, title=title, content=content)
-                    else:
-                        await self.add(user_id, wiki_type, title, content, tags=[md_file.parent.name], source=str(md_file))
-                    results["imported"] += 1
-                except Exception:
-                    results["errors"] += 1
-        return results
-
-    async def _find_by_source(self, user_id: str, source: str) -> int | None:
-        return await find_by_source(self._cm, "user_wiki", user_id, source)
-
-    def _guess_type(self, path: Path, content: str) -> str:
-        name = path.stem.lower()
-        for t in ALL_WIKI_TYPES:
-            if t in name:
-                return t
-        if any(w in content.lower() for w in ["дневник", "diary", "сегодня"]):
-            return "diary"
-        if any(w in content.lower() for w in ["проект", "процесс", "задача"]):
-            return "work_notes"
-        return "diary"
-
-    def _row_to_entry(self, row) -> WikiEntry:
-        return WikiEntry(
-            entry_id=row["entry_id"],
-            user_id=row["user_id"],
-            wiki_type=row["wiki_type"],
-            title=row["title"],
-            content=row["content"],
-            tags=parse_tags(row["tags"]),
-            importance=row["importance"],
-            created_at=row["created_at"],
-            updated_at=row["updated_at"],
-        )