merge_base.py
python
| 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 json |
| 35 | import logging |
| 36 | import pathlib |
| 37 | |
| 38 | import typer |
| 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 | app = typer.Typer() |
| 49 | |
| 50 | _FORMAT_CHOICES = ("json", "text") |
| 51 | |
| 52 | |
| 53 | def _resolve_ref(root: pathlib.Path, ref: str) -> str | None: |
| 54 | """Resolve a branch name, HEAD, or full 64-char commit ID to a commit ID. |
| 55 | |
| 56 | Returns ``None`` when the ref cannot be resolved to a known commit. |
| 57 | """ |
| 58 | if ref.upper() == "HEAD": |
| 59 | branch = read_current_branch(root) |
| 60 | return get_head_commit_id(root, branch) |
| 61 | |
| 62 | # Try as branch name first. |
| 63 | cid = get_head_commit_id(root, ref) |
| 64 | if cid is not None: |
| 65 | return cid |
| 66 | |
| 67 | # Try as full commit ID. |
| 68 | try: |
| 69 | validate_object_id(ref) |
| 70 | record = read_commit(root, ref) |
| 71 | return record.commit_id if record else None |
| 72 | except ValueError: |
| 73 | return None |
| 74 | |
| 75 | |
| 76 | @app.callback(invoke_without_command=True) |
| 77 | def merge_base( |
| 78 | ctx: typer.Context, |
| 79 | commit_a: str = typer.Argument(..., help="First commit ID, branch name, or HEAD."), |
| 80 | commit_b: str = typer.Argument(..., help="Second commit ID, branch name, or HEAD."), |
| 81 | fmt: str = typer.Option( |
| 82 | "json", "--format", "-f", help="Output format: json or text." |
| 83 | ), |
| 84 | ) -> None: |
| 85 | """Find the lowest common ancestor of two commits. |
| 86 | |
| 87 | Accepts full SHA-256 commit IDs, branch names, or ``HEAD``. The result is |
| 88 | the commit that is reachable from both inputs and is closest to both tips — |
| 89 | the point at which their histories diverged. |
| 90 | |
| 91 | Use this to compute how far apart two branches are before merging, or to |
| 92 | identify the base for a ``snapshot-diff`` range query. |
| 93 | """ |
| 94 | if fmt not in _FORMAT_CHOICES: |
| 95 | typer.echo( |
| 96 | json.dumps( |
| 97 | {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"} |
| 98 | ) |
| 99 | ) |
| 100 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 101 | |
| 102 | root = require_repo() |
| 103 | |
| 104 | resolved_a = _resolve_ref(root, commit_a) |
| 105 | if resolved_a is None: |
| 106 | typer.echo(json.dumps({"error": f"Cannot resolve ref: {commit_a!r}"})) |
| 107 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 108 | |
| 109 | resolved_b = _resolve_ref(root, commit_b) |
| 110 | if resolved_b is None: |
| 111 | typer.echo(json.dumps({"error": f"Cannot resolve ref: {commit_b!r}"})) |
| 112 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 113 | |
| 114 | try: |
| 115 | base = find_merge_base(root, resolved_a, resolved_b) |
| 116 | except Exception as exc: |
| 117 | logger.debug("merge-base DAG walk failed: %s", exc) |
| 118 | typer.echo(json.dumps({"error": str(exc)})) |
| 119 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 120 | |
| 121 | if fmt == "text": |
| 122 | if base is None: |
| 123 | typer.echo("(no common ancestor)") |
| 124 | else: |
| 125 | typer.echo(base) |
| 126 | return |
| 127 | |
| 128 | if base is None: |
| 129 | typer.echo( |
| 130 | json.dumps( |
| 131 | { |
| 132 | "commit_a": resolved_a, |
| 133 | "commit_b": resolved_b, |
| 134 | "merge_base": None, |
| 135 | "error": "no common ancestor", |
| 136 | } |
| 137 | ) |
| 138 | ) |
| 139 | return |
| 140 | |
| 141 | typer.echo( |
| 142 | json.dumps( |
| 143 | { |
| 144 | "commit_a": resolved_a, |
| 145 | "commit_b": resolved_b, |
| 146 | "merge_base": base, |
| 147 | } |
| 148 | ) |
| 149 | ) |