aaronrene / gabriel-muse public
merge_base.py python
166 lines 4.7 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
1 """muse plumbing 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 }
14
15 When no common ancestor exists::
16
17 {
18 "commit_a": "<sha256>",
19 "commit_b": "<sha256>",
20 "merge_base": null,
21 "error": "no common ancestor"
22 }
23
24 Plumbing contract
25 -----------------
26
27 - Exit 0: operation completed (check ``merge_base`` field for null vs. found).
28 - Exit 1: a commit ID or ref cannot be resolved; bad ``--format`` value.
29 - Exit 3: DAG walk failed (I/O error or malformed graph).
30 """
31
32 from __future__ import annotations
33
34 import argparse
35 import json
36 import logging
37 import pathlib
38 import sys
39
40 from muse.core.errors import ExitCode
41 from muse.core.merge_engine import find_merge_base
42 from muse.core.repo import require_repo
43 from muse.core.store import get_head_commit_id, read_commit, read_current_branch
44 from muse.core.validation import validate_object_id
45
46 logger = logging.getLogger(__name__)
47
48 _FORMAT_CHOICES = ("json", "text")
49
50
51 def _resolve_ref(root: pathlib.Path, ref: str) -> str | None:
52 """Resolve a branch name, HEAD, or full 64-char commit ID to a commit ID.
53
54 Returns ``None`` when the ref cannot be resolved to a known commit.
55 """
56 if ref.upper() == "HEAD":
57 muse_dir = root / ".muse"
58 branch = read_current_branch(root)
59 return get_head_commit_id(root, branch)
60
61 # Try as branch name first.
62 cid = get_head_commit_id(root, ref)
63 if cid is not None:
64 return cid
65
66 # Try as full commit ID.
67 try:
68 validate_object_id(ref)
69 record = read_commit(root, ref)
70 return record.commit_id if record else None
71 except ValueError:
72 return None
73
74
75 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
76 """Register the merge-base subcommand."""
77 parser = subparsers.add_parser(
78 "merge-base",
79 help="Find the lowest common ancestor of two commits.",
80 description=__doc__,
81 )
82 parser.add_argument(
83 "commit_a",
84 help="First commit ID, branch name, or HEAD.",
85 )
86 parser.add_argument(
87 "commit_b",
88 help="Second commit ID, branch name, or HEAD.",
89 )
90 parser.add_argument(
91 "--format", "-f",
92 dest="fmt",
93 default="json",
94 metavar="FORMAT",
95 help="Output format: json or text. (default: json)",
96 )
97 parser.add_argument(
98 "--json", action="store_const", const="json", dest="fmt",
99 help="Shorthand for --format json."
100 )
101 parser.set_defaults(func=run)
102
103
104 def run(args: argparse.Namespace) -> None:
105 """Find the lowest common ancestor of two commits.
106
107 Accepts full SHA-256 commit IDs, branch names, or ``HEAD``. The result is
108 the commit that is reachable from both inputs and is closest to both tips —
109 the point at which their histories diverged.
110 """
111 fmt: str = args.fmt
112 commit_a: str = args.commit_a
113 commit_b: str = args.commit_b
114
115 if fmt not in _FORMAT_CHOICES:
116 print(
117 json.dumps({"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"}),
118 file=sys.stderr,
119 )
120 raise SystemExit(ExitCode.USER_ERROR)
121
122 root = require_repo()
123 muse_dir = root / ".muse"
124
125 resolved_a = _resolve_ref(root, commit_a)
126 if resolved_a is None:
127 print(json.dumps({"error": f"Cannot resolve ref: {commit_a!r}"}), file=sys.stderr)
128 raise SystemExit(ExitCode.USER_ERROR)
129
130 resolved_b = _resolve_ref(root, commit_b)
131 if resolved_b is None:
132 print(json.dumps({"error": f"Cannot resolve ref: {commit_b!r}"}), file=sys.stderr)
133 raise SystemExit(ExitCode.USER_ERROR)
134
135 try:
136 base = find_merge_base(root, resolved_a, resolved_b)
137 except Exception as exc:
138 logger.debug("merge-base DAG walk failed: %s", exc)
139 print(json.dumps({"error": str(exc)}), file=sys.stderr)
140 raise SystemExit(ExitCode.INTERNAL_ERROR)
141
142 if fmt == "text":
143 if base is None:
144 print("(no common ancestor)")
145 else:
146 print(base)
147 return
148
149 if base is None:
150 print(
151 json.dumps({
152 "commit_a": resolved_a,
153 "commit_b": resolved_b,
154 "merge_base": None,
155 "error": "no common ancestor",
156 })
157 )
158 return
159
160 print(
161 json.dumps({
162 "commit_a": resolved_a,
163 "commit_b": resolved_b,
164 "merge_base": base,
165 })
166 )
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 33 days ago