"""muse plumbing show-ref — list all refs known to this repository. Enumerates every branch ref stored under ``.muse/refs/heads/`` and reports the commit ID each one points to. Optionally filters by a glob pattern, reports only the HEAD ref, verifies that a specific ref exists, or emits only a count. Output (JSON, default):: { "refs": [ {"ref": "refs/heads/dev", "commit_id": ""}, {"ref": "refs/heads/main", "commit_id": ""} ], "head": { "ref": "refs/heads/main", "branch": "main", "commit_id": "" }, "count": 2 } Text output (``--format text``):: refs/heads/dev * refs/heads/main (HEAD) Plumbing contract ----------------- - Exit 0: refs enumerated successfully (list may be empty). - Exit 0: verify mode and ref exists. - Exit 1: bad ``--format`` value; verify mode and ref does not exist. - Exit 3: I/O error reading refs directory. Agent use --------- Check whether a branch exists before starting work:: muse plumbing show-ref --verify refs/heads/feat/my-task --json # → {"ref": "refs/heads/feat/my-task", "exists": true} Count branches for capacity planning:: muse plumbing show-ref --count # → {"count": 42} Filter to a lane:: muse plumbing show-ref --pattern 'refs/heads/task/*' --json Read HEAD without a round-trip to status:: muse plumbing show-ref --head --json # → {"head": {"ref": "refs/heads/main", "branch": "main", "commit_id": ""}} """ from __future__ import annotations import argparse import fnmatch import json import logging import pathlib import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.store import ( get_head_commit_id, read_current_branch, ) from muse.core.validation import sanitize_display, validate_object_id logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("json", "text") _SHA256_HEX_LEN = 64 class _RefEntry(TypedDict): ref: str commit_id: str class _HeadInfo(TypedDict): ref: str branch: str commit_id: str class _ShowRefResult(TypedDict): refs: list[_RefEntry] head: _HeadInfo | None count: int def _list_branch_refs(root: pathlib.Path) -> list[_RefEntry]: """Return every branch ref under ``.muse/refs/heads/``, sorted by name. Ref files that are symlinks are skipped — they could point outside the repository and are not a valid Muse ref format. Ref files whose content is not a valid 64-char hex SHA256 are also skipped with a debug log — this guards against corrupt or adversarially crafted ref files injecting invalid commit IDs into consumer pipelines. """ heads_dir = root / ".muse" / "refs" / "heads" if not heads_dir.exists(): return [] refs: list[_RefEntry] = [] for child in sorted(heads_dir.iterdir()): # Reject symlinks — they can point outside the repository. if child.is_symlink(): logger.debug("show-ref: skipping symlink ref %s", child.name) continue if not child.is_file(): continue branch = child.name commit_id = child.read_text(encoding="utf-8").strip() if not commit_id: continue # Validate commit ID format before including in output. try: validate_object_id(commit_id) except ValueError: logger.debug( "show-ref: skipping ref %r — invalid commit ID %r", branch, commit_id, ) continue refs.append({"ref": f"refs/heads/{branch}", "commit_id": commit_id}) return refs def _head_info(root: pathlib.Path) -> _HeadInfo | None: """Return metadata about what HEAD currently points to, or ``None``.""" try: muse_dir = root / ".muse" branch = read_current_branch(root) except Exception: return None commit_id = get_head_commit_id(root, branch) if commit_id is None: return None return { "ref": f"refs/heads/{branch}", "branch": branch, "commit_id": commit_id, } def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the show-ref subcommand.""" parser = subparsers.add_parser( "show-ref", help="List all branch refs and their commit IDs.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--pattern", "-p", default=None, dest="pattern", metavar="GLOB", help="fnmatch glob filter applied to the full ref name (e.g. 'refs/heads/feat/*').", ) parser.add_argument( "--head", "-H", action="store_true", dest="head_only", help="Print only the HEAD ref and its commit ID.", ) parser.add_argument( "--verify", "-v", default=None, dest="verify_ref", metavar="REF", help=( "Check whether the given ref exists. " "Exits 0 if found, 1 if not. " "With --json, emits {\"ref\": ..., \"exists\": bool} to stdout." ), ) parser.add_argument( "--count", action="store_true", dest="count_only", help=( "Emit only the branch count. " "With --json, emits {\"count\": N}. " "With --pattern, counts matching refs. " "Useful for capacity planning in agent pipelines." ), ) parser.add_argument( "--format", "-f", dest="fmt", default="json", metavar="FORMAT", help="Output format: json or text. (default: json)", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """List all refs known to this repository. Reads every branch pointer from ``.muse/refs/heads/`` and reports their commit IDs. The output is sorted lexicographically by ref name. Ref files that are symlinks or contain non-hex content are silently skipped to guard against corrupt or adversarially crafted ref stores. """ fmt: str = args.fmt pattern: str | None = args.pattern head_only: bool = args.head_only verify_ref: str | None = args.verify_ref count_only: bool = args.count_only if fmt not in _FORMAT_CHOICES: print( json.dumps( {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"} ), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() muse_dir = root / ".muse" # --verify mode: existence check. if verify_ref is not None: try: all_refs = _list_branch_refs(root) except Exception as exc: logger.debug("show-ref I/O error: %s", exc) raise SystemExit(ExitCode.INTERNAL_ERROR) exists = any(r["ref"] == verify_ref for r in all_refs) if fmt == "json": print(json.dumps({"ref": verify_ref, "exists": exists})) raise SystemExit(0 if exists else ExitCode.USER_ERROR) # --head mode: only HEAD. if head_only: info = _head_info(root) if info is None: if fmt == "text": print("(no HEAD commit)") else: print(json.dumps({"head": None})) return if fmt == "text": print( f"{sanitize_display(info['commit_id'])} " f"{sanitize_display(info['ref'])} (HEAD)" ) else: print(json.dumps({"head": dict(info)})) return # Collect refs, applying optional glob filter. try: refs = _list_branch_refs(root) except Exception as exc: logger.debug("show-ref I/O error: %s", exc) print(json.dumps({"error": str(exc)}), file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) if pattern is not None: refs = [r for r in refs if fnmatch.fnmatch(r["ref"], pattern)] # --count mode: emit count only. if count_only: if fmt == "text": print(str(len(refs))) else: print(json.dumps({"count": len(refs)})) return head = _head_info(root) if fmt == "text": head_ref = head["ref"] if head else None for r in refs: is_head = r["ref"] == head_ref marker = "* " if is_head else " " suffix = " (HEAD)" if is_head else "" print( f"{sanitize_display(r['commit_id'])} " f"{marker}{sanitize_display(r['ref'])}{suffix}" ) return result: _ShowRefResult = { "refs": refs, "head": head, "count": len(refs), } print(json.dumps(result))