merge_base.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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 | import argparse |
| 48 | import json |
| 49 | import logging |
| 50 | import pathlib |
| 51 | import sys |
| 52 | from typing import TypedDict |
| 53 | |
| 54 | from muse.core.envelope import EnvelopeJson, make_envelope |
| 55 | from muse.core.errors import ExitCode |
| 56 | from muse.core.merge_engine import find_merge_base |
| 57 | from muse.core.repo import require_repo |
| 58 | from muse.core.refs import ( |
| 59 | get_head_commit_id, |
| 60 | read_current_branch, |
| 61 | ) |
| 62 | from muse.core.commits import read_commit |
| 63 | from muse.core.validation import validate_object_id |
| 64 | from muse.core.timing import start_timer |
| 65 | |
| 66 | logger = logging.getLogger(__name__) |
| 67 | |
| 68 | # --------------------------------------------------------------------------- |
| 69 | # Wire-format TypedDicts |
| 70 | # --------------------------------------------------------------------------- |
| 71 | |
| 72 | class _MergeBaseFoundJson(EnvelopeJson): |
| 73 | """Stable JSON envelope when a merge base is found.""" |
| 74 | commit_a: str |
| 75 | commit_b: str |
| 76 | merge_base: str # sha256:… commit ID |
| 77 | |
| 78 | class _MergeBaseNotFoundJson(EnvelopeJson): |
| 79 | """Stable JSON envelope when no common ancestor exists.""" |
| 80 | commit_a: str |
| 81 | commit_b: str |
| 82 | merge_base: None |
| 83 | error: str # "no common ancestor" |
| 84 | |
| 85 | class _MergeBaseErrorJson(EnvelopeJson): |
| 86 | """Error payload for usage/internal errors in --json mode.""" |
| 87 | status: str # "error" |
| 88 | error: str |
| 89 | |
| 90 | # --------------------------------------------------------------------------- |
| 91 | # Helpers |
| 92 | # --------------------------------------------------------------------------- |
| 93 | |
| 94 | def _emit_error(json_out: bool, msg: str, code: ExitCode, elapsed: float) -> None: |
| 95 | """Print an error and raise SystemExit. Never returns. |
| 96 | |
| 97 | In ``--json`` mode the error goes to stdout as a JSON payload so agents |
| 98 | always get parseable output. In text mode it goes to stderr. |
| 99 | """ |
| 100 | if json_out: |
| 101 | print(json.dumps(_MergeBaseErrorJson( |
| 102 | **make_envelope(elapsed, exit_code=int(code)), |
| 103 | status="error", |
| 104 | error=msg, |
| 105 | ))) |
| 106 | else: |
| 107 | print(f"❌ {msg}", file=sys.stderr) |
| 108 | raise SystemExit(code) |
| 109 | |
| 110 | def _resolve_ref(root: pathlib.Path, ref: str) -> str | None: |
| 111 | """Resolve a branch name, HEAD, or sha256-prefixed commit ID. |
| 112 | |
| 113 | Returns ``None`` when the ref cannot be resolved to a known commit. |
| 114 | """ |
| 115 | if ref.upper() == "HEAD": |
| 116 | branch = read_current_branch(root) |
| 117 | return get_head_commit_id(root, branch) |
| 118 | |
| 119 | # Try as branch name first. Guard against refs that are not valid branch |
| 120 | # names (e.g. sha256:-prefixed commit IDs) — get_head_commit_id calls |
| 121 | # validate_branch_name which raises ValueError for colons and slashes. |
| 122 | try: |
| 123 | cid = get_head_commit_id(root, ref) |
| 124 | if cid is not None: |
| 125 | return cid |
| 126 | except (ValueError, OSError): |
| 127 | pass # not a valid branch name; fall through to commit-ID lookup |
| 128 | |
| 129 | # Try as full sha256-prefixed commit ID. |
| 130 | try: |
| 131 | validate_object_id(ref) |
| 132 | record = read_commit(root, ref) |
| 133 | return record.commit_id if record else None |
| 134 | except ValueError: |
| 135 | return None |
| 136 | |
| 137 | # --------------------------------------------------------------------------- |
| 138 | # Registration |
| 139 | # --------------------------------------------------------------------------- |
| 140 | |
| 141 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 142 | """Register the merge-base subcommand.""" |
| 143 | parser = subparsers.add_parser( |
| 144 | "merge-base", |
| 145 | help="Find the lowest common ancestor of two commits.", |
| 146 | description=__doc__, |
| 147 | ) |
| 148 | parser.add_argument( |
| 149 | "commit_a", |
| 150 | help="First commit ID, branch name, or HEAD.", |
| 151 | ) |
| 152 | parser.add_argument( |
| 153 | "commit_b", |
| 154 | help="Second commit ID, branch name, or HEAD.", |
| 155 | ) |
| 156 | parser.add_argument( |
| 157 | "--json", "-j", |
| 158 | action="store_true", |
| 159 | dest="json_out", |
| 160 | help="Emit machine-readable JSON.", |
| 161 | ) |
| 162 | parser.set_defaults(func=run) |
| 163 | |
| 164 | # --------------------------------------------------------------------------- |
| 165 | # Entry point |
| 166 | # --------------------------------------------------------------------------- |
| 167 | |
| 168 | def run(args: argparse.Namespace) -> None: |
| 169 | """Find the lowest common ancestor of two commits. |
| 170 | |
| 171 | Accepts sha256-prefixed commit IDs, branch names, or ``HEAD``. The result |
| 172 | is the commit reachable from both inputs closest to both tips — the |
| 173 | divergence point between two histories. ``merge_base`` is ``null`` when |
| 174 | no common ancestor exists (disconnected histories). |
| 175 | |
| 176 | Agent quickstart |
| 177 | ---------------- |
| 178 | :: |
| 179 | |
| 180 | muse merge-base dev feat/billing --json |
| 181 | muse merge-base HEAD feat/x --json |
| 182 | muse merge-base sha256:<a> sha256:<b> --json |
| 183 | |
| 184 | JSON fields |
| 185 | ----------- |
| 186 | commit_a Resolved commit ID of the first input. |
| 187 | commit_b Resolved commit ID of the second input. |
| 188 | merge_base Common ancestor commit ID, or ``null`` if none exists. |
| 189 | |
| 190 | Exit codes |
| 191 | ---------- |
| 192 | 0 Success (check ``merge_base`` — ``null`` means no common ancestor). |
| 193 | 1 Cannot resolve one of the refs, or invalid ``--format``. |
| 194 | 3 Internal error during DAG walk. |
| 195 | """ |
| 196 | elapsed = start_timer() |
| 197 | json_out: bool = args.json_out |
| 198 | commit_a: str = args.commit_a |
| 199 | commit_b: str = args.commit_b |
| 200 | |
| 201 | root = require_repo() |
| 202 | |
| 203 | resolved_a = _resolve_ref(root, commit_a) |
| 204 | if resolved_a is None: |
| 205 | _emit_error(json_out, f"Cannot resolve ref: {commit_a!r}", ExitCode.USER_ERROR, elapsed) |
| 206 | |
| 207 | resolved_b = _resolve_ref(root, commit_b) |
| 208 | if resolved_b is None: |
| 209 | _emit_error(json_out, f"Cannot resolve ref: {commit_b!r}", ExitCode.USER_ERROR, elapsed) |
| 210 | |
| 211 | try: |
| 212 | base = find_merge_base(root, resolved_a, resolved_b) |
| 213 | except Exception as exc: |
| 214 | logger.debug("merge-base DAG walk failed: %s", exc) |
| 215 | _emit_error(json_out, str(exc), ExitCode.INTERNAL_ERROR, elapsed) |
| 216 | |
| 217 | if not json_out: |
| 218 | if base is None: |
| 219 | print("(no common ancestor)") |
| 220 | else: |
| 221 | print(base) |
| 222 | return |
| 223 | |
| 224 | if base is None: |
| 225 | print(json.dumps(_MergeBaseNotFoundJson( |
| 226 | **make_envelope(elapsed), |
| 227 | commit_a=resolved_a, |
| 228 | commit_b=resolved_b, |
| 229 | merge_base=None, |
| 230 | error="no common ancestor", |
| 231 | ))) |
| 232 | return |
| 233 | |
| 234 | print(json.dumps(_MergeBaseFoundJson( |
| 235 | **make_envelope(elapsed), |
| 236 | commit_a=resolved_a, |
| 237 | commit_b=resolved_b, |
| 238 | merge_base=base, |
| 239 | ))) |