ls_files.py python
173 lines 5.1 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """muse plumbing ls-files — list tracked files in a snapshot.
2
3 Lists every file tracked in a commit's snapshot, along with the SHA-256
4 object ID of its content. Defaults to the HEAD commit of the current branch.
5
6 Output (JSON, default)::
7
8 {
9 "commit_id": "<sha256>",
10 "snapshot_id": "<sha256>",
11 "file_count": 3,
12 "files": [
13 {"path": "tracks/drums.mid", "object_id": "<sha256>"},
14 ...
15 ]
16 }
17
18 Output (--format text)::
19
20 <object_id>\t<path>
21 ...
22
23 Plumbing contract
24 -----------------
25
26 - Exit 0: manifest listed successfully.
27 - Exit 1: commit or snapshot not found, or unknown --format value.
28
29 Agent use
30 ---------
31
32 Filter to a subdirectory to avoid pulling the full manifest::
33
34 muse plumbing ls-files --path-prefix src/
35 muse plumbing ls-files --path-prefix src/ --commit <sha>
36 """
37
38 from __future__ import annotations
39
40 import argparse
41 import json
42 import logging
43 import sys
44
45 from muse.core.errors import ExitCode
46 from muse.core.repo import require_repo
47 from muse.core.store import (
48 get_commit_snapshot_manifest,
49 get_head_commit_id,
50 read_commit,
51 read_current_branch,
52 )
53 from muse.core.validation import sanitize_display, validate_object_id, validate_path_prefix
54
55 logger = logging.getLogger(__name__)
56
57 _FORMAT_CHOICES = ("json", "text")
58
59
60 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
61 """Register the ls-files subcommand."""
62 parser = subparsers.add_parser(
63 "ls-files",
64 help="List all tracked files and their object IDs in a snapshot.",
65 description=__doc__,
66 formatter_class=argparse.RawDescriptionHelpFormatter,
67 )
68 parser.add_argument(
69 "--commit", "-c",
70 default=None,
71 dest="commit",
72 metavar="COMMIT_ID",
73 help="Commit ID to read (default: HEAD).",
74 )
75 parser.add_argument(
76 "--path-prefix", "-p",
77 default=None,
78 dest="path_prefix",
79 metavar="PREFIX",
80 help="Only list files whose path starts with PREFIX.",
81 )
82 parser.add_argument(
83 "--format", "-f",
84 dest="fmt",
85 default="json",
86 metavar="FORMAT",
87 help="Output format: json or text. (default: json)",
88 )
89 parser.add_argument(
90 "--json", action="store_const", const="json", dest="fmt",
91 help="Shorthand for --format json.",
92 )
93 parser.set_defaults(func=run)
94
95
96 def run(args: argparse.Namespace) -> None:
97 """List all tracked files and their object IDs in a snapshot.
98
99 Analogous to ``git ls-files --stage``. Reads the snapshot manifest of
100 the given commit (or HEAD) and prints each tracked file path together
101 with its content-addressed object ID.
102
103 Use ``--path-prefix`` to scope the listing to a subdirectory — agents
104 should prefer this over listing the full manifest and filtering client-side.
105 """
106 fmt: str = args.fmt
107 commit: str | None = args.commit
108 path_prefix: str | None = args.path_prefix
109
110 if fmt not in _FORMAT_CHOICES:
111 print(
112 f"❌ Unknown format {fmt!r}. Valid choices: {', '.join(_FORMAT_CHOICES)}",
113 file=sys.stderr,
114 )
115 raise SystemExit(ExitCode.USER_ERROR)
116
117 root = require_repo()
118
119 if commit is None:
120 branch = read_current_branch(root)
121 commit_id = get_head_commit_id(root, branch)
122 if commit_id is None:
123 print(json.dumps({"error": "No commits on current branch."}))
124 raise SystemExit(ExitCode.USER_ERROR)
125 else:
126 try:
127 validate_object_id(commit)
128 except ValueError as exc:
129 print(json.dumps({"error": f"Invalid commit ID: {exc}"}))
130 raise SystemExit(ExitCode.USER_ERROR)
131 commit_id = commit
132
133 commit_record = read_commit(root, commit_id)
134 if commit_record is None:
135 print(json.dumps({"error": f"Commit not found: {commit_id}"}))
136 raise SystemExit(ExitCode.USER_ERROR)
137
138 manifest = get_commit_snapshot_manifest(root, commit_id)
139 if manifest is None:
140 print(json.dumps({"error": f"Snapshot not found for commit: {commit_id}"}))
141 raise SystemExit(ExitCode.USER_ERROR)
142
143 if path_prefix is not None:
144 try:
145 validate_path_prefix(path_prefix)
146 except ValueError as exc:
147 print(json.dumps({"error": f"Invalid --path-prefix: {exc}"}))
148 raise SystemExit(ExitCode.USER_ERROR)
149
150 items = sorted(manifest.items())
151 if path_prefix is not None:
152 items = [(p, oid) for p, oid in items if p.startswith(path_prefix)]
153
154 files = [{"path": p, "object_id": oid} for p, oid in items]
155
156 if fmt == "text":
157 for entry in files:
158 # sanitize_display guards against ANSI sequences in file paths —
159 # valid on most filesystems but dangerous when echoed to a terminal.
160 print(f"{entry['object_id']}\t{sanitize_display(entry['path'])}")
161 return
162
163 print(
164 json.dumps(
165 {
166 "commit_id": commit_id,
167 "snapshot_id": commit_record.snapshot_id,
168 "file_count": len(files),
169 "files": files,
170 },
171 indent=2,
172 )
173 )
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago