gabriel / muse public
rev_parse.py python
116 lines 3.3 KB
1c3feb01 audit and harden all 13 plumbing commands for production Gabriel Cardona <gabriel@tellurstori.com> 2d 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 Output (--format text)::
11
12 <sha256>
13
14 Plumbing contract
15 -----------------
16
17 - Exit 0: ref resolved successfully.
18 - Exit 1: ref not found, ambiguous, or unknown --format value.
19 """
20
21 from __future__ import annotations
22
23 import json
24 import logging
25
26 import typer
27
28 from muse.core.errors import ExitCode
29 from muse.core.repo import require_repo
30 from muse.core.store import (
31 find_commits_by_prefix,
32 get_head_commit_id,
33 read_commit,
34 read_current_branch,
35 )
36
37 logger = logging.getLogger(__name__)
38
39 app = typer.Typer()
40
41 _FORMAT_CHOICES = ("json", "text")
42
43
44 @app.callback(invoke_without_command=True)
45 def rev_parse(
46 ctx: typer.Context,
47 ref: str = typer.Argument(
48 ...,
49 help="Ref to resolve: branch name, 'HEAD', or commit ID prefix.",
50 ),
51 fmt: str = typer.Option(
52 "json", "--format", "-f", help="Output format: json or text."
53 ),
54 ) -> None:
55 """Resolve a branch name, HEAD, or SHA prefix to a full commit ID.
56
57 Analogous to ``git rev-parse``. Useful for canonicalising refs in
58 scripts and agent pipelines before passing them to other plumbing
59 commands.
60 """
61 if fmt not in _FORMAT_CHOICES:
62 typer.echo(
63 f"❌ Unknown format {fmt!r}. Valid choices: {', '.join(_FORMAT_CHOICES)}",
64 err=True,
65 )
66 raise typer.Exit(code=ExitCode.USER_ERROR)
67
68 root = require_repo()
69
70 commit_id: str | None = None
71
72 if ref.upper() == "HEAD":
73 branch = read_current_branch(root)
74 commit_id = get_head_commit_id(root, branch)
75 if commit_id is None:
76 typer.echo(
77 json.dumps({"ref": ref, "commit_id": None, "error": "HEAD has no commits"})
78 )
79 raise typer.Exit(code=ExitCode.USER_ERROR)
80 else:
81 # Try as branch name first.
82 candidate = get_head_commit_id(root, ref)
83 if candidate is not None:
84 commit_id = candidate
85 else:
86 # Try as full or abbreviated commit ID.
87 if len(ref) == 64:
88 record = read_commit(root, ref)
89 if record is not None:
90 commit_id = record.commit_id
91 else:
92 matches = find_commits_by_prefix(root, ref)
93 if len(matches) == 1:
94 commit_id = matches[0].commit_id
95 elif len(matches) > 1:
96 typer.echo(
97 json.dumps(
98 {
99 "ref": ref,
100 "commit_id": None,
101 "error": "ambiguous",
102 "candidates": [m.commit_id for m in matches],
103 }
104 )
105 )
106 raise typer.Exit(code=ExitCode.USER_ERROR)
107
108 if commit_id is None:
109 typer.echo(json.dumps({"ref": ref, "commit_id": None, "error": "not found"}))
110 raise typer.Exit(code=ExitCode.USER_ERROR)
111
112 if fmt == "text":
113 typer.echo(commit_id)
114 return
115
116 typer.echo(json.dumps({"ref": ref, "commit_id": commit_id}))