gabriel / muse public
show_ref.py python
308 lines 8.9 KB
Raw
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
1 """muse plumbing show-ref — list all refs known to this repository.
2
3 Enumerates every branch ref stored under ``.muse/refs/heads/`` and reports
4 the commit ID each one points to. Optionally filters by a glob pattern,
5 reports only the HEAD ref, verifies that a specific ref exists, or emits
6 only a count.
7
8 Output (JSON, default)::
9
10 {
11 "refs": [
12 {"ref": "refs/heads/dev", "commit_id": "<sha256>"},
13 {"ref": "refs/heads/main", "commit_id": "<sha256>"}
14 ],
15 "head": {
16 "ref": "refs/heads/main",
17 "branch": "main",
18 "commit_id": "<sha256>"
19 },
20 "count": 2
21 }
22
23 Text output (``--format text``)::
24
25 <sha256> refs/heads/dev
26 * <sha256> refs/heads/main (HEAD)
27
28 Plumbing contract
29 -----------------
30
31 - Exit 0: refs enumerated successfully (list may be empty).
32 - Exit 0: verify mode and ref exists.
33 - Exit 1: bad ``--format`` value; verify mode and ref does not exist.
34 - Exit 3: I/O error reading refs directory.
35
36 Agent use
37 ---------
38
39 Check whether a branch exists before starting work::
40
41 muse plumbing show-ref --verify refs/heads/feat/my-task --json
42 # → {"ref": "refs/heads/feat/my-task", "exists": true}
43
44 Count branches for capacity planning::
45
46 muse plumbing show-ref --count
47 # → {"count": 42}
48
49 Filter to a lane::
50
51 muse plumbing show-ref --pattern 'refs/heads/task/*' --json
52
53 Read HEAD without a round-trip to status::
54
55 muse plumbing show-ref --head --json
56 # → {"head": {"ref": "refs/heads/main", "branch": "main", "commit_id": "<sha256>"}}
57 """
58
59 from __future__ import annotations
60
61 import argparse
62 import fnmatch
63 import json
64 import logging
65 import pathlib
66 import sys
67 from typing import TypedDict
68
69 from muse.core.errors import ExitCode
70 from muse.core.repo import require_repo
71 from muse.core.store import (
72 get_head_commit_id,
73 read_current_branch,
74 )
75 from muse.core.validation import sanitize_display, validate_object_id
76
77 logger = logging.getLogger(__name__)
78
79 _FORMAT_CHOICES = ("json", "text")
80 _SHA256_HEX_LEN = 64
81
82
83 class _RefEntry(TypedDict):
84 ref: str
85 commit_id: str
86
87
88 class _HeadInfo(TypedDict):
89 ref: str
90 branch: str
91 commit_id: str
92
93
94 class _ShowRefResult(TypedDict):
95 refs: list[_RefEntry]
96 head: _HeadInfo | None
97 count: int
98
99
100 def _list_branch_refs(root: pathlib.Path) -> list[_RefEntry]:
101 """Return every branch ref under ``.muse/refs/heads/``, sorted by name.
102
103 Ref files that are symlinks are skipped — they could point outside the
104 repository and are not a valid Muse ref format. Ref files whose content
105 is not a valid 64-char hex SHA256 are also skipped with a debug log —
106 this guards against corrupt or adversarially crafted ref files injecting
107 invalid commit IDs into consumer pipelines.
108 """
109 heads_dir = root / ".muse" / "refs" / "heads"
110 if not heads_dir.exists():
111 return []
112
113 refs: list[_RefEntry] = []
114 for child in sorted(heads_dir.iterdir()):
115 # Reject symlinks — they can point outside the repository.
116 if child.is_symlink():
117 logger.debug("show-ref: skipping symlink ref %s", child.name)
118 continue
119 if not child.is_file():
120 continue
121 branch = child.name
122 commit_id = child.read_text(encoding="utf-8").strip()
123 if not commit_id:
124 continue
125 # Validate commit ID format before including in output.
126 try:
127 validate_object_id(commit_id)
128 except ValueError:
129 logger.debug(
130 "show-ref: skipping ref %r — invalid commit ID %r",
131 branch, commit_id,
132 )
133 continue
134 refs.append({"ref": f"refs/heads/{branch}", "commit_id": commit_id})
135 return refs
136
137
138 def _head_info(root: pathlib.Path) -> _HeadInfo | None:
139 """Return metadata about what HEAD currently points to, or ``None``."""
140 try:
141 muse_dir = root / ".muse"
142 branch = read_current_branch(root)
143 except Exception:
144 return None
145 commit_id = get_head_commit_id(root, branch)
146 if commit_id is None:
147 return None
148 return {
149 "ref": f"refs/heads/{branch}",
150 "branch": branch,
151 "commit_id": commit_id,
152 }
153
154
155 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
156 """Register the show-ref subcommand."""
157 parser = subparsers.add_parser(
158 "show-ref",
159 help="List all branch refs and their commit IDs.",
160 description=__doc__,
161 formatter_class=argparse.RawDescriptionHelpFormatter,
162 )
163 parser.add_argument(
164 "--pattern", "-p",
165 default=None,
166 dest="pattern",
167 metavar="GLOB",
168 help="fnmatch glob filter applied to the full ref name (e.g. 'refs/heads/feat/*').",
169 )
170 parser.add_argument(
171 "--head", "-H",
172 action="store_true",
173 dest="head_only",
174 help="Print only the HEAD ref and its commit ID.",
175 )
176 parser.add_argument(
177 "--verify", "-v",
178 default=None,
179 dest="verify_ref",
180 metavar="REF",
181 help=(
182 "Check whether the given ref exists. "
183 "Exits 0 if found, 1 if not. "
184 "With --json, emits {\"ref\": ..., \"exists\": bool} to stdout."
185 ),
186 )
187 parser.add_argument(
188 "--count",
189 action="store_true",
190 dest="count_only",
191 help=(
192 "Emit only the branch count. "
193 "With --json, emits {\"count\": N}. "
194 "With --pattern, counts matching refs. "
195 "Useful for capacity planning in agent pipelines."
196 ),
197 )
198 parser.add_argument(
199 "--format", "-f",
200 dest="fmt",
201 default="json",
202 metavar="FORMAT",
203 help="Output format: json or text. (default: json)",
204 )
205 parser.add_argument(
206 "--json", action="store_const", const="json", dest="fmt",
207 help="Shorthand for --format json.",
208 )
209 parser.set_defaults(func=run)
210
211
212 def run(args: argparse.Namespace) -> None:
213 """List all refs known to this repository.
214
215 Reads every branch pointer from ``.muse/refs/heads/`` and reports their
216 commit IDs. The output is sorted lexicographically by ref name.
217
218 Ref files that are symlinks or contain non-hex content are silently
219 skipped to guard against corrupt or adversarially crafted ref stores.
220 """
221 fmt: str = args.fmt
222 pattern: str | None = args.pattern
223 head_only: bool = args.head_only
224 verify_ref: str | None = args.verify_ref
225 count_only: bool = args.count_only
226
227 if fmt not in _FORMAT_CHOICES:
228 print(
229 json.dumps(
230 {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"}
231 ),
232 file=sys.stderr,
233 )
234 raise SystemExit(ExitCode.USER_ERROR)
235
236 root = require_repo()
237 muse_dir = root / ".muse"
238
239 # --verify mode: existence check.
240 if verify_ref is not None:
241 try:
242 all_refs = _list_branch_refs(root)
243 except Exception as exc:
244 logger.debug("show-ref I/O error: %s", exc)
245 raise SystemExit(ExitCode.INTERNAL_ERROR)
246 exists = any(r["ref"] == verify_ref for r in all_refs)
247 if fmt == "json":
248 print(json.dumps({"ref": verify_ref, "exists": exists}))
249 raise SystemExit(0 if exists else ExitCode.USER_ERROR)
250
251 # --head mode: only HEAD.
252 if head_only:
253 info = _head_info(root)
254 if info is None:
255 if fmt == "text":
256 print("(no HEAD commit)")
257 else:
258 print(json.dumps({"head": None}))
259 return
260
261 if fmt == "text":
262 print(
263 f"{sanitize_display(info['commit_id'])} "
264 f"{sanitize_display(info['ref'])} (HEAD)"
265 )
266 else:
267 print(json.dumps({"head": dict(info)}))
268 return
269
270 # Collect refs, applying optional glob filter.
271 try:
272 refs = _list_branch_refs(root)
273 except Exception as exc:
274 logger.debug("show-ref I/O error: %s", exc)
275 print(json.dumps({"error": str(exc)}), file=sys.stderr)
276 raise SystemExit(ExitCode.INTERNAL_ERROR)
277
278 if pattern is not None:
279 refs = [r for r in refs if fnmatch.fnmatch(r["ref"], pattern)]
280
281 # --count mode: emit count only.
282 if count_only:
283 if fmt == "text":
284 print(str(len(refs)))
285 else:
286 print(json.dumps({"count": len(refs)}))
287 return
288
289 head = _head_info(root)
290
291 if fmt == "text":
292 head_ref = head["ref"] if head else None
293 for r in refs:
294 is_head = r["ref"] == head_ref
295 marker = "* " if is_head else " "
296 suffix = " (HEAD)" if is_head else ""
297 print(
298 f"{sanitize_display(r['commit_id'])} "
299 f"{marker}{sanitize_display(r['ref'])}{suffix}"
300 )
301 return
302
303 result: _ShowRefResult = {
304 "refs": refs,
305 "head": head,
306 "count": len(refs),
307 }
308 print(json.dumps(result))
File History 2 commits
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 101 days ago