fix(cli): add source-scoped unmine command

Author: MichaelWave369

mempalace/cli.py

@@ -17,6 +17,7 @@
     mempalace mine <dir> --mode convos    Mine conversation exports
     mempalace mine <dir> --mode extract   Mine binary office documents (PDF/DOCX/etc.)
     mempalace mine <source> --source NAME Mine through a registered source adapter
+    mempalace unmine <source-file>         Remove filed data for one exact source (dry-run first)
     mempalace search "query"              Find anything, exact words
     mempalace mcp                         Show MCP setup command
     mempalace wake-up                     Show L0 + L1 wake-up context
@@ -1160,6 +1161,63 @@ def cmd_sync(args):
     print(f"\n{'=' * 55}\n")
 
 
+def cmd_unmine(args):
+    """Remove all filed data associated with one exact source path."""
+    palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
+    source_file = os.path.abspath(os.path.expanduser(args.source_file))
+
+    if not os.path.isdir(palace_path):
+        print(f"\n  No palace found at {palace_path}")
+        return
+
+    from .palace import MineAlreadyRunning
+    from .sync import unmine_source
+    from .wal import _wal_log
+
+    print(f"\n{'=' * 55}")
+    print("  MemPalace Unmine -- Source-scoped removal")
+    print(f"{'=' * 55}")
+    print(f"  Palace:  {palace_path}")
+    print(f"  Source:  {source_file}")
+    print(f"  Wing:    {args.wing or 'all wings'}")
+    print(f"  Mode:    {'DRY RUN (no deletions)' if args.dry_run else 'APPLY'}")
+    print(f"{'-' * 55}\n")
+
+    try:
+        report = unmine_source(
+            palace_path=palace_path,
+            source_file=source_file,
+            wing=args.wing,
+            dry_run=args.dry_run,
+            wal_log=_wal_log,
+        )
+    except MineAlreadyRunning as exc:
+        print(f"mempalace: {exc}", file=sys.stderr)
+        sys.exit(1)
+    except ValueError as exc:
+        print(f"mempalace: {exc}", file=sys.stderr)
+        sys.exit(2)
+    except Exception as exc:
+        print(f"mempalace: unmine failed: {exc}", file=sys.stderr)
+        sys.exit(1)
+
+    wings = ", ".join(report["affected_wings"]) if report["affected_wings"] else "none"
+    print(f"  Matched drawers: {report['matched_drawers']}")
+    print(f"  Matched closets: {report['matched_closets']}")
+    print(f"  Affected wings:  {wings}")
+
+    if args.dry_run:
+        if report["matched_drawers"] or report["matched_closets"]:
+            print("\n  Re-run with --apply to remove exactly this source.")
+        else:
+            print("\n  Nothing filed from that source path.")
+    else:
+        print(
+            f"\n  Removed {report['removed_drawers']} drawers, {report['removed_closets']} closets."
+        )
+    print(f"\n{'=' * 55}\n")
+
+
 def _submit_daemon_cli_job(kind: str, payload: dict, args, *, background: bool) -> None:
     palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
     backend = _backend_arg(args)
@@ -2676,6 +2734,27 @@ def main():
         help="With --daemon, return a job id immediately instead of waiting",
     )
 
+    # unmine
+    p_unmine = sub.add_parser(
+        "unmine",
+        help="Remove all filed data for one exact source path (dry-run by default)",
+    )
+    p_unmine.add_argument("source_file", help="Source file path stored on the filed drawers")
+    p_unmine.add_argument("--wing", default=None, help="Limit removal to one wing")
+    p_unmine.add_argument(
+        "--dry-run",
+        dest="dry_run",
+        action="store_true",
+        default=True,
+        help="Preview only (default)",
+    )
+    p_unmine.add_argument(
+        "--apply",
+        dest="dry_run",
+        action="store_false",
+        help="Actually remove matching drawers and closets",
+    )
+
     # search
     p_search = sub.add_parser("search", help="Find anything, exact words")
     p_search.add_argument("query", help="What to search for")
@@ -3165,6 +3244,7 @@ def _add_logstream_filters(p):
         "search": cmd_search,
         "sweep": cmd_sweep,
         "sync": cmd_sync,
+        "unmine": cmd_unmine,
         "mcp": cmd_mcp,
         "serve": cmd_serve,
         "compress": cmd_compress,

mempalace/sync.py

@@ -42,6 +42,17 @@ class SyncReport(TypedDict):
     by_source: dict[str, int]
 
 
