rev_parse.py
python
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
13 days ago
| 1 | """muse plumbing rev-parse — resolve a ref to a full commit ID. |
| 2 | |
| 3 | Resolves a branch name, ``HEAD``, or an abbreviated SHA prefix to the full |
| 4 | 64-character SHA-256 commit ID. |
| 5 | |
| 6 | Output (JSON, default):: |
| 7 | |
| 8 | {"ref": "main", "commit_id": "<sha256>"} |
| 9 | |
| 10 | With ``--abbrev-ref``:: |
| 11 | |
| 12 | {"ref": "HEAD", "branch": "main"} # JSON |
| 13 | main # text |
| 14 | |
| 15 | Output (--format text):: |
| 16 | |
| 17 | <sha256> |
| 18 | |
| 19 | Plumbing contract |
| 20 | ----------------- |
| 21 | |
| 22 | - Exit 0: ref resolved successfully. |
| 23 | - Exit 1: ref not found, ambiguous, empty, or unknown --format value. |
| 24 | |
| 25 | Agent use |
| 26 | --------- |
| 27 | |
| 28 | Canonical "what branch am I on?" query:: |
| 29 | |
| 30 | muse plumbing rev-parse --abbrev-ref HEAD --json |
| 31 | # → {"ref": "HEAD", "branch": "main"} |
| 32 | |
| 33 | Canonical "what is HEAD?" query:: |
| 34 | |
| 35 | muse plumbing rev-parse HEAD --format text |
| 36 | # → <64 hex chars> |
| 37 | """ |
| 38 | |
| 39 | from __future__ import annotations |
| 40 | |
| 41 | import argparse |
| 42 | import json |
| 43 | import logging |
| 44 | import sys |
| 45 | |
| 46 | from muse.core.errors import ExitCode |
| 47 | from muse.core.repo import require_repo |
| 48 | from muse.core.store import ( |
| 49 | find_commits_by_prefix, |
| 50 | get_head_commit_id, |
| 51 | read_commit, |
| 52 | read_current_branch, |
| 53 | ) |
| 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 rev-parse subcommand.""" |
| 62 | parser = subparsers.add_parser( |
| 63 | "rev-parse", |
| 64 | help="Resolve branch/HEAD/SHA prefix → full commit_id.", |
| 65 | description=__doc__, |
| 66 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 67 | ) |
| 68 | parser.add_argument( |
| 69 | "ref", |
| 70 | help="Ref to resolve: branch name, 'HEAD', or commit ID prefix.", |
| 71 | ) |
| 72 | parser.add_argument( |
| 73 | "--abbrev-ref", |
| 74 | action="store_true", |
| 75 | dest="abbrev_ref", |
| 76 | help=( |
| 77 | "Resolve HEAD (or a commit_id) to the branch name rather than " |
| 78 | "the commit ID. Canonical agent query: " |
| 79 | "`muse plumbing rev-parse --abbrev-ref HEAD`." |
| 80 | ), |
| 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 | """Resolve a branch name, HEAD, or SHA prefix to a full commit ID. |
| 98 | |
| 99 | Analogous to ``git rev-parse``. Useful for canonicalising refs in |
| 100 | scripts and agent pipelines before passing them to other plumbing |
| 101 | commands. |
| 102 | |
| 103 | With ``--abbrev-ref``: return the symbolic branch name rather than the |
| 104 | commit hash — the idiomatic way to answer "what branch am I on?". |
| 105 | """ |
| 106 | fmt: str = args.fmt |
| 107 | ref: str = args.ref |
| 108 | abbrev_ref: bool = args.abbrev_ref |
| 109 | |
| 110 | if not ref: |
| 111 | print( |
| 112 | json.dumps({"ref": ref, "commit_id": None, "error": "ref must not be empty"}), |
| 113 | file=sys.stderr, |
| 114 | ) |
| 115 | raise SystemExit(ExitCode.USER_ERROR) |
| 116 | |
| 117 | if fmt not in _FORMAT_CHOICES: |
| 118 | print( |
| 119 | f"❌ Unknown format {fmt!r}. Valid choices: {', '.join(_FORMAT_CHOICES)}", |
| 120 | file=sys.stderr, |
| 121 | ) |
| 122 | raise SystemExit(ExitCode.USER_ERROR) |
| 123 | |
| 124 | root = require_repo() |
| 125 | |
| 126 | # ── --abbrev-ref: resolve HEAD → branch name ───────────────────────────── |
| 127 | if abbrev_ref: |
| 128 | branch = read_current_branch(root) |
| 129 | if fmt == "text": |
| 130 | print(branch) |
| 131 | else: |
| 132 | print(json.dumps({"ref": ref, "branch": branch})) |
| 133 | return |
| 134 | |
| 135 | # ── normal ref resolution ───────────────────────────────────────────────── |
| 136 | commit_id: str | None = None |
| 137 | |
| 138 | if ref.upper() == "HEAD": |
| 139 | branch = read_current_branch(root) |
| 140 | commit_id = get_head_commit_id(root, branch) |
| 141 | if commit_id is None: |
| 142 | print(json.dumps({"ref": ref, "commit_id": None, "error": "HEAD has no commits"})) |
| 143 | raise SystemExit(ExitCode.USER_ERROR) |
| 144 | else: |
| 145 | # Try as branch name first. validate_branch_name (called inside |
| 146 | # get_head_commit_id) raises ValueError for refs containing control |
| 147 | # characters, ANSI escapes, or other forbidden sequences — treat these |
| 148 | # as "not found" rather than letting the exception escape. |
| 149 | try: |
| 150 | candidate = get_head_commit_id(root, ref) |
| 151 | except ValueError: |
| 152 | candidate = None |
| 153 | if candidate is not None: |
| 154 | commit_id = candidate |
| 155 | else: |
| 156 | # Try as full or abbreviated commit ID. |
| 157 | if len(ref) == 64: |
| 158 | record = read_commit(root, ref) |
| 159 | if record is not None: |
| 160 | commit_id = record.commit_id |
| 161 | else: |
| 162 | matches = find_commits_by_prefix(root, ref) |
| 163 | if len(matches) == 1: |
| 164 | commit_id = matches[0].commit_id |
| 165 | elif len(matches) > 1: |
| 166 | print( |
| 167 | json.dumps( |
| 168 | { |
| 169 | "ref": ref, |
| 170 | "commit_id": None, |
| 171 | "error": "ambiguous", |
| 172 | "candidates": [m.commit_id for m in matches], |
| 173 | } |
| 174 | ) |
| 175 | ) |
| 176 | raise SystemExit(ExitCode.USER_ERROR) |
| 177 | |
| 178 | if commit_id is None: |
| 179 | print(json.dumps({"ref": ref, "commit_id": None, "error": "not found"})) |
| 180 | raise SystemExit(ExitCode.USER_ERROR) |
| 181 | |
| 182 | if fmt == "text": |
| 183 | print(commit_id) |
| 184 | return |
| 185 | |
| 186 | print(json.dumps({"ref": ref, "commit_id": commit_id})) |
File History
1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
13 days ago