Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 80 additions & 0 deletions mempalace/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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,
Expand Down
133 changes: 129 additions & 4 deletions mempalace/sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -317,5 +440,7 @@ def sync_palace(
__all__ = [
"MineAlreadyRunning",
"SyncReport",
"UnmineReport",
"sync_palace",
"unmine_source",
]
Loading