feat: add mempalace re-mine command

Extracts source file paths from existing palace metadata, drops
the collection, and re-mines with the currently configured embedding
model. This is the recommended recovery path when intentionally
switching models.

Supports --dry-run to preview what would be re-mined.
Reports missing source files (deleted since original mining).

Author: FabioLissi

mempalace/cli.py

@@ -166,6 +166,119 @@ def cmd_status(args):
     status(palace_path=palace_path)
 
 
+def _extract_source_files(palace_path: str) -> set:
+    """Extract all unique source_file paths from palace metadata."""
+    from .palace import get_collection, iter_all_metadatas
+
+    try:
+        col = get_collection(palace_path, force=True)
+    except Exception:
+        return set()
+
+    sources = set()
+    for meta in iter_all_metadatas(col):
+        sf = meta.get("source_file")
+        if sf:
+            sources.add(sf)
+    return sources
+
+
+def cmd_remine(args):
+    """Re-mine palace with the currently configured embedding model."""
+    from .config import get_embedding_model_name
+
+    palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
+
+    if not os.path.isdir(palace_path):
+        print(f"\n  No palace found at {palace_path}")
+        return
+
+    print(f"\n{'=' * 55}")
+    print("  MemPalace Re-mine")
+    print(f"{'=' * 55}\n")
+    print(f"  Palace: {palace_path}")
+    print(f"  Target model: {get_embedding_model_name()}")
+
+    # Step 1: Extract source files
+    print("\n  Extracting source file paths from existing drawers...")
+    sources = _extract_source_files(palace_path)
+
+    if not sources:
+        print("  No drawers found. Nothing to re-mine.")
+        return
+
+    # Step 2: Partition into existing vs missing
+    existing = {s for s in sources if os.path.isfile(s)}
+    missing = sources - existing
+
+    print(f"  Found {len(sources)} unique source files.")
+    print(f"    Still exist: {len(existing)}")
+    print(f"    Missing:     {len(missing)}")
+
+    if missing:
+        print("\n  Missing files (will be skipped):")
+        for f in sorted(missing)[:20]:
+            print(f"    - {f}")
+        if len(missing) > 20:
+            print(f"    ... and {len(missing) - 20} more")
+
+    if not existing:
+        print("\n  No source files found on disk. Nothing to re-mine.")
+        return
+
+    if args.dry_run:
+        print(f"\n  (dry run — would re-mine {len(existing)} files)")
+        return
+
+    # Step 3: Backup palace before destructive operation
+    import shutil
+    import chromadb
+
+    backup_path = palace_path.rstrip(os.sep) + ".pre-remine-backup"
+    if os.path.exists(backup_path):
+        shutil.rmtree(backup_path)
+    print(f"\n  Backing up to {backup_path}...")
+    shutil.copytree(palace_path, backup_path)
+
+    # Step 4: Drop and re-create
+    print("  Dropping existing collection...")
+    client = chromadb.PersistentClient(path=palace_path)
+    client.delete_collection("mempalace_drawers")
+
+    # Step 5: Re-mine with exact file list (not directory scan)
+    # NOTE: Files originally mined with --mode convos will be re-mined
+    # as project files (fixed-size chunks instead of exchange-pair chunks).
+    # The vectors will be correct for the new model, but chunk boundaries
+    # will differ. A future improvement could store the original mining
+    # mode in drawer metadata to preserve it across re-mines.
+    print("  Re-mining...")
+
+    from .miner import mine
+
+    # Group files by parent dir (mine() needs a project_dir for config)
+    from collections import defaultdict
+
+    dir_files = defaultdict(list)
+    for f in sorted(existing):
+        dir_files[os.path.dirname(f)].append(f)
+
+    for source_dir, file_list in sorted(dir_files.items()):
+        if os.path.isdir(source_dir):
+            print(f"\n  Mining: {source_dir} ({len(file_list)} files)")
+            mine(
+                project_dir=source_dir,
+                palace_path=palace_path,
+                source_files=file_list,
+            )
+
+    print(f"\n{'=' * 55}")
+    print(f"  Re-mine complete. Model: {get_embedding_model_name()}")
+    if missing:
+        print(f"  Skipped {len(missing)} missing source files.")
+    print(f"  Backup saved at {backup_path}")
+    print(f"{'=' * 55}\n")
+
+
 def cmd_repair(args):
     """Rebuild palace vector index from SQLite metadata."""
     import chromadb
@@ -551,6 +664,15 @@ def main():
 
     sub.add_parser("status", help="Show what's been filed")
 
+    # re-mine
+    p_remine = sub.add_parser(
+        "re-mine",
+        help="Re-mine palace with the currently configured embedding model",
+    )
+    p_remine.add_argument(
+        "--dry-run", action="store_true", help="Show what would be re-mined without doing it"
+    )
+
     args = parser.parse_args()
 
     if not args.command:
@@ -585,6 +707,7 @@ def main():
         "repair": cmd_repair,
         "migrate": cmd_migrate,
         "status": cmd_status,
+        "re-mine": cmd_remine,
     }
     dispatch[args.command](args)
 

