feat(cli): add clean subcommand to remove wings and rooms

mempalace/cli.py

@@ -18,6 +18,9 @@
     mempalace wake-up                     Show L0 + L1 wake-up context
     mempalace wake-up --wing my_app       Wake-up for a specific project
     mempalace status                      Show what's been filed
+    mempalace clean --wing my_app         Remove all drawers in a wing
+    mempalace clean --wing my_app --room costs
+                                          Remove drawers in one room of a wing
 
 Examples:
     mempalace init ~/projects/my_app
@@ -955,6 +958,129 @@ def cmd_repair(args):
     print(f"\n{'=' * 55}\n")
 
 
+def cmd_clean(args):
+    """Remove drawers from a wing or a specific room within a wing."""
+    import chromadb
+    from .palace import find_drawer_ids
+
+    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}")
+        sys.exit(1)
+
+    wing = args.wing
+    room = args.room
+    scope = f"wing '{wing}'" + (f", room '{room}'" if room else "")
+
+    try:
+        client = chromadb.PersistentClient(path=palace_path)
+    except Exception as e:
+        print(f"\n  Error opening palace: {e}")
+        sys.exit(1)
+
+    try:
+        drawers_col = client.get_collection("mempalace_drawers")
+    except Exception:
+        print(f"\n  No drawers collection in {palace_path}")
+        sys.exit(1)
+
+    try:
+        compressed_col = client.get_collection("mempalace_compressed")
+    except Exception:
+        compressed_col = None
+
+    # Fetch matching IDs once per collection. These lists drive both the
+    # preview counts and the subsequent delete, so no second scan is needed.
+    drawer_ids = find_drawer_ids(drawers_col, wing, room)
+    compressed_ids = find_drawer_ids(compressed_col, wing, room) if compressed_col else []
+    drawer_count = len(drawer_ids)
+    compressed_count = len(compressed_ids)
+
+    print(f"\n{'=' * 55}")
+    print("  MemPalace Clean")
+    print(f"{'=' * 55}\n")
+    print(f"  Palace: {palace_path}")
+    print(f"  Target: {scope}")
+    print(f"  Drawers to remove:    {drawer_count}")
+    if compressed_col is not None:
+        print(f"  Compressed to remove: {compressed_count}")
+
+    if drawer_count == 0 and compressed_count == 0:
+        print("\n  Nothing to clean.")
+        print(
+            "\n  Note: the knowledge graph (~/.mempalace/knowledge_graph.sqlite3) "
+            "is not touched by clean."
+        )
+        return
+
+    if args.dry_run:
+        print("\n  (dry run — nothing removed)")
+        return
+
+    if not args.yes:
+        prompt = f"\n  Delete {drawer_count} drawer(s) from {scope}? [y/N] "
+        try:
+            answer = input(prompt).strip().lower()
+        except EOFError:
+            answer = ""
+        if answer not in ("y", "yes"):
+            print("  Cancelled.")
+            return
+
+    # Delete by ID — no additional scans. Guarded against empty lists
+    # because some ChromaDB versions reject delete() with no ids/where.
+    if drawer_ids:
+        drawers_col.delete(ids=drawer_ids)
+    if compressed_ids:
+        compressed_col.delete(ids=compressed_ids)
+
+    removed_drawers = drawer_count
+    removed_compressed = compressed_count
+
+    # Log the operation to the WAL for audit trail
+    try:
+        import json
+        from datetime import datetime
+        from pathlib import Path
+
+        wal_dir = Path(os.path.expanduser("~/.mempalace/wal"))
+        wal_dir.mkdir(parents=True, exist_ok=True)
+        try:
+            wal_dir.chmod(0o700)
+        except (OSError, NotImplementedError):
+            pass
+        wal_file = wal_dir / "write_log.jsonl"
+        entry = {
+            "timestamp": datetime.now().isoformat(),
+            "operation": "clean_room" if room else "clean_wing",
+            "params": {
+                "palace_path": palace_path,
+                "wing": wing,
+                "room": room,
+                "drawers_deleted": removed_drawers,
+                "compressed_deleted": removed_compressed,
+            },
+        }
+        with open(wal_file, "a", encoding="utf-8") as f:
+            f.write(json.dumps(entry) + "\n")
+        try:
+            wal_file.chmod(0o600)
+        except (OSError, NotImplementedError):
+            pass
+    except Exception:
+        pass
+
+    print(f"\n  Removed {removed_drawers} drawer(s) from {scope}.")
+    if compressed_col is not None:
+        print(f"  Removed {removed_compressed} compressed drawer(s).")
+    print(
+        "\n  Note: the knowledge graph (~/.mempalace/knowledge_graph.sqlite3) "
+        "is not touched by clean."
+    )
+    print(f"\n{'=' * 55}\n")
+
+
 def cmd_hook(args):
     """Run hook logic: reads JSON from stdin, outputs JSON to stdout."""
     from .hooks_cli import run_hook
