"""``muse rm`` — remove files from tracking and optionally from disk. Mirrors ``git rm`` exactly. Files removed with ``muse rm`` are staged for deletion in the next commit; the commit then drops them from the snapshot. Usage:: muse rm [ …] # stage deletion + delete from disk muse rm --cached [ …] # stage deletion only (keep on disk) muse rm -r # recursive — required when is a dir muse rm -n / --dry-run # preview what would be removed muse rm -f / --force # override safety checks muse rm --json # machine-readable output Staged changes after ``muse rm`` --------------------------------- ``muse status`` will show removed files as "D" (deleted / staged for deletion). Run ``muse commit`` to record the removal in the next snapshot. To un-do a staged removal before committing:: muse code reset # unstage the deletion; file is back in next commit Safety model ------------ By default ``muse rm`` refuses to remove: - **Modified files** — the on-disk content differs from the HEAD snapshot. Pass ``--force`` / ``-f`` to override. - **Staged-but-uncommitted additions** — the file was added with ``muse code add`` but never committed. Pass ``--force`` / ``-f`` to remove. - **Directories** — pass ``-r`` (recursive) to allow removing all tracked files under a directory. The ``--cached`` flag removes only the stage entry (and marks the file deleted in the next commit) without touching the on-disk copy. This is the correct way to untrack a file while keeping it in the working tree — e.g. when the file should be listed in ``.museignore`` instead. JSON output schema:: { "status": "removed" | "dry_run" | "nothing_to_remove", "removed": ["relative/path/a.txt", ...], "cached": true | false, "dry_run": true | false, "count": N } Exit codes:: 0 — one or more files removed (or would be removed in dry-run) 1 — user error: file not tracked, directory without -r, safety check failed 2 — not a Muse repository 3 — I/O error during deletion """ from __future__ import annotations import argparse import logging import pathlib import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.object_store import write_object_from_path from muse.core.repo import require_repo from muse.core.snapshot import hash_file from muse.core.store import Manifest, get_head_commit_id, read_commit, read_current_branch, read_snapshot from muse.core.validation import sanitize_display from muse.plugins.code.stage import ( StagedEntry, StagedFileMap, make_entry, read_stage, write_stage, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # JSON wire format # --------------------------------------------------------------------------- class _RmResultJson(TypedDict): """JSON output for ``muse rm``.""" status: str # "removed" | "dry_run" | "nothing_to_remove" removed: list[str] cached: bool dry_run: bool count: int # --------------------------------------------------------------------------- # Private helpers # --------------------------------------------------------------------------- def _head_manifest(root: pathlib.Path) -> Manifest: """Return the manifest from the current HEAD commit, or ``{}`` if none. Returns an empty dict for repositories with no commits yet. Never raises — callers treat an empty manifest as "nothing committed." """ try: branch = read_current_branch(root) commit_id = get_head_commit_id(root, branch) if not commit_id: return {} commit = read_commit(root, commit_id) if commit is None: return {} snap = read_snapshot(root, commit.snapshot_id) return dict(snap.manifest) if snap else {} except Exception: return {} def _collect_targets( root: pathlib.Path, raw_paths: list[str], recursive: bool, head_manifest: Manifest, stage: StagedFileMap, ) -> list[str]: """Expand *raw_paths* into a sorted list of tracked relative paths. Directories are only expanded when *recursive* is ``True``. A path that is neither tracked in HEAD nor staged raises ``SystemExit(USER_ERROR)`` immediately. Returns relative POSIX paths (e.g. ``"src/auth.py"``). """ result: list[str] = [] for raw in raw_paths: p = pathlib.Path(raw) if not p.is_absolute(): # Prefer CWD-relative resolution (normal real-world usage). # Fall back to root-relative when CWD is outside the repository # (common in tests and agent pipelines using -C or MUSE_REPO_ROOT). cwd_candidate = (pathlib.Path.cwd() / p).resolve() try: cwd_candidate.relative_to(root.resolve()) abs_target = cwd_candidate except ValueError: abs_target = (root / p).resolve() else: abs_target = p.resolve() # Make it relative to root. try: rel = abs_target.relative_to(root.resolve()) except ValueError: print( f"❌ fatal: '{sanitize_display(raw)}' is outside the repository root.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) rel_posix = rel.as_posix() if abs_target.is_dir(): if not recursive: print( f"❌ fatal: not removing '{sanitize_display(rel_posix)}' " f"recursively without -r.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Collect all tracked files under this directory. dir_files = [ p for p in list(head_manifest.keys()) + [ k for k, v in stage.items() if v["mode"] != "D" ] if (p == rel_posix or p.startswith(rel_posix + "/")) and p not in result ] if not dir_files: print( f"❌ fatal: pathspec '{sanitize_display(rel_posix)}' did not match " f"any tracked files.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) result.extend(dir_files) else: # Must be tracked in HEAD or staged (as A or M, not already D). in_head = rel_posix in head_manifest staged_entry = stage.get(rel_posix) in_stage = staged_entry is not None and staged_entry["mode"] != "D" if not in_head and not in_stage: print( f"❌ fatal: pathspec '{sanitize_display(rel_posix)}' did not match " f"any tracked files.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if rel_posix not in result: result.append(rel_posix) return sorted(set(result)) def _check_safety( root: pathlib.Path, rel_path: str, head_manifest: Manifest, stage: StagedFileMap, force: bool, cached: bool, ) -> None: """Raise ``SystemExit(USER_ERROR)`` if removing *rel_path* is unsafe. Safety checks (skipped when *force* is ``True``): 1. **Staged-but-uncommitted addition** — the file is in stage with mode ``"A"`` (it was added with ``muse code add`` but never committed). Removing it would discard unstaged work that has never been recorded. 2. **Modified on disk vs HEAD** — the on-disk content has diverged from the HEAD snapshot (only checked when *cached* is ``False``, because ``--cached`` never touches the disk). Both checks are skipped entirely when *force* is ``True``. """ if force: return staged_entry = stage.get(rel_path) in_head = rel_path in head_manifest # Check 1: staged addition that was never committed. if staged_entry is not None and staged_entry["mode"] == "A" and not in_head: print( f"❌ error: '{sanitize_display(rel_path)}' has staged changes that have " f"never been committed.\n" f" Use --force to override, or 'muse code reset {rel_path}' to unstage.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Check 2: on-disk content differs from HEAD (only when we'd delete the file). if not cached and in_head: abs_path = root / rel_path if abs_path.exists(): try: disk_object_id = hash_file(abs_path) head_object_id = head_manifest[rel_path] if disk_object_id != head_object_id: print( f"❌ error: '{sanitize_display(rel_path)}' has local modifications.\n" f" Use --force to override, or --cached to keep the file on disk.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) except OSError: pass # File unreadable — proceed; deletion will surface the error. # --------------------------------------------------------------------------- # Command registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse rm`` subcommand.""" parser = subparsers.add_parser( "rm", help="Remove files from tracking (and optionally from disk).", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "paths", nargs="+", metavar="PATH", help="File(s) or directory(ies) to remove from tracking.", ) parser.add_argument( "--cached", action="store_true", help=( "Stage the deletion without deleting the file from disk. " "Use this to untrack a file while keeping it in the working tree." ), ) parser.add_argument( "-r", "--recursive", action="store_true", dest="recursive", help="Allow recursive removal when a directory is given.", ) parser.add_argument( "-f", "--force", action="store_true", help=( "Override safety checks — remove even if the file has local " "modifications or staged-but-uncommitted changes." ), ) parser.add_argument( "-n", "--dry-run", action="store_true", dest="dry_run", help="Preview what would be removed without making any changes.", ) parser.add_argument( "--json", action="store_true", dest="output_json", help="Emit machine-readable JSON on stdout.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Main handler # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Remove files from tracking and optionally from disk. Files are staged for deletion (mode ``"D"`` in the stage index) and will be absent from the snapshot produced by the next ``muse commit``. Behaviour per flag combination: - No flags: stage deletion + delete file from disk (requires file is unmodified vs HEAD, or ``--force``). - ``--cached``: stage deletion only; on-disk file is untouched. - ``--force`` / ``-f``: bypass the modified-file and staged-addition safety checks. Use when you're certain you want to discard the on-disk changes. - ``--dry-run`` / ``-n``: print what would be removed without writing anything to the stage or the disk. - ``-r``: required when a path argument is a directory; expands to all tracked files under that directory. - ``--json``: machine-readable JSON on stdout (always, including dry-run). JSON schema:: { "status": "removed" | "dry_run" | "nothing_to_remove", "removed": ["relative/path/a.txt", ...], "cached": true | false, "dry_run": true | false, "count": N } Exit codes:: 0 — success (files removed or would be removed in dry-run) 1 — user error: file not tracked, directory without -r, safety check failed 2 — not a Muse repository 3 — I/O error during deletion Examples:: muse rm song.txt # stage deletion + delete from disk muse rm --cached compiled/app.css # untrack without deleting muse rm -r --cached build/ # untrack entire build/ directory muse rm -f modified.py # force-remove locally modified file muse rm -n --json *.txt # dry-run, JSON output """ import json as _json cached: bool = args.cached recursive: bool = args.recursive force: bool = args.force dry_run: bool = args.dry_run output_json: bool = args.output_json root = require_repo() head_manifest = _head_manifest(root) stage = read_stage(root) # Expand path arguments into a sorted list of relative tracked paths. targets = _collect_targets( root, args.paths, recursive, head_manifest, stage ) if not targets: if output_json: result: _RmResultJson = _RmResultJson( status="nothing_to_remove", removed=[], cached=cached, dry_run=dry_run, count=0, ) print(_json.dumps(result)) else: print("Nothing to remove.") return # Run safety checks before touching anything. for rel_path in targets: _check_safety(root, rel_path, head_manifest, stage, force, cached) # At this point all targets are safe to remove. removed: list[str] = [] if dry_run: for rel_path in targets: if not output_json: print(f"[dry-run] Would remove: {sanitize_display(rel_path)}") removed.append(rel_path) else: new_stage: StagedFileMap = dict(stage) for rel_path in targets: staged_entry = stage.get(rel_path) in_head = rel_path in head_manifest if in_head: # File exists in the last commit → stage it as deleted. new_stage[rel_path] = make_entry(object_id="", mode="D") else: # File was only staged (mode "A") but never committed → # remove it from the stage entirely (un-track it). new_stage.pop(rel_path, None) # Delete from disk unless --cached. if not cached: abs_path = root / rel_path if abs_path.exists(): try: abs_path.unlink() except OSError as exc: print( f"❌ error: could not delete " f"'{sanitize_display(rel_path)}': {exc}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc removed.append(rel_path) if not output_json: verb = "rm (cached)" if cached else "rm" print(f"{verb}: {sanitize_display(rel_path)}") write_stage(root, new_stage) count = len(removed) if output_json: status = "dry_run" if dry_run else "removed" result = _RmResultJson( status=status, removed=removed, cached=cached, dry_run=dry_run, count=count, ) print(_json.dumps(result))