guard.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Shared CLI pre-flight guards for operations that modify the working tree. |
| 2 | |
| 3 | Any command that calls :func:`muse.core.workdir.apply_manifest` — merge, |
| 4 | checkout, reset, revert, cherry-pick, pull — should call |
| 5 | :func:`require_clean_workdir` first so users never lose uncommitted work |
| 6 | silently. The check mirrors Git's behaviour: modified/deleted tracked files |
| 7 | block the operation; newly untracked files are left alone. |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import json |
| 13 | import pathlib |
| 14 | import sys |
| 15 | |
| 16 | from muse.core.errors import ExitCode |
| 17 | from muse.core.snapshot import diff_workdir_vs_snapshot |
| 18 | from muse.core.store import get_head_snapshot_manifest, read_current_branch |
| 19 | from muse.core.validation import sanitize_display |
| 20 | |
| 21 | |
| 22 | def require_clean_workdir( |
| 23 | root: pathlib.Path, |
| 24 | operation: str, |
| 25 | *, |
| 26 | force: bool = False, |
| 27 | ) -> None: |
| 28 | """Abort with a friendly error if the working tree has uncommitted changes. |
| 29 | |
| 30 | Protects against silent data loss when *operation* (merge, checkout, |
| 31 | reset, …) is about to overwrite files via ``apply_manifest``. Pass |
| 32 | ``force=True`` to bypass the check (e.g. for ``--force`` flags). |
| 33 | |
| 34 | Only modified and deleted tracked files are considered dangerous — |
| 35 | brand-new files that have never been committed are left untouched by any |
| 36 | manifest-apply, so they are not flagged. |
| 37 | |
| 38 | Args: |
| 39 | root: Repository root (directory that contains ``.muse/``). |
| 40 | operation: Human-readable name of the operation (used in the message). |
| 41 | force: When ``True`` the guard is a no-op. |
| 42 | |
| 43 | Raises: |
| 44 | SystemExit: With :attr:`ExitCode.USER_ERROR` when dirty files exist. |
| 45 | """ |
| 46 | if force: |
| 47 | return |
| 48 | |
| 49 | branch = read_current_branch(root) |
| 50 | repo_id = str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 51 | head_manifest = get_head_snapshot_manifest(root, repo_id, branch) or {} |
| 52 | |
| 53 | # If the branch has no commits yet there is nothing to protect. |
| 54 | if not head_manifest: |
| 55 | return |
| 56 | |
| 57 | added, modified, deleted, _, _added_dirs, _deleted_dirs = diff_workdir_vs_snapshot(root, head_manifest) |
| 58 | |
| 59 | # Added paths are brand-new files never seen in a snapshot; apply_manifest |
| 60 | # won't touch them, so they are safe to ignore. Modified and deleted |
| 61 | # paths ARE in the snapshot and would be overwritten or removed. |
| 62 | dirty = modified | deleted |
| 63 | if not dirty: |
| 64 | return |
| 65 | |
| 66 | print( |
| 67 | f"❌ error: Your local changes to the following files would be " |
| 68 | f"overwritten by {sanitize_display(operation)}:", |
| 69 | file=sys.stderr, |
| 70 | ) |
| 71 | shown = sorted(dirty)[:10] |
| 72 | for path in shown: |
| 73 | print(f" {sanitize_display(path)}", file=sys.stderr) |
| 74 | if len(dirty) > 10: |
| 75 | print(f" … and {len(dirty) - 10} more", file=sys.stderr) |
| 76 | print( |
| 77 | f"\nRun one of:\n" |
| 78 | f" muse stash — shelve changes and retry\n" |
| 79 | f" muse {sanitize_display(operation)} --merge — carry changes forward (three-way merge)\n" |
| 80 | f" muse {sanitize_display(operation)} --force — discard changes\n" |
| 81 | f" muse {sanitize_display(operation)} --autostash — stash automatically, switch, pop on arrival", |
| 82 | file=sys.stderr, |
| 83 | ) |
| 84 | raise SystemExit(ExitCode.USER_ERROR) |