gabriel / muse public
clean.py python
352 lines 11.7 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """``muse clean`` — remove untracked files from the working tree.
2
3 Scans the working tree against HEAD's snapshot and removes files that are
4 not tracked in any commit. By design, ``--force`` is required to actually
5 delete files; without it the command behaves as a dry-run (equivalent to
6 passing ``-n``).
7
8 Usage::
9
10 muse clean -n # preview — show what would be removed
11 muse clean -f # delete untracked files
12 muse clean -f -d # also delete untracked directories
13 muse clean -f -x # also delete .museignore-excluded files
14 muse clean -f -d -x # everything untracked + ignored
15 muse clean -f --json # machine-readable result
16
17 All subcommands accept ``--json`` for machine-readable output::
18
19 {
20 "status": "clean" | "removed",
21 "removed": ["path/to/file.txt", ...],
22 "dirs_removed": ["path/to/dir", ...],
23 "count": N,
24 "dry_run": true | false
25 }
26
27 Exit codes::
28
29 0 — nothing to clean, or clean completed successfully
30 1 — untracked files exist but neither --force nor --dry-run given
31 2 — not a Muse repository
32 3 — I/O error during deletion
33
34 Security model::
35
36 Every candidate path returned by ``walk_workdir`` is validated to sit
37 inside the repository root before any deletion is attempted. Paths that
38 resolve outside the root are skipped with a warning; they cannot be
39 produced by ``walk_workdir`` under normal operation but the guard ensures
40 correctness even if the walker is extended in the future.
41
42 Directory removal only touches directories whose direct children were all
43 removed in the current run. The repository root, ``.muse/``, and any
44 path inside ``.muse/`` are unconditionally protected.
45 """
46
47 from __future__ import annotations
48
49 import argparse
50 import fnmatch
51 import logging
52 import pathlib
53 import sys
54 from typing import TypedDict
55
56 from muse.core.errors import ExitCode
57 from muse.core.ignore import load_ignore_config, resolve_patterns
58 from muse.core.repo import require_repo
59 from muse.core.snapshot import walk_workdir
60 from muse.core.store import get_head_commit_id, read_commit, read_current_branch, read_snapshot
61 from muse.core.validation import sanitize_display
62 from muse.plugins.registry import read_domain
63 from muse.core._types import Manifest
64
65 logger = logging.getLogger(__name__)
66
67
68 # ---------------------------------------------------------------------------
69 # JSON wire format
70 # ---------------------------------------------------------------------------
71
72
73 class _CleanResultJson(TypedDict):
74 """JSON output for ``muse clean``."""
75
76 status: str # "clean" | "removed"
77 removed: list[str]
78 dirs_removed: list[str]
79 count: int
80 dry_run: bool
81
82
83 # ---------------------------------------------------------------------------
84 # Helpers
85 # ---------------------------------------------------------------------------
86
87
88 def _is_ignored(path: str, patterns: list[str]) -> bool:
89 """Return ``True`` if *path* matches any ``.museignore`` pattern.
90
91 Uses last-match-wins semantics so that negation patterns (lines starting
92 with ``!``) can un-ignore previously matched paths.
93
94 Uses ``fnmatch.fnmatch`` against both the full relative path and the
95 filename component (``path.rsplit("/", 1)[-1]``) to mirror the behaviour
96 of ``.gitignore`` pattern matching.
97 """
98 result = False
99 basename = path.rsplit("/", 1)[-1]
100 for pat in patterns:
101 negate = pat.startswith("!")
102 effective = pat[1:] if negate else pat
103 if fnmatch.fnmatch(path, effective) or fnmatch.fnmatch(basename, effective):
104 result = not negate
105 return result
106
107
108 def _safe_to_delete(root: pathlib.Path, target: pathlib.Path) -> bool:
109 """Return ``True`` if *target* is safe to delete.
110
111 Guards:
112 - Target must resolve inside *root* (prevents path-traversal).
113 - Target must not be a directory (directories are handled separately).
114 - The ``.muse/`` subtree is unconditionally protected.
115 """
116 try:
117 target.resolve().relative_to(root.resolve())
118 except ValueError:
119 logger.warning(
120 "⚠️ Skipping %s — resolves outside repository root", target
121 )
122 return False
123 muse_dir = root / ".muse"
124 try:
125 target.relative_to(muse_dir)
126 logger.warning("⚠️ Skipping %s — inside .muse/", target)
127 return False
128 except ValueError:
129 pass
130 return True
131
132
133 def _safe_to_rmdir(root: pathlib.Path, d: pathlib.Path) -> bool:
134 """Return ``True`` if *d* is safe to remove as an empty directory.
135
136 Protects the repository root, ``.muse/``, and any path inside ``.muse/``.
137 """
138 if d == root:
139 return False
140 muse_dir = root / ".muse"
141 if d == muse_dir:
142 return False
143 try:
144 d.relative_to(muse_dir)
145 return False # inside .muse/
146 except ValueError:
147 pass
148 try:
149 d.resolve().relative_to(root.resolve())
150 except ValueError:
151 return False # outside root
152 return True
153
154
155 # ---------------------------------------------------------------------------
156 # Command registration
157 # ---------------------------------------------------------------------------
158
159
160 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
161 """Register the ``muse clean`` subcommand."""
162 parser = subparsers.add_parser(
163 "clean",
164 help="Remove untracked files from the working tree.",
165 description=__doc__,
166 formatter_class=argparse.RawDescriptionHelpFormatter,
167 )
168 parser.add_argument(
169 "-n", "--dry-run",
170 action="store_true",
171 dest="dry_run",
172 help="Preview — show what would be removed without deleting.",
173 )
174 parser.add_argument(
175 "-f", "--force",
176 action="store_true",
177 help="Delete untracked files (required unless --dry-run is passed).",
178 )
179 parser.add_argument(
180 "-x", "--include-ignored",
181 action="store_true",
182 dest="include_ignored",
183 help="Also delete .museignore-excluded files.",
184 )
185 parser.add_argument(
186 "-d", "--directories",
187 action="store_true",
188 help="Also remove empty untracked directories after file deletion.",
189 )
190 parser.add_argument(
191 "--json",
192 action="store_true",
193 dest="output_json",
194 help="Emit machine-readable JSON on stdout.",
195 )
196 parser.set_defaults(func=run)
197
198
199 # ---------------------------------------------------------------------------
200 # Main handler
201 # ---------------------------------------------------------------------------
202
203
204 def run(args: argparse.Namespace) -> None:
205 """Remove untracked files from the working tree.
206
207 Files not tracked in the HEAD snapshot are considered untracked.
208 ``--force`` is required to delete; without it (and without ``--dry-run``)
209 the command exits with USER_ERROR.
210
211 The working tree is walked with the same rules as ``muse commit``:
212 hidden files, symlinks, and ``.muse/`` itself are always excluded.
213
214 Every candidate path is validated to sit inside the repository root before
215 deletion. The ``.muse/`` subtree is unconditionally protected regardless
216 of what the working-tree walker returns.
217
218 With ``--json``::
219
220 {"status":"removed","removed":["a.txt","b.txt"],"dirs_removed":[],"count":2,"dry_run":false}
221
222 Examples::
223
224 muse clean -n # preview
225 muse clean -n --json # agent-friendly preview
226 muse clean -f # delete untracked files
227 muse clean -f -d -x # delete untracked + empty dirs + ignored files
228 muse clean -f --json # delete and emit JSON summary
229 """
230 dry_run: bool = args.dry_run
231 force: bool = args.force
232 include_ignored: bool = args.include_ignored
233 directories: bool = args.directories
234 output_json: bool = args.output_json
235
236 if not force and not dry_run:
237 print(
238 "⚠️ fatal: clean.requireForce is set to true.\n"
239 " Use --force to remove files, or --dry-run / -n to preview.",
240 file=sys.stderr,
241 )
242 raise SystemExit(ExitCode.USER_ERROR)
243
244 root = require_repo()
245 branch = read_current_branch(root)
246 domain = read_domain(root)
247
248 # Build committed manifest (empty for a branch with no commits yet).
249 committed: Manifest = {}
250 head_commit_id = get_head_commit_id(root, branch)
251 if head_commit_id:
252 commit = read_commit(root, head_commit_id)
253 if commit:
254 snap = read_snapshot(root, commit.snapshot_id)
255 if snap:
256 committed = snap.manifest
257
258 # Build current workdir manifest.
259 current = walk_workdir(root)
260
261 # Load ignore patterns; warn on failure but continue.
262 ignored_patterns: list[str] = []
263 if not include_ignored:
264 try:
265 ignore_cfg = load_ignore_config(root)
266 ignored_patterns = resolve_patterns(ignore_cfg, domain)
267 except OSError as exc:
268 logger.warning("⚠️ Could not load ignore config: %s", exc)
269
270 # Collect untracked paths.
271 untracked: list[str] = []
272 for rel_path in sorted(current):
273 if rel_path in committed:
274 continue
275 if not include_ignored and _is_ignored(rel_path, ignored_patterns):
276 continue
277 untracked.append(rel_path)
278
279 if not untracked:
280 if output_json:
281 result = _CleanResultJson(
282 status="clean",
283 removed=[],
284 dirs_removed=[],
285 count=0,
286 dry_run=dry_run,
287 )
288 print(__import__("json").dumps(result))
289 else:
290 print("Nothing to clean.")
291 return
292
293 prefix = "[dry-run] " if dry_run else ""
294 verb = "Would remove" if dry_run else "Removing"
295
296 removed_files: list[str] = []
297 removed_dirs_list: list[str] = []
298 candidate_dirs: set[pathlib.Path] = set()
299
300 for rel_path in untracked:
301 target = root / rel_path
302 if not output_json:
303 print(f"{prefix}{verb}: {sanitize_display(rel_path)}")
304 if not dry_run:
305 if not _safe_to_delete(root, target):
306 continue
307 try:
308 target.unlink(missing_ok=True)
309 removed_files.append(rel_path)
310 if directories:
311 candidate_dirs.add(target.parent)
312 except OSError as exc:
313 print(
314 f"❌ Could not remove {sanitize_display(rel_path)}: {exc}",
315 file=sys.stderr,
316 )
317 raise SystemExit(ExitCode.INTERNAL_ERROR) from exc
318 else:
319 removed_files.append(rel_path)
320
321 # Remove empty directories (bottom-up), protected by _safe_to_rmdir.
322 if not dry_run and directories:
323 for d in sorted(candidate_dirs, key=lambda p: len(p.parts), reverse=True):
324 if not _safe_to_rmdir(root, d):
325 continue
326 try:
327 if d.is_dir() and not any(d.iterdir()):
328 d.rmdir()
329 rel_dir = str(d.relative_to(root))
330 removed_dirs_list.append(rel_dir)
331 if not output_json:
332 print(f"Removing directory: {sanitize_display(rel_dir)}")
333 except OSError:
334 pass
335
336 count = len(removed_files)
337 if output_json:
338 import json as _json
339
340 result = _CleanResultJson(
341 status="removed" if not dry_run else "clean",
342 removed=removed_files,
343 dirs_removed=removed_dirs_list,
344 count=count,
345 dry_run=dry_run,
346 )
347 print(_json.dumps(result))
348 else:
349 if dry_run:
350 print(f"\n{count} untracked file(s) would be removed.")
351 else:
352 print(f"\n✅ Removed {count} untracked file(s).")
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 101 days ago