"""``muse rebase`` — replay commits from one branch onto another. Muse rebase is cherry-pick of a range: it takes the commits unique to the current branch (those not reachable from the upstream) and replays them one-by-one on top of the upstream. Because commits are content-addressed, each replayed commit gets a new ID — the originals are untouched in the store. Usage:: muse rebase # replay HEAD's unique commits onto upstream muse rebase --onto # replay onto a different base muse rebase --squash [] # collapse all commits into one muse rebase --squash -m "msg" [] # squash with explicit commit message muse rebase --dry-run # preview which commits would be replayed muse rebase --status # show progress of an in-progress rebase muse rebase --abort # restore original HEAD muse rebase --continue # resume after resolving a conflict muse rebase --max-commits N # cap the number of commits replayed All subcommands accept ``--json`` for machine-readable output:: muse rebase --json main muse rebase --abort --json muse rebase --status --json Exit codes:: 0 — success (completed, aborted, dry-run, status) 1 — conflict encountered, bad arguments, or user error 3 — internal error """ from __future__ import annotations import argparse import datetime import json import logging import pathlib import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.merge_engine import find_merge_base, write_merge_state from muse.core.rebase import ( RebaseState, _write_branch_ref, clear_rebase_state, collect_commits_to_replay, get_rebase_progress, load_rebase_state, replay_one, save_rebase_state, ) from muse.core.reflog import append_reflog from muse.core.repo import read_repo_id, require_repo from muse.core.snapshot import compute_commit_id, compute_snapshot_id, directories_from_manifest from muse.core.store import ( Manifest, CommitRecord, SnapshotRecord, get_head_commit_id, read_commit, read_current_branch, read_snapshot, resolve_commit_ref, write_commit, write_snapshot, ) from muse.core.validation import sanitize_display, validate_branch_name from muse.core.workdir import apply_manifest from muse.domain import MuseDomainPlugin, SnapshotManifest from muse.plugins.registry import read_domain, resolve_plugin logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # JSON wire formats # --------------------------------------------------------------------------- class _RebaseResultJson(TypedDict): """JSON output for a completed rebase (normal or squash).""" status: str # "completed" | "conflict" | "aborted" | "up_to_date" | "dry_run" branch: str new_head: str | None onto: str squash: bool replayed: int conflicts: list[str] class _RebaseStatusJson(TypedDict): """JSON output for ``muse rebase --status``.""" active: bool original_branch: str original_head: str onto: str total: int done: int remaining: int squash: bool class _RebaseDryRunCommitJson(TypedDict): """One entry in the ``--dry-run`` commits list.""" commit_id: str message: str class _RebaseDryRunJson(TypedDict): """JSON output for ``muse rebase --dry-run``.""" branch: str onto: str commits: list[_RebaseDryRunCommitJson] count: int squash: bool # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _resolve_ref_to_id( root: pathlib.Path, repo_id: str, branch: str, ref: str, ) -> str | None: """Resolve a ref string (branch name, commit SHA, or HEAD) to a commit ID. Branch names are validated with ``validate_branch_name`` before being used as path components to prevent directory traversal attacks. """ if ref.upper() == "HEAD": return get_head_commit_id(root, branch) # Try as a branch ref — validate before using as a path component. is_valid_branch = True try: validate_branch_name(ref) except ValueError: is_valid_branch = False if is_valid_branch: ref_path = root / ".muse" / "refs" / "heads" / ref if ref_path.exists(): raw = ref_path.read_text(encoding="utf-8").strip() if raw and len(raw) == 64 and all(c in "0123456789abcdef" for c in raw): return raw # Fall back to commit SHA prefix resolution. rec = resolve_commit_ref(root, repo_id, branch, ref) return rec.commit_id if rec else None def _run_replay_loop( root: pathlib.Path, state: RebaseState, repo_id: str, branch: str, plugin: MuseDomainPlugin, domain: str, output_json: bool, ) -> bool: """Run the replay loop. Emits progress text (or NDJSON per commit when *output_json* is True) to stdout and writes ``MERGE_STATE.json`` on conflict. Returns: ``True`` if all commits were replayed cleanly; ``False`` on conflict. """ current_parent = state["completed"][-1] if state["completed"] else state["onto"] while state["remaining"]: orig_commit_id = state["remaining"][0] commit = read_commit(root, orig_commit_id) if commit is None: logger.warning("⚠️ Commit %s not found — skipping.", orig_commit_id[:12]) state["remaining"].pop(0) save_rebase_state(root, state) continue if not output_json: total = len(state["completed"]) + len(state["remaining"]) done = len(state["completed"]) + 1 print( f" [{done}/{total}] Replaying {orig_commit_id[:12]}: " f"{sanitize_display(commit.message)}" ) result = replay_one( root, commit, current_parent, plugin, domain, repo_id, branch ) if isinstance(result, list): # Conflict — write state and pause. state["remaining"].insert(0, orig_commit_id) save_rebase_state(root, state) write_merge_state( root, base_commit=commit.parent_commit_id or "", ours_commit=current_parent, theirs_commit=orig_commit_id, conflict_paths=result, ) if output_json: result_payload = _RebaseResultJson( status="conflict", branch=branch, new_head=None, onto=state["onto"], squash=False, replayed=len(state["completed"]), conflicts=sorted(result), ) print(json.dumps(result_payload)) else: print(f"\n❌ Rebase stopped at {orig_commit_id[:12]} due to conflict(s):", file=sys.stderr) for p in sorted(result): print(f" CONFLICT: {p}", file=sys.stderr) print( "\nResolve conflicts then run:\n" " muse rebase --continue to resume\n" " muse rebase --abort to restore original HEAD", file=sys.stderr, ) return False # Clean replay — advance. current_parent = result.commit_id state["remaining"].pop(0) state["completed"].append(result.commit_id) save_rebase_state(root, state) old_id = ( state["completed"][-2] if len(state["completed"]) >= 2 else state["onto"] ) append_reflog( root, branch, old_id=old_id, new_id=result.commit_id, author="user", operation=f"rebase: replayed {orig_commit_id[:12]} onto {state['onto'][:12]}", ) return True # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse rebase`` subcommand and all its flags.""" parser = subparsers.add_parser( "rebase", help="Replay commits from the current branch onto a new base.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "upstream", nargs="?", default=None, metavar="UPSTREAM", help="Branch or commit to rebase onto.", ) parser.add_argument( "--onto", default=None, metavar="NEWBASE", help="Replay commits onto this base instead of upstream.", ) parser.add_argument( "--squash", action="store_true", help="Collapse all replayed commits into one.", ) parser.add_argument( "--message", "-m", dest="squash_message", default=None, metavar="MSG", help="Commit message for the squashed commit (only with --squash).", ) parser.add_argument( "--abort", action="store_true", help="Abort an in-progress rebase and restore original HEAD.", ) parser.add_argument( "--continue", dest="continue_", action="store_true", help="Resume a paused rebase after resolving conflicts.", ) parser.add_argument( "-n", "--dry-run", dest="dry_run", action="store_true", help="Show which commits would be replayed without actually replaying them.", ) parser.add_argument( "--status", action="store_true", help="Show progress of an in-progress rebase.", ) parser.add_argument( "--max-commits", type=int, default=10_000, dest="max_commits", metavar="N", help="Maximum number of commits to replay (default: 10 000).", ) parser.add_argument( "--json", action="store_true", dest="output_json", help="Emit machine-readable JSON on stdout.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Main handler # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Replay commits from the current branch onto a new base. The most common invocation replays all commits unique to the current branch on top of *upstream*'s HEAD:: muse rebase main # rebase current branch onto main muse rebase --json main # same, machine-readable Use ``--onto`` when you need to replay onto a commit that is not the tip of *upstream*:: muse rebase --onto newbase upstream Use ``--squash`` to collapse all replayed commits into one:: muse rebase --squash main muse rebase --squash -m "feat: combined" main Use ``--dry-run`` to preview which commits would be replayed:: muse rebase --dry-run main muse rebase --dry-run --json main Use ``--status`` to inspect an in-progress rebase:: muse rebase --status muse rebase --status --json When a conflict is encountered, the rebase pauses. Resolve the conflict, stage the resolved files, then:: muse rebase --continue Or discard the entire rebase:: muse rebase --abort JSON output ----------- All outcomes emit a consistent payload when ``--json`` is supplied: Completed rebase:: {"status":"completed","branch":"feat/x","new_head":"","onto":"", "squash":false,"replayed":3,"conflicts":[]} Conflict:: {"status":"conflict","branch":"feat/x","new_head":null,"onto":"", "squash":false,"replayed":1,"conflicts":["path/to/file"]} Aborted:: {"status":"aborted","branch":"feat/x","new_head":"","onto":"", "squash":false,"replayed":0,"conflicts":[]} Already up to date:: {"status":"up_to_date","branch":"feat/x","new_head":"","onto":"", "squash":false,"replayed":0,"conflicts":[]} Dry-run:: {"branch":"feat/x","onto":"","commits":[...],"count":3,"squash":false} Status:: {"active":true,"original_branch":"feat/x","original_head":"","onto":"", "total":5,"done":2,"remaining":3,"squash":false} """ upstream: str | None = args.upstream onto: str | None = getattr(args, "onto", None) squash: bool = args.squash squash_message: str | None = args.squash_message abort: bool = args.abort continue_: bool = args.continue_ dry_run: bool = args.dry_run show_status: bool = args.status max_commits: int = args.max_commits output_json: bool = args.output_json root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) # --status — does not need plugin/domain if show_status: progress = get_rebase_progress(root) if output_json: status_payload = _RebaseStatusJson( active=progress["active"], original_branch=progress["original_branch"], original_head=progress["original_head"], onto=progress["onto"], total=progress["total"], done=progress["done"], remaining=progress["remaining"], squash=progress["squash"], ) print(json.dumps(status_payload)) else: if not progress["active"]: print("No rebase in progress.") else: print( f"Rebase in progress on '{sanitize_display(progress['original_branch'])}'\n" f" onto: {progress['onto'][:12]}\n" f" progress: {progress['done']}/{progress['total']} commits done " f"({progress['remaining']} remaining)\n" f" squash: {progress['squash']}" ) return plugin = resolve_plugin(root) domain = read_domain(root) active_state = load_rebase_state(root) # --abort if abort: if active_state is None: print("❌ No rebase in progress.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) original_head = active_state["original_head"] original_branch = active_state["original_branch"] _write_branch_ref(root, original_branch, original_head) orig_commit = read_commit(root, original_head) if orig_commit: snap = read_snapshot(root, orig_commit.snapshot_id) if snap: apply_manifest(root, snap.manifest) append_reflog( root, original_branch, old_id=active_state["completed"][-1] if active_state["completed"] else active_state["onto"], new_id=original_head, author="user", operation="rebase: abort", ) clear_rebase_state(root) if output_json: result_payload = _RebaseResultJson( status="aborted", branch=original_branch, new_head=original_head, onto=active_state["onto"], squash=active_state["squash"], replayed=len(active_state["completed"]), conflicts=[], ) print(json.dumps(result_payload)) else: print(f"✅ Rebase aborted. HEAD restored to {original_head[:12]}.") return # --continue if continue_: if active_state is None: print("❌ No rebase in progress. Nothing to continue.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) current_parent = ( active_state["completed"][-1] if active_state["completed"] else active_state["onto"] ) orig_commit_id = active_state["remaining"][0] if active_state["remaining"] else "" orig_commit = read_commit(root, orig_commit_id) if orig_commit_id else None snap_result = plugin.snapshot(root) manifest: Manifest = snap_result["files"] manifest_dirs = directories_from_manifest(manifest) snapshot_id = compute_snapshot_id(manifest, manifest_dirs) committed_at = datetime.datetime.now(datetime.timezone.utc) message = orig_commit.message if orig_commit else "rebase: continued" new_commit_id = compute_commit_id( parent_ids=[current_parent] if current_parent else [], snapshot_id=snapshot_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=manifest, directories=manifest_dirs)) new_commit = CommitRecord( commit_id=new_commit_id, repo_id=repo_id, branch=branch, snapshot_id=snapshot_id, message=message, committed_at=committed_at, parent_commit_id=current_parent if current_parent else None, author=orig_commit.author if orig_commit else "", ) write_commit(root, new_commit) active_state["completed"].append(new_commit_id) if active_state["remaining"]: active_state["remaining"].pop(0) save_rebase_state(root, active_state) append_reflog( root, branch, old_id=current_parent, new_id=new_commit_id, author="user", operation=f"rebase: continue — replayed {orig_commit_id[:12] if orig_commit_id else '?'}", ) if not active_state["remaining"]: _write_branch_ref(root, branch, new_commit_id) clear_rebase_state(root) if output_json: result_payload = _RebaseResultJson( status="completed", branch=branch, new_head=new_commit_id, onto=active_state["onto"], squash=False, replayed=len(active_state["completed"]), conflicts=[], ) print(json.dumps(result_payload)) else: print(f"✅ Rebase complete. HEAD is now {new_commit_id[:12]}.") return clean = _run_replay_loop( root, active_state, repo_id, branch, plugin, domain, output_json ) if clean: final_id = active_state["completed"][-1] _write_branch_ref(root, branch, final_id) clear_rebase_state(root) if output_json: result_payload = _RebaseResultJson( status="completed", branch=branch, new_head=final_id, onto=active_state["onto"], squash=False, replayed=len(active_state["completed"]), conflicts=[], ) print(json.dumps(result_payload)) else: print(f"✅ Rebase complete. HEAD is now {final_id[:12]}.") return # New rebase if active_state is not None: print("❌ Rebase in progress. Use --continue or --abort.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if upstream is None: print("❌ Provide an upstream branch or commit to rebase onto.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) head_commit_id = get_head_commit_id(root, branch) if head_commit_id is None: print("❌ Current branch has no commits.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) upstream_id = _resolve_ref_to_id(root, repo_id, branch, upstream) if upstream_id is None: print( f"❌ Upstream '{sanitize_display(upstream)}' not found.", file=sys.stderr ) raise SystemExit(ExitCode.USER_ERROR) if onto is not None: onto_id = _resolve_ref_to_id(root, repo_id, branch, onto) if onto_id is None: print( f"❌ --onto '{sanitize_display(onto)}' not found.", file=sys.stderr ) raise SystemExit(ExitCode.USER_ERROR) else: onto_id = upstream_id merge_base_id = find_merge_base(root, head_commit_id, upstream_id) stop_at = merge_base_id or "" if head_commit_id == upstream_id or head_commit_id == onto_id: if output_json: result_payload = _RebaseResultJson( status="up_to_date", branch=branch, new_head=head_commit_id, onto=onto_id, squash=squash, replayed=0, conflicts=[], ) print(json.dumps(result_payload)) else: print("Already up to date.") return commits_to_replay = collect_commits_to_replay( root, stop_at, head_commit_id, max_commits=max_commits ) if not commits_to_replay: if output_json: result_payload = _RebaseResultJson( status="up_to_date", branch=branch, new_head=head_commit_id, onto=onto_id, squash=squash, replayed=0, conflicts=[], ) print(json.dumps(result_payload)) else: print("Already up to date.") return # --dry-run — show plan and exit if dry_run: if output_json: commit_entries = [ _RebaseDryRunCommitJson( commit_id=c.commit_id, message=sanitize_display(c.message), ) for c in commits_to_replay ] dry_payload = _RebaseDryRunJson( branch=branch, onto=onto_id, commits=commit_entries, count=len(commits_to_replay), squash=squash, ) print(json.dumps(dry_payload)) else: print( f"Would rebase {len(commits_to_replay)} commit(s) " f"onto {onto_id[:12]} (from {branch})" ) for c in commits_to_replay: print(f" {c.commit_id[:12]} {sanitize_display(c.message)}") return if not output_json: print( f"Rebasing {len(commits_to_replay)} commit(s) " f"onto {onto_id[:12]} (from {branch})" ) # Squash mode if squash: current_parent = onto_id squash_manifest: Manifest = {} onto_commit = read_commit(root, onto_id) if onto_commit: onto_snap = read_snapshot(root, onto_commit.snapshot_id) if onto_snap: squash_manifest = dict(onto_snap.manifest) conflict_occurred = False conflict_paths: list[str] = [] for commit in commits_to_replay: base_manifest: Manifest = {} if commit.parent_commit_id: pc = read_commit(root, commit.parent_commit_id) if pc: ps = read_snapshot(root, pc.snapshot_id) if ps: base_manifest = ps.manifest theirs_snap = read_snapshot(root, commit.snapshot_id) if theirs_snap is None: print( f"❌ Rebase aborted: snapshot {commit.snapshot_id[:8]} for " f"commit {commit.commit_id[:8]} ({commit.message!r}) is missing " "or corrupt. A squash with a missing snapshot would silently " "delete all files from that commit. " "Run `muse verify-pack` to audit the store.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) theirs_manifest = theirs_snap.manifest result = plugin.merge( SnapshotManifest(files=base_manifest, domain=domain, directories=directories_from_manifest(base_manifest)), SnapshotManifest(files=squash_manifest, domain=domain, directories=directories_from_manifest(squash_manifest)), SnapshotManifest(files=theirs_manifest, domain=domain, directories=directories_from_manifest(theirs_manifest)), repo_root=root, ) if not result.is_clean: conflict_paths = sorted(result.conflicts) if output_json: result_payload = _RebaseResultJson( status="conflict", branch=branch, new_head=None, onto=onto_id, squash=True, replayed=0, conflicts=conflict_paths, ) print(json.dumps(result_payload)) else: print( f"❌ Conflict during squash at {commit.commit_id[:12]}:", file=sys.stderr, ) for p in conflict_paths: print(f" CONFLICT: {p}", file=sys.stderr) print( "Resolve conflicts and try again. " "Squash does not support --continue.", file=sys.stderr, ) conflict_occurred = True break squash_manifest = result.merged["files"] if conflict_occurred: raise SystemExit(ExitCode.USER_ERROR) apply_manifest(root, squash_manifest) squash_dirs = directories_from_manifest(squash_manifest) snapshot_id = compute_snapshot_id(squash_manifest, squash_dirs) committed_at = datetime.datetime.now(datetime.timezone.utc) final_message = squash_message or commits_to_replay[-1].message new_commit_id = compute_commit_id( parent_ids=[onto_id], snapshot_id=snapshot_id, message=final_message, committed_at_iso=committed_at.isoformat(), ) write_snapshot( root, SnapshotRecord(snapshot_id=snapshot_id, manifest=squash_manifest, directories=squash_dirs) ) write_commit( root, CommitRecord( commit_id=new_commit_id, repo_id=repo_id, branch=branch, snapshot_id=snapshot_id, message=final_message, committed_at=committed_at, parent_commit_id=onto_id, ), ) _write_branch_ref(root, branch, new_commit_id) append_reflog( root, branch, old_id=head_commit_id, new_id=new_commit_id, author="user", operation=f"rebase --squash onto {onto_id[:12]}", ) if output_json: result_payload = _RebaseResultJson( status="completed", branch=branch, new_head=new_commit_id, onto=onto_id, squash=True, replayed=len(commits_to_replay), conflicts=[], ) print(json.dumps(result_payload)) else: print(f"✅ Squash-rebase complete. HEAD is now {new_commit_id[:12]}.") return # Normal replay loop state = RebaseState( original_branch=branch, original_head=head_commit_id, onto=onto_id, remaining=[c.commit_id for c in commits_to_replay], completed=[], squash=False, ) save_rebase_state(root, state) clean = _run_replay_loop(root, state, repo_id, branch, plugin, domain, output_json) if clean: final_id = state["completed"][-1] if state["completed"] else onto_id _write_branch_ref(root, branch, final_id) clear_rebase_state(root) append_reflog( root, branch, old_id=head_commit_id, new_id=final_id, author="user", operation=f"rebase: finished onto {onto_id[:12]}", ) if output_json: result_payload = _RebaseResultJson( status="completed", branch=branch, new_head=final_id, onto=onto_id, squash=False, replayed=len(state["completed"]), conflicts=[], ) print(json.dumps(result_payload)) else: print(f"✅ Rebase complete. HEAD is now {final_id[:12]}.")