fix: Windows compatibility — room routing, Unicode output, test cleanup

mempalace/compat.py

@@ -0,0 +1,15 @@
+"""Cross-platform console compatibility helpers."""
+
+import sys
+
+
+def _stdout_supports(chars: str) -> bool:
+    encoding = getattr(sys.stdout, "encoding", "ascii") or "ascii"
+    try:
+        chars.encode(encoding)
+        return True
+    except (UnicodeEncodeError, LookupError, TypeError):
+        return False
+
+
+CHECKMARK = "✓" if _stdout_supports("✓") else "+"

mempalace/convo_miner.py

@@ -15,6 +15,7 @@
 from datetime import datetime
 from collections import defaultdict
 
+from .compat import CHECKMARK as _CHECKMARK
 from .normalize import normalize
 from .palace import SKIP_DIRS, get_collection, file_already_mined
 
@@ -356,7 +357,7 @@ def mine_convos(
                     raise
 
         total_drawers += drawers_added
-        print(f"  ✓ [{i:4}/{len(files)}] {filepath.name[:50]:50} +{drawers_added}")
+        print(f"  {_CHECKMARK} [{i:4}/{len(files)}] {filepath.name[:50]:50} +{drawers_added}")
 
     print(f"\n{'=' * 55}")
     print("  Done.")

mempalace/entity_detector.py

@@ -18,6 +18,8 @@
 import re
 import os
 from pathlib import Path
+
+from .compat import _stdout_supports
 from collections import defaultdict
 
 
@@ -708,8 +710,10 @@ def _print_entity_list(entities: list, label: str):
     if not entities:
         print("    (none detected)")
         return
+    filled_char, empty_char = ("●", "○") if _stdout_supports("●○") else ("#", ".")
     for i, e in enumerate(entities):
-        confidence_bar = "●" * int(e["confidence"] * 5) + "○" * (5 - int(e["confidence"] * 5))
+        filled = int(e["confidence"] * 5)
+        confidence_bar = filled_char * filled + empty_char * (5 - filled)
         signals_str = ", ".join(e["signals"][:2]) if e["signals"] else ""
         print(f"    {i + 1:2}. {e['name']:20} [{confidence_bar}] {signals_str}")
 

mempalace/miner.py

@@ -17,6 +17,7 @@
 
 import chromadb
 
+from .compat import CHECKMARK as _CHECKMARK
 from .palace import SKIP_DIRS, get_collection, file_already_mined
 
 READABLE_EXTENSIONS = {
@@ -48,6 +49,7 @@
     "mempal.yaml",
     "mempal.yml",
     ".gitignore",
+    ".mpignore",
     "package-lock.json",
 }
 
@@ -70,13 +72,13 @@ def __init__(self, base_dir: Path, rules: list):
         self.rules = rules
 
     @classmethod
-    def from_dir(cls, dir_path: Path):
-        gitignore_path = dir_path / ".gitignore"
-        if not gitignore_path.is_file():
+    def from_dir(cls, dir_path: Path, filename: str = ".gitignore"):
+        ignore_path = dir_path / filename
+        if not ignore_path.is_file():
             return None
 
         try:
-            lines = gitignore_path.read_text(encoding="utf-8", errors="replace").splitlines()
+            lines = ignore_path.read_text(encoding="utf-8", errors="replace").splitlines()
         except Exception:
             return None
 
@@ -178,15 +180,16 @@ def matches(path_index: int, pattern_index: int) -> bool:
         return matches(0, 0)
 
 
-def load_gitignore_matcher(dir_path: Path, cache: dict):
-    """Load and cache one directory's .gitignore matcher."""
-    if dir_path not in cache:
-        cache[dir_path] = GitignoreMatcher.from_dir(dir_path)
-    return cache[dir_path]
+def load_ignore_matcher(dir_path: Path, cache: dict, filename: str = ".gitignore"):
+    """Load and cache one directory's ignore-file matcher."""
+    key = (dir_path, filename)
+    if key not in cache:
+        cache[key] = GitignoreMatcher.from_dir(dir_path, filename=filename)
+    return cache[key]
 
 
-def is_gitignored(path: Path, matchers: list, is_dir: bool = False) -> bool:
-    """Apply active .gitignore matchers in ancestor order; last match wins."""
+def is_ignored(path: Path, matchers: list, is_dir: bool = False) -> bool:
+    """Apply active ignore matchers (.gitignore / .mpignore) in ancestor order; last match wins."""
     ignored = False
     for matcher in matchers:
         decision = matcher.matches(path, is_dir=is_dir)
@@ -289,16 +292,20 @@ def detect_room(filepath: Path, content: str, rooms: list, project_path: Path) -
     content_lower = content[:2000].lower()
 
     # Priority 1: folder path matches room name or keywords
+    # Exact match or hyphen-component match to avoid false positives from
+    # short directory names (e.g. "ml/" matching room "visualizeml" via substring)
+    # while still allowing "backend" to match keyword "backend-api".
     path_parts = relative.replace("\\", "/").split("/")
     for part in path_parts[:-1]:  # skip filename itself
         for room in rooms:
             candidates = [room["name"].lower()] + [k.lower() for k in room.get("keywords", [])]
-            if any(part == c or c in part or part in c for c in candidates):
+            if any(part == c or part in c.split("-") for c in candidates):
                 return room["name"]
 
-    # Priority 2: filename matches room name
+    # Priority 2: filename matches room name or keywords
     for room in rooms:
-        if room["name"].lower() in filename or filename in room["name"].lower():
+        candidates = [room["name"].lower()] + [k.lower() for k in room.get("keywords", [])]
+        if any(filename == c or filename in c.split("-") for c in candidates):
             return room["name"]
 
     # Priority 3: keyword scoring from room keywords + name
@@ -466,35 +473,43 @@ def scan_project(
     """Return list of all readable file paths."""
     project_path = Path(project_dir).expanduser().resolve()
     files = []
-    active_matchers = []
+    active_gi_matchers = []
+    active_mp_matchers = []
     matcher_cache = {}
     include_paths = normalize_include_paths(include_ignored)
 
+    def _prune_matchers(matchers, root):
+        return [m for m in matchers if root.is_relative_to(m.base_dir)]
+
     for root, dirs, filenames in os.walk(project_path):
         root_path = Path(root)
 
+        # .mpignore is always active (independent of --no-gitignore)
+        active_mp_matchers = _prune_matchers(active_mp_matchers, root_path)
+        mp_matcher = load_ignore_matcher(root_path, matcher_cache, ".mpignore")
+        if mp_matcher is not None:
+            active_mp_matchers.append(mp_matcher)
+
         if respect_gitignore:
-            active_matchers = [
-                matcher
-                for matcher in active_matchers
-                if root_path == matcher.base_dir or matcher.base_dir in root_path.parents
-            ]
-            current_matcher = load_gitignore_matcher(root_path, matcher_cache)
-            if current_matcher is not None:
-                active_matchers.append(current_matcher)
+            active_gi_matchers = _prune_matchers(active_gi_matchers, root_path)
+            gi_matcher = load_ignore_matcher(root_path, matcher_cache, ".gitignore")
+            if gi_matcher is not None:
+                active_gi_matchers.append(gi_matcher)
+
+        all_matchers = active_gi_matchers + active_mp_matchers if respect_gitignore else active_mp_matchers
 
         dirs[:] = [
             d
             for d in dirs
             if is_force_included(root_path / d, project_path, include_paths)
             or not should_skip_dir(d)
         ]
-        if respect_gitignore and active_matchers:
+        if all_matchers:
             dirs[:] = [
                 d
                 for d in dirs
                 if is_force_included(root_path / d, project_path, include_paths)
-                or not is_gitignored(root_path / d, active_matchers, is_dir=True)
+                or not is_ignored(root_path / d, all_matchers, is_dir=True)
             ]
 
         for filename in filenames:
@@ -506,8 +521,8 @@ def scan_project(
                 continue
             if filepath.suffix.lower() not in READABLE_EXTENSIONS and not exact_force_include:
                 continue
-            if respect_gitignore and active_matchers and not force_include:
-                if is_gitignored(filepath, active_matchers, is_dir=False):
+            if all_matchers and not force_include:
+                if is_ignored(filepath, all_matchers, is_dir=False):
                     continue
             # Skip symlinks — prevents following links to /dev/urandom, etc.
             if filepath.is_symlink():
@@ -563,7 +578,7 @@ def mine(
     if dry_run:
         print("  DRY RUN — nothing will be filed")
     if not respect_gitignore:
-        print("  .gitignore: DISABLED")
+        print("  .gitignore: DISABLED  (.mpignore still active)")
     if include_ignored:
         print(f"  Include: {', '.join(sorted(normalize_include_paths(include_ignored)))}")
     print(f"{'─' * 55}\n")
@@ -593,7 +608,7 @@ def mine(
             total_drawers += drawers
             room_counts[room] += 1
             if not dry_run:
-                print(f"  ✓ [{i:4}/{len(files)}] {filepath.name[:50]:50} +{drawers}")
+                print(f"  {_CHECKMARK} [{i:4}/{len(files)}] {filepath.name[:50]:50} +{drawers}")
 
     print(f"\n{'=' * 55}")
     print("  Done.")

mempalace/split_mega_files.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 """
 split_mega_files.py — Split concatenated transcript files into per-session files
-=================================================================================
 
 Scans a directory for .txt files that contain multiple Claude Code sessions
 (identified by "Claude Code v" headers). Splits each into individual files
@@ -28,6 +27,8 @@
 import re
 from pathlib import Path
 
+from .compat import CHECKMARK as _CHECKMARK
+
 HOME = Path.home()
 LUMI_DIR = Path(os.environ.get("MEMPALACE_SOURCE_DIR", str(HOME / "Desktop/transcripts")))
 
@@ -224,7 +225,7 @@ def split_file(filepath, output_dir, dry_run=False):
             print(f"  [{i + 1}/{len(boundaries) - 1}] {name}  ({len(chunk)} lines)")
         else:
             out_path.write_text("".join(chunk), encoding="utf-8")
-            print(f"  ✓ {name}  ({len(chunk)} lines)")
+            print(f"  {_CHECKMARK} {name}  ({len(chunk)} lines)")
 
         written.append(out_path)
 

tests/conftest.py

@@ -10,8 +10,10 @@
 instead of the real user profile.
 """
 
+import gc
 import os
 import shutil
+import sys
 import tempfile
 
 # ── Isolate HOME before any mempalace imports ──────────────────────────
@@ -34,6 +36,21 @@
 from mempalace.knowledge_graph import KnowledgeGraph  # noqa: E402
 
 
+def force_cleanup_tempdir(path):
+    """Best-effort temp dir removal; ChromaDB may hold file locks on Windows."""
+    try:
+        shutil.rmtree(path)
+    except PermissionError:
+        if sys.platform == "win32":
+            gc.collect()
+            import time
+
+            time.sleep(0.5)
+            shutil.rmtree(path, ignore_errors=True)
+        else:
+            raise
+
+
 @pytest.fixture(autouse=True)
 def _reset_mcp_cache():
     """Reset the MCP server's cached ChromaDB client/collection between tests."""

tests/test_convo_miner.py

@@ -1,26 +1,32 @@
+import gc
 import os
 import tempfile
-import shutil
 import chromadb
 from mempalace.convo_miner import mine_convos
 
+from conftest import force_cleanup_tempdir as _force_cleanup
+
 
 def test_convo_mining():
     tmpdir = tempfile.mkdtemp()
-    with open(os.path.join(tmpdir, "chat.txt"), "w") as f:
-        f.write(
-            "> What is memory?\nMemory is persistence.\n\n> Why does it matter?\nIt enables continuity.\n\n> How do we build it?\nWith structured storage.\n"
-        )
+    try:
+        with open(os.path.join(tmpdir, "chat.txt"), "w") as f:
+            f.write(
+                "> What is memory?\nMemory is persistence.\n\n> Why does it matter?\nIt enables continuity.\n\n> How do we build it?\nWith structured storage.\n"
+            )
 
-    palace_path = os.path.join(tmpdir, "palace")
-    mine_convos(tmpdir, palace_path, wing="test_convos")
+        palace_path = os.path.join(tmpdir, "palace")
+        mine_convos(tmpdir, palace_path, wing="test_convos")
 
-    client = chromadb.PersistentClient(path=palace_path)
-    col = client.get_collection("mempalace_drawers")
-    assert col.count() >= 2
+        client = chromadb.PersistentClient(path=palace_path)
+        col = client.get_collection("mempalace_drawers")
+        assert col.count() >= 2
 
-    # Verify search works
-    results = col.query(query_texts=["memory persistence"], n_results=1)
-    assert len(results["documents"][0]) > 0
+        # Verify search works
+        results = col.query(query_texts=["memory persistence"], n_results=1)
+        assert len(results["documents"][0]) > 0
 
-    shutil.rmtree(tmpdir, ignore_errors=True)
+        del col, client
+        gc.collect()
+    finally:
+        _force_cleanup(tmpdir)