mempalace/miner.py

@@ -544,20 +544,29 @@ def mine(
     dry_run: bool = False,
     respect_gitignore: bool = True,
     include_ignored: list = None,
+    source_files: list = None,
 ):
-    """Mine a project directory into the palace."""
+    """Mine a project directory into the palace.
+
+    If *source_files* is provided, only those files are mined (skipping
+    the directory scan).  This is used by ``mempalace re-mine`` to
+    re-embed exactly the files that were in the palace before.
+    """
 
     project_path = Path(project_dir).expanduser().resolve()
     config = load_config(project_dir)
 
     wing = wing_override or config["wing"]
     rooms = config.get("rooms", [{"name": "general", "description": "All project files"}])
 
-    files = scan_project(
-        project_dir,
-        respect_gitignore=respect_gitignore,
-        include_ignored=include_ignored,
-    )
+    if source_files is not None:
+        files = [Path(f) for f in source_files if os.path.isfile(f)]
+    else:
+        files = scan_project(
+            project_dir,
+            respect_gitignore=respect_gitignore,
+            include_ignored=include_ignored,
+        )
     if limit > 0:
         files = files[:limit]
 

tests/test_remine.py

@@ -0,0 +1,105 @@
+"""Tests for the mempalace re-mine command."""
+
+import os
+from unittest.mock import MagicMock
+
+import pytest
+
+from mempalace.palace import get_collection
+
+
+@pytest.fixture(autouse=True)
+def reset_embedding_cache():
+    import mempalace.config as cfg_mod
+    cfg_mod._embedding_function = None
+    cfg_mod._embedding_function_resolved = False
+    yield
+    cfg_mod._embedding_function = None
+    cfg_mod._embedding_function_resolved = False
+
+
+@pytest.fixture
+def populated_palace(tmp_path, monkeypatch):
+    monkeypatch.delenv("MEMPALACE_EMBEDDING_MODEL", raising=False)
+    monkeypatch.delenv("MEMPALACE_EMBEDDING_DEVICE", raising=False)
+
+    palace_path = str(tmp_path / "palace")
+    source_dir = tmp_path / "source"
+    source_dir.mkdir()
+
+    (source_dir / "file1.txt").write_text("Hello world content for file one")
+    (source_dir / "file2.txt").write_text("Second file with different content")
+
+    col = get_collection(palace_path)
+    col.add(
+        documents=["Hello world content for file one"],
+        ids=["drawer-1"],
+        metadatas=[{
+            "wing": "test",
+            "room": "general",
+            "source_file": str(source_dir / "file1.txt"),
+        }],
+    )
+    col.add(
+        documents=["Second file with different content"],
+        ids=["drawer-2"],
+        metadatas=[{
+            "wing": "test",
+            "room": "general",
+            "source_file": str(source_dir / "file2.txt"),
+        }],
+    )
+
+    return {"palace_path": palace_path, "source_dir": source_dir, "col": col}
+
+
+class TestRemineExtractSources:
+    def test_extracts_unique_source_files(self, populated_palace):
+        from mempalace.cli import _extract_source_files
+        sources = _extract_source_files(populated_palace["palace_path"])
+        assert len(sources) == 2
+        assert str(populated_palace["source_dir"] / "file1.txt") in sources
+        assert str(populated_palace["source_dir"] / "file2.txt") in sources
+
+    def test_empty_palace_returns_empty(self, tmp_path, monkeypatch):
+        monkeypatch.delenv("MEMPALACE_EMBEDDING_MODEL", raising=False)
+        monkeypatch.delenv("MEMPALACE_EMBEDDING_DEVICE", raising=False)
+        palace_path = str(tmp_path / "palace")
+        get_collection(palace_path)
+        from mempalace.cli import _extract_source_files
+        sources = _extract_source_files(palace_path)
+        assert sources == set()
+
+
+class TestRemineDryRun:
+    def test_dry_run_reports_counts(self, populated_palace, capsys):
+        from mempalace.cli import cmd_remine
+        args = MagicMock()
+        args.palace = populated_palace["palace_path"]
+        args.dry_run = True
+        cmd_remine(args)
+        output = capsys.readouterr().out
+        assert "2" in output
+        assert "dry run" in output.lower()
+
+    def test_dry_run_does_not_drop_collection(self, populated_palace):
+        from mempalace.cli import cmd_remine
+        args = MagicMock()
+        args.palace = populated_palace["palace_path"]
+        args.dry_run = True
+        cmd_remine(args)
+        col = get_collection(populated_palace["palace_path"])
+        assert col.count() == 2
+
+
+class TestRemineMissingFiles:
+    def test_reports_missing_files(self, populated_palace, capsys):
+        from mempalace.cli import cmd_remine
+        os.remove(populated_palace["source_dir"] / "file2.txt")
+        args = MagicMock()
+        args.palace = populated_palace["palace_path"]
+        args.dry_run = True
+        cmd_remine(args)
+        output = capsys.readouterr().out
+        assert "1" in output  # 1 still exists
+        assert "missing" in output.lower() or "Missing" in output