@@ -1155,11 +1281,12 @@ def main():
 
     version_label = f"MemPalace {__version__}"
     parser = argparse.ArgumentParser(
-        description="MemPalace — Give your AI a memory. No API key required.",
+        description=f"MemPalace v{__version__} — Give your AI a memory. No API key required.",
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog=f"{version_label}\n\n{__doc__}",
     )
     parser.add_argument(
+        "-V",
         "--version",
         action="version",
         version=version_label,
@@ -1392,6 +1519,32 @@ def main():
         help="Only split files containing at least N sessions (default: 2)",
     )
 
+    # clean
+    p_clean = sub.add_parser(
+        "clean",
+        help="Remove drawers from a wing or a specific room",
+        description=(
+            "Remove drawers from the palace. Scope to a whole wing or a specific "
+            "room within a wing. The knowledge graph is not touched."
+        ),
+    )
+    p_clean.add_argument("--wing", required=True, help="Wing to clean (required)")
+    p_clean.add_argument(
+        "--room",
+        default=None,
+        help="Room to clean (optional — limits deletion to this room within the wing)",
+    )
+    p_clean.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be removed without deleting",
+    )
+    p_clean.add_argument(
+        "--yes",
+        action="store_true",
+        help="Skip the interactive confirmation prompt",
+    )
+
     # hook
     p_hook = sub.add_parser(
         "hook",
@@ -1562,6 +1715,7 @@ def main():
         "repair-status": cmd_repair_status,
         "migrate": cmd_migrate,
         "status": cmd_status,
+        "clean": cmd_clean,
     }
     dispatch[args.command](args)
 

mempalace/palace.py

@@ -602,6 +602,33 @@ def mine_palace_lock(palace_path: str):
 mine_global_lock = mine_palace_lock
 
 
+def _build_where(wing: str, room: str | None = None) -> dict:
+    """Build a ChromaDB where filter for wing/room scoped operations."""
+    if room:
+        return {"$and": [{"wing": wing}, {"room": room}]}
+    return {"wing": wing}
+
+
+def find_drawer_ids(collection, wing: str, room: str | None = None) -> list[str]:
+    """Return drawer IDs matching a wing (and optional room) in one scan.
+
+    Uses ``include=[]`` so ChromaDB only fetches IDs — no documents,
+    embeddings, or metadatas. Callers that need both a count and a
+    subsequent delete should call this once and reuse the result to
+    avoid a second scan.
+    """
+    try:
+        results = collection.get(where=_build_where(wing, room), include=[])
+        return list(results.get("ids", []))
+    except Exception:
+        return []
+
+
+def count_drawers(collection, wing: str, room: str | None = None) -> int:
+    """Count drawers in a wing (and optionally a specific room)."""
+    return len(find_drawer_ids(collection, wing, room))
+
+
 def _metadata_matches_extract_mode(meta: dict, extract_mode: Optional[str]) -> bool:
     if extract_mode is None:
         return True