+class UnmineReport(TypedDict):
+    source_file: str
+    wing: Optional[str]
+    matched_drawers: int
+    matched_closets: int
+    removed_drawers: int
+    removed_closets: int
+    affected_wings: list[str]
+    dry_run: bool
+
+
 def _resolve_project_root(source_file: Path, project_roots: list) -> Optional[Path]:
     """Return the longest project_root that source_file lives under.
 
@@ -184,22 +195,134 @@ def _normalize_project_dirs(project_dirs) -> list:
     return sorted(resolved, key=lambda p: (-len(str(p)), str(p)))
 
 
-def _delete_in_batches(col, ids: list, batch_size: int, wal_log: Optional[Callable]):
-    """Delete drawer IDs in batches, optionally logging each batch to WAL."""
+def _delete_in_batches(
+    col,
+    ids: list,
+    batch_size: int,
+    wal_log: Optional[Callable],
+    *,
+    operation: str = "sync_prune",
+    params: Optional[dict] = None,
+):
+    """Delete IDs in batches, optionally logging each batch to WAL."""
     deleted = 0
     for i in range(0, len(ids), batch_size):
         chunk = ids[i : i + batch_size]
         col.delete(ids=chunk)
         deleted += len(chunk)
         if wal_log is not None:
+            wal_params = dict(params or {})
+            wal_params["first_id"] = chunk[0]
             wal_log(
-                "sync_prune",
-                {"first_id": chunk[0]},
+                operation,
+                wal_params,
                 {"removed_count": len(chunk)},
             )
     return deleted
 
 
+def _source_where(source_file: str, wing: Optional[str]):
+    if wing:
+        return {"$and": [{"source_file": source_file}, {"wing": wing}]}
+    return {"source_file": source_file}
+
+
+def _matching_source_rows(col, source_file: str, wing: Optional[str]) -> tuple[list[str], set[str]]:
+    """Return IDs and observed wings for one exact source-file match."""
+    ids: list[str] = []
+    wings: set[str] = set()
+    offset = 0
+    where = _source_where(source_file, wing)
+    while True:
+        batch = col.get(
+            where=where,
+            limit=_BATCH,
+            offset=offset,
+            include=["metadatas"],
+        )
+        batch_ids = batch.get("ids") or []
+        metadatas = batch.get("metadatas") or []
+        if not batch_ids:
+            break
+        ids.extend(batch_ids)
+        for meta in metadatas:
+            value = (meta or {}).get("wing")
+            if isinstance(value, str) and value:
+                wings.add(value)
+        offset += len(batch_ids)
+    return ids, wings
+
+
+def unmine_source(
+    palace_path: str,
+    source_file: str,
+    wing: Optional[str] = None,
+    dry_run: bool = True,
+    batch_size: int = _BATCH,
+    wal_log: Optional[Callable] = None,
+) -> UnmineReport:
+    """Remove every drawer and closet filed from one exact source path.
+
+    Dry-run is the default. The operation is source-scoped rather than
+    content-scoped on purpose: it gives operators a precise recovery path
+    for accidental/duplicate mines without defining a global policy for when
+    repeated text is semantically redundant. ``wing`` can further narrow a
+    source that was intentionally filed into more than one wing.
+
+    Registry sentinels are ordinary drawer rows carrying ``source_file`` and
+    are removed too, so a later mine of the source is not falsely skipped.
+    """
+    if not source_file:
+        raise ValueError("source_file must be a non-empty path")
+
+    with mine_palace_lock(palace_path):
+        col = get_collection(palace_path, create=False)
+        drawer_ids, affected_wings = _matching_source_rows(col, source_file, wing)
+
+        closets_col = None
+        closet_ids: list[str] = []
+        try:
+            closets_col = get_closets_collection(palace_path, create=False)
+        except Exception as exc:
+            logger.debug("Unmine closet lookup skipped (collection unavailable): %s", exc)
+        if closets_col is not None:
+            closet_ids, closet_wings = _matching_source_rows(closets_col, source_file, wing)
+            affected_wings.update(closet_wings)
+
+        report: UnmineReport = {
+            "source_file": source_file,
+            "wing": wing,
+            "matched_drawers": len(drawer_ids),
+            "matched_closets": len(closet_ids),
+            "removed_drawers": 0,
+            "removed_closets": 0,
+            "affected_wings": sorted(affected_wings),
+            "dry_run": dry_run,
+        }
+        if dry_run:
+            return report
+
+        wal_params = {"source_file": source_file}
+        if wing:
+            wal_params["wing"] = wing
+        report["removed_drawers"] = _delete_in_batches(
+            col,
+            drawer_ids,
+            batch_size,
+            wal_log,
+            operation="unmine_source",
+            params=wal_params,
+        )
+        if closets_col is not None:
+            report["removed_closets"] = _delete_in_batches(
+                closets_col,
+                closet_ids,
+                batch_size,
+                None,
+            )
+        return report
+
+
 def sync_palace(
     palace_path: str,
     project_dirs: Optional[list] = None,
@@ -317,5 +440,7 @@ def sync_palace(
 __all__ = [
     "MineAlreadyRunning",
     "SyncReport",
+    "UnmineReport",
     "sync_palace",
+    "unmine_source",
 ]