gabriel / muse public
ls_tree.py python
443 lines 14.3 KB
Raw
sha256:2fa778aba8ab0ec15295b8624c6480a573482ffc9c206a6d9546f1c41d2c2b7b feat: supercharge muse blame + remove --porcelain everywhere Human patch 164 days ago
1 """``muse ls-tree`` — list the contents of a tree object at a given ref.
2
3 Displays the files and synthetic directory entries recorded in a commit's
4 snapshot manifest. Because Muse uses a flat manifest (path → object_id)
5 instead of nested tree objects, directory entries are synthesized on the fly
6 from shared path prefixes.
7
8 Modes
9 -----
10 ``muse ls-tree HEAD``
11 Non-recursive listing of the root. Blobs at the root level appear as
12 ``blob`` entries; any path that has children in a subdirectory is
13 collapsed into a synthetic ``tree`` entry (e.g. ``src/``).
14
15 ``muse ls-tree -r HEAD``
16 Recursive listing — all blobs, no synthetic tree entries.
17
18 ``muse ls-tree HEAD src/``
19 Scope the listing to the ``src/`` prefix.
20
21 ``muse ls-tree -d HEAD``
22 Show only synthetic directory (tree) entries, not blobs.
23
24 Output formats
25 --------------
26 Default text::
27
28 <mode> <type> <object_id>\\t<path>
29
30 ``--name-only`` text::
31
32 <path>
33
34 ``--long`` (``-l``) text adds the byte size between ``<object_id>`` and the
35 tab::
36
37 <mode> <type> <object_id> <size>\\t<path>
38
39 JSON (``--json``)::
40
41 {
42 "treeish": "HEAD",
43 "commit_id": "<sha256>",
44 "entries": [
45 {"mode": "100644", "type": "blob",
46 "object_id": "...", "size": 12, "path": "file.py"},
47 {"mode": "040000", "type": "tree",
48 "object_id": "...", "size": null, "path": "src/"}
49 ]
50 }
51
52 When ``--name-only`` is combined with ``--json`` the entries contain only
53 ``path`` (no ``object_id``, ``mode``, ``type``, or ``size``).
54
55 Exit codes::
56
57 0 — success
58 1 — user error: bad ref, path traversal, ANSI in ref
59 2 — not a Muse repository
60 3 — I/O error
61 """
62
63 from __future__ import annotations
64
65 import argparse
66 import hashlib
67 import json as _json
68 import logging
69 import pathlib
70 import sys
71
72 from muse.core.errors import ExitCode
73 from muse.core.object_store import read_object
74 from muse.core.repo import read_repo_id, require_repo
75 from muse.core.store import (
76 get_head_commit_id,
77 read_commit,
78 read_current_branch,
79 read_snapshot,
80 resolve_commit_ref,
81 )
82 from muse.core.validation import sanitize_display
83
84 logger = logging.getLogger(__name__)
85
86 _BLOB_MODE = "100644"
87 _TREE_MODE = "040000"
88
89
90 # ---------------------------------------------------------------------------
91 # Internal helpers
92 # ---------------------------------------------------------------------------
93
94
95 def _synthetic_tree_id(manifest: dict[str, str], prefix: str) -> str:
96 """Return a deterministic 64-hex-char ID for the synthetic tree at *prefix*.
97
98 The ID is the SHA-256 of the sorted ``(path, object_id)`` pairs for all
99 manifest entries that fall under *prefix* (direct and indirect children).
100
101 Args:
102 manifest: Full flat manifest (path → object_id).
103 prefix: Directory prefix ending with ``/`` (e.g. ``"src/"``).
104
105 Returns:
106 64-char lowercase hex string.
107 """
108 h = hashlib.sha256()
109 for path in sorted(manifest):
110 if path.startswith(prefix):
111 line = f"{path}\x00{manifest[path]}\n"
112 h.update(line.encode())
113 return h.hexdigest()
114
115
116 def _build_tree_entries(
117 manifest: dict[str, str],
118 path_prefix: str,
119 recursive: bool,
120 ) -> list[dict]:
121 """Build the list of tree entries for a given prefix and recursion mode.
122
123 Args:
124 manifest: Full flat manifest (path → object_id).
125 path_prefix: Repo-relative POSIX prefix to scope the listing, e.g.
126 ``""`` for root or ``"src/"`` for a subdirectory.
127 recursive: If True, return all blobs (no synthetic tree entries).
128 If False, return immediate children only — blobs for files
129 in this directory level, synthetic tree entries for
130 subdirectories.
131
132 Returns:
133 Sorted list of entry dicts with keys:
134 ``mode``, ``type``, ``object_id``, ``size`` (None for trees), ``path``.
135 """
136 if recursive:
137 # Return every blob whose path starts with the prefix.
138 entries = []
139 for path, oid in sorted(manifest.items()):
140 if path.startswith(path_prefix):
141 entries.append({
142 "mode": _BLOB_MODE,
143 "type": "blob",
144 "object_id": oid,
145 "size": None,
146 "path": path,
147 })
148 return entries
149
150 # Non-recursive: collect immediate children at this directory level.
151 seen_dirs: set[str] = set()
152 entries: list[dict] = []
153
154 for path, oid in sorted(manifest.items()):
155 if not path.startswith(path_prefix):
156 continue
157 rel = path[len(path_prefix):] # path relative to the prefix
158 slash = rel.find("/")
159 if slash == -1:
160 # Direct blob child.
161 entries.append({
162 "mode": _BLOB_MODE,
163 "type": "blob",
164 "object_id": oid,
165 "size": None,
166 "path": path,
167 })
168 else:
169 # The path passes through a subdirectory — emit a synthetic tree.
170 dir_name = rel[:slash + 1] # e.g. "src/"
171 dir_full = path_prefix + dir_name # e.g. "src/" or "pkg/sub/"
172 if dir_full not in seen_dirs:
173 seen_dirs.add(dir_full)
174 entries.append({
175 "mode": _TREE_MODE,
176 "type": "tree",
177 "object_id": _synthetic_tree_id(manifest, dir_full),
178 "size": None,
179 "path": dir_full,
180 })
181
182 return sorted(entries, key=lambda e: e["path"])
183
184
185 def _resolve_manifest(
186 root: pathlib.Path,
187 treeish: str,
188 ) -> tuple[str, dict[str, str]]:
189 """Resolve *treeish* to a ``(commit_id, manifest)`` pair.
190
191 Resolution order:
192 1. ``"HEAD"`` — current branch tip.
193 2. Branch name — ``.muse/refs/heads/<treeish>``.
194 3. Full or abbreviated commit ID — prefix scan of commits dir.
195
196 Args:
197 root: Absolute repo root.
198 treeish: Branch name, commit ID, or ``"HEAD"``.
199
200 Returns:
201 ``(commit_id, manifest)`` tuple.
202
203 Raises:
204 SystemExit(USER_ERROR): ref not found or repo is empty.
205 """
206 try:
207 branch = read_current_branch(root)
208 repo_id = read_repo_id(root)
209 commit = None
210
211 if treeish.upper() == "HEAD":
212 commit_id = get_head_commit_id(root, branch)
213 if not commit_id:
214 print(
215 "❌ Repository has no commits yet.",
216 file=sys.stderr,
217 )
218 raise SystemExit(ExitCode.USER_ERROR)
219 commit = read_commit(root, commit_id)
220 else:
221 # Try as a branch name first (direct ref file lookup).
222 branch_ref = root / ".muse" / "refs" / "heads" / treeish
223 if branch_ref.exists():
224 commit_id = branch_ref.read_text(encoding="utf-8").strip()
225 commit = read_commit(root, commit_id)
226 else:
227 # Fall back to commit-ID prefix scan.
228 commit = resolve_commit_ref(root, repo_id, branch, treeish)
229
230 if commit is None:
231 print(
232 f"❌ '{sanitize_display(treeish)}' is not a known branch or commit ID.",
233 file=sys.stderr,
234 )
235 raise SystemExit(ExitCode.USER_ERROR)
236
237 commit_id = commit.commit_id
238 snap = read_snapshot(root, commit.snapshot_id)
239 manifest = dict(snap.manifest) if snap else {}
240 return commit_id, manifest
241 except SystemExit:
242 raise
243 except Exception as exc:
244 print(f"❌ Failed to resolve '{sanitize_display(treeish)}': {exc}", file=sys.stderr)
245 raise SystemExit(ExitCode.USER_ERROR)
246
247
248 def _validate_path_prefix(root: pathlib.Path, raw: str) -> str:
249 """Validate and normalise a user-supplied path prefix.
250
251 Rejects path-traversal attempts (``..`` components, absolute paths that
252 escape the repo root).
253
254 Args:
255 root: Absolute repo root.
256 raw: Raw path string as given by the user.
257
258 Returns:
259 Normalised repo-relative POSIX path with trailing ``/`` if it looks
260 like a directory prefix, or as-is for explicit file paths.
261
262 Raises:
263 SystemExit(USER_ERROR): path traversal detected.
264 """
265 # Reject paths that try to escape the repo.
266 try:
267 candidate = (root / raw).resolve()
268 candidate.relative_to(root.resolve())
269 except ValueError:
270 print(
271 f"❌ Path '{sanitize_display(raw)}' is outside the repository root.",
272 file=sys.stderr,
273 )
274 raise SystemExit(ExitCode.USER_ERROR)
275
276 # Build the normalised relative POSIX path.
277 try:
278 rel = candidate.relative_to(root.resolve()).as_posix()
279 except ValueError:
280 print(
281 f"❌ Path '{sanitize_display(raw)}' is outside the repository root.",
282 file=sys.stderr,
283 )
284 raise SystemExit(ExitCode.USER_ERROR)
285
286 if rel == ".":
287 return ""
288
289 # Preserve trailing slash for directory-prefix semantics.
290 if raw.endswith("/"):
291 return rel + "/"
292 return rel
293
294
295 # ---------------------------------------------------------------------------
296 # Registration
297 # ---------------------------------------------------------------------------
298
299
300 def register(
301 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
302 ) -> None:
303 """Register the ``muse ls-tree`` subcommand."""
304 parser = subparsers.add_parser(
305 "ls-tree",
306 help="List the contents of a snapshot at a given ref.",
307 description=__doc__,
308 formatter_class=argparse.RawDescriptionHelpFormatter,
309 )
310 parser.add_argument(
311 "-r", "--recursive",
312 action="store_true",
313 dest="recursive",
314 help="Recurse into subtrees, listing all blobs.",
315 )
316 parser.add_argument(
317 "-d", "--dirs-only",
318 action="store_true",
319 dest="dirs_only",
320 help="Show only tree (directory) entries, not blobs.",
321 )
322 parser.add_argument(
323 "-l", "--long",
324 action="store_true",
325 dest="long",
326 help="Include object size in the listing.",
327 )
328 parser.add_argument(
329 "--name-only",
330 action="store_true",
331 dest="name_only",
332 help="Show only path names, omitting mode/type/object_id.",
333 )
334 parser.add_argument(
335 "--json",
336 action="store_true",
337 dest="output_json",
338 help="Emit machine-readable JSON on stdout.",
339 )
340 parser.add_argument(
341 "treeish",
342 metavar="TREEISH",
343 nargs="?",
344 default="HEAD",
345 help="Branch name or commit ID to inspect (default: HEAD).",
346 )
347 parser.add_argument(
348 "path",
349 metavar="PATH",
350 nargs="?",
351 default=None,
352 help="Optional path prefix to scope the listing.",
353 )
354 parser.set_defaults(func=run)
355
356
357 # ---------------------------------------------------------------------------
358 # Run
359 # ---------------------------------------------------------------------------
360
361
362 def run(args: argparse.Namespace) -> None:
363 """List snapshot contents for a given ref.
364
365 Exit codes::
366
367 0 — success
368 1 — bad ref, path traversal, ANSI in ref, empty repo
369 2 — not inside a Muse repo
370 3 — I/O error
371 """
372 treeish: str = args.treeish or "HEAD"
373 raw_path: str | None = args.path
374 recursive: bool = args.recursive
375 dirs_only: bool = args.dirs_only
376 long_fmt: bool = args.long
377 name_only: bool = args.name_only
378 output_json: bool = args.output_json
379
380 root = require_repo()
381
382 # ── Validate ref — reject ANSI and other control characters ──────────────
383 if any(ord(c) < 32 for c in treeish):
384 print(
385 f"❌ Invalid ref '{sanitize_display(treeish)}': control characters not allowed.",
386 file=sys.stderr,
387 )
388 raise SystemExit(ExitCode.USER_ERROR)
389
390 # ── Resolve the ref to a manifest ────────────────────────────────────────
391 commit_id, manifest = _resolve_manifest(root, treeish)
392
393 # ── Validate and normalise path prefix ───────────────────────────────────
394 path_prefix = ""
395 if raw_path is not None:
396 path_prefix = _validate_path_prefix(root, raw_path)
397 # Ensure directory prefixes end with /
398 if path_prefix and not path_prefix.endswith("/"):
399 path_prefix += "/"
400
401 # ── Build entries ─────────────────────────────────────────────────────────
402 entries = _build_tree_entries(manifest, path_prefix, recursive)
403
404 # Apply --dirs-only filter.
405 if dirs_only:
406 entries = [e for e in entries if e["type"] == "tree"]
407
408 # ── Populate sizes when --long is requested ───────────────────────────────
409 if long_fmt:
410 for entry in entries:
411 if entry["type"] == "blob":
412 data = read_object(root, entry["object_id"])
413 entry["size"] = len(data) if data is not None else None
414
415 # ── Output ───────────────────────────────────────────────────────────────
416 if output_json:
417 if name_only:
418 out_entries = [{"path": e["path"]} for e in entries]
419 else:
420 out_entries = []
421 for e in entries:
422 rec: dict = {
423 "mode": e["mode"],
424 "type": e["type"],
425 "object_id": e["object_id"],
426 "size": e["size"],
427 "path": e["path"],
428 }
429 out_entries.append(rec)
430 print(_json.dumps({
431 "treeish": treeish,
432 "commit_id": commit_id,
433 "entries": out_entries,
434 }))
435 else:
436 for e in entries:
437 if name_only:
438 print(e["path"])
439 elif long_fmt:
440 size_str = str(e["size"]) if e["size"] is not None else "-"
441 print(f"{e['mode']} {e['type']} {e['object_id']} {size_str}\t{e['path']}")
442 else:
443 print(f"{e['mode']} {e['type']} {e['object_id']}\t{e['path']}")
File History 1 commit
sha256:2fa778aba8ab0ec15295b8624c6480a573482ffc9c206a6d9546f1c41d2c2b7b feat: supercharge muse blame + remove --porcelain everywhere Human patch 164 days ago