gabriel / muse public
merge_base.py python
253 lines 7.6 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 142 days ago
1 """muse merge-base — find the lowest common ancestor of two commits.
2
3 Walks the commit DAG from two starting points and returns the nearest shared
4 ancestor (the Lowest Common Ancestor, or LCA). Used by merge engines, CI
5 systems, and agent pipelines to compute the divergence point between branches.
6
7 Output (JSON, default)::
8
9 {
10 "commit_a": "<sha256>",
11 "commit_b": "<sha256>",
12 "merge_base": "<sha256>",
13 "duration_ms": 1.2,
14 "exit_code": 0
15 }
16
17 When no common ancestor exists::
18
19 {
20 "commit_a": "<sha256>",
21 "commit_b": "<sha256>",
22 "merge_base": null,
23 "error": "no common ancestor",
24 "duration_ms": 0.8,
25 "exit_code": 0
26 }
27
28 JSON error schema (usage / internal errors)::
29
30 {
31 "status": "error",
32 "error": "<human-readable message>",
33 "exit_code": <int>
34 }
35
36 When ``--json`` is active all errors go to stdout as JSON — no prose on
37 stderr. Agents should parse stdout and check ``exit_code``.
38
39 Output contract
40 ---------------
41
42 - Exit 0: operation completed (check ``merge_base`` field for null vs. found).
43 - Exit 1: a commit ID or ref cannot be resolved; bad ``--format`` value.
44 - Exit 3: DAG walk failed (I/O error or malformed graph).
45 """
46
47 from __future__ import annotations
48
49 import argparse
50 import json
51 import logging
52 import pathlib
53 import sys
54 from typing import TypedDict
55
56 from muse.core.errors import ExitCode
57 from muse.core.merge_engine import find_merge_base
58 from muse.core.repo import require_repo
59 from muse.core.store import get_head_commit_id, read_commit, read_current_branch
60 from muse.core.validation import validate_object_id
61 from muse.core.timing import start_timer
62
63 logger = logging.getLogger(__name__)
64
65 _FORMAT_CHOICES = ("json", "text")
66
67
68 # ---------------------------------------------------------------------------
69 # Wire-format TypedDicts
70 # ---------------------------------------------------------------------------
71
72
73 class _MergeBaseFoundJson(TypedDict):
74 """Stable JSON envelope when a merge base is found."""
75 commit_a: str
76 commit_b: str
77 merge_base: str # sha256:… commit ID
78 duration_ms: float
79 exit_code: int # always 0
80
81
82 class _MergeBaseNotFoundJson(TypedDict):
83 """Stable JSON envelope when no common ancestor exists."""
84 commit_a: str
85 commit_b: str
86 merge_base: None
87 error: str # "no common ancestor"
88 duration_ms: float
89 exit_code: int # always 0
90
91
92 class _MergeBaseErrorJson(TypedDict):
93 """Error payload for usage/internal errors in --json mode."""
94 status: str # "error"
95 error: str
96 exit_code: int
97
98
99 # ---------------------------------------------------------------------------
100 # Helpers
101 # ---------------------------------------------------------------------------
102
103
104 def _emit_error(fmt: str, msg: str, code: ExitCode) -> None:
105 """Print an error and raise SystemExit. Never returns.
106
107 In ``--json`` mode the error goes to stdout as a JSON payload so agents
108 always get parseable output. In text mode it goes to stderr.
109 """
110 if fmt == "json":
111 print(json.dumps(_MergeBaseErrorJson(
112 status="error",
113 error=msg,
114 exit_code=int(code),
115 )))
116 else:
117 print(f"❌ {msg}", file=sys.stderr)
118 raise SystemExit(code)
119
120
121 def _resolve_ref(root: pathlib.Path, ref: str) -> str | None:
122 """Resolve a branch name, HEAD, or sha256-prefixed commit ID.
123
124 Returns ``None`` when the ref cannot be resolved to a known commit.
125 """
126 if ref.upper() == "HEAD":
127 branch = read_current_branch(root)
128 return get_head_commit_id(root, branch)
129
130 # Try as branch name first. Guard against refs that are not valid branch
131 # names (e.g. sha256:-prefixed commit IDs) — get_head_commit_id calls
132 # validate_branch_name which raises ValueError for colons and slashes.
133 try:
134 cid = get_head_commit_id(root, ref)
135 if cid is not None:
136 return cid
137 except (ValueError, OSError):
138 pass # not a valid branch name; fall through to commit-ID lookup
139
140 # Try as full sha256-prefixed commit ID.
141 try:
142 validate_object_id(ref)
143 record = read_commit(root, ref)
144 return record.commit_id if record else None
145 except ValueError:
146 return None
147
148
149 # ---------------------------------------------------------------------------
150 # Registration
151 # ---------------------------------------------------------------------------
152
153
154 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
155 """Register the merge-base subcommand."""
156 parser = subparsers.add_parser(
157 "merge-base",
158 help="Find the lowest common ancestor of two commits.",
159 description=__doc__,
160 )
161 parser.add_argument(
162 "commit_a",
163 help="First commit ID, branch name, or HEAD.",
164 )
165 parser.add_argument(
166 "commit_b",
167 help="Second commit ID, branch name, or HEAD.",
168 )
169 parser.add_argument(
170 "--format", "-f",
171 dest="fmt",
172 default="json",
173 metavar="FORMAT",
174 help="Output format: json or text. (default: json)",
175 )
176 parser.add_argument(
177 "--json", action="store_const", const="json", dest="fmt",
178 help="Shorthand for --format json.",
179 )
180 parser.set_defaults(func=run)
181
182
183 # ---------------------------------------------------------------------------
184 # Entry point
185 # ---------------------------------------------------------------------------
186
187
188 def run(args: argparse.Namespace) -> None:
189 """Find the lowest common ancestor of two commits.
190
191 Accepts sha256-prefixed commit IDs, branch names, or ``HEAD``. The result
192 is the commit reachable from both inputs that is closest to both tips —
193 the divergence point between two histories.
194
195 In ``--json`` mode all errors go to stdout as a JSON payload so agents can
196 always parse stdout and check ``exit_code``. Exit 0 means the operation
197 completed; inspect ``merge_base`` (null vs. a commit ID) for the answer.
198 """
199 elapsed = start_timer()
200 fmt: str = args.fmt
201 commit_a: str = args.commit_a
202 commit_b: str = args.commit_b
203
204 if fmt not in _FORMAT_CHOICES:
205 # For an invalid format we cannot know whether the caller wants JSON,
206 # so fall back to stderr prose (safest for shell scripts).
207 print(
208 f"❌ Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}",
209 file=sys.stderr,
210 )
211 raise SystemExit(ExitCode.USER_ERROR)
212
213 root = require_repo()
214
215 resolved_a = _resolve_ref(root, commit_a)
216 if resolved_a is None:
217 _emit_error(fmt, f"Cannot resolve ref: {commit_a!r}", ExitCode.USER_ERROR)
218
219 resolved_b = _resolve_ref(root, commit_b)
220 if resolved_b is None:
221 _emit_error(fmt, f"Cannot resolve ref: {commit_b!r}", ExitCode.USER_ERROR)
222
223 try:
224 base = find_merge_base(root, resolved_a, resolved_b)
225 except Exception as exc:
226 logger.debug("merge-base DAG walk failed: %s", exc)
227 _emit_error(fmt, str(exc), ExitCode.INTERNAL_ERROR)
228
229 if fmt == "text":
230 if base is None:
231 print("(no common ancestor)")
232 else:
233 print(base)
234 return
235
236 if base is None:
237 print(json.dumps(_MergeBaseNotFoundJson(
238 commit_a=resolved_a,
239 commit_b=resolved_b,
240 merge_base=None,
241 error="no common ancestor",
242 duration_ms=elapsed(),
243 exit_code=0,
244 )))
245 return
246
247 print(json.dumps(_MergeBaseFoundJson(
248 commit_a=resolved_a,
249 commit_b=resolved_b,
250 merge_base=base,
251 duration_ms=elapsed(),
252 exit_code=0,
253 )))
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 142 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 145 days ago