"""``muse checkout`` — switch branches or restore working tree from a commit. Usage:: muse checkout — switch to existing branch muse checkout -b — create and switch to new branch muse checkout — detach HEAD at a specific commit muse checkout --ours — resolve conflict by accepting our version muse checkout --theirs — resolve conflict by accepting their version muse checkout --dry-run — show what would happen without switching Conflict resolution ------------------- After a ``muse merge`` that leaves conflicts, use ``--ours`` or ``--theirs`` to accept one side of each conflicted file. The file is restored from the appropriate commit in MERGE_STATE.json and removed from the unresolved conflict list. Once all conflicts are cleared, run ``muse commit`` to complete the merge. # Accept our version of AGENTS.md (feature branch wins) muse checkout --ours AGENTS.md # Accept their version of pyproject.toml (main branch wins) muse checkout --theirs pyproject.toml # Bulk-resolve every conflict to one side muse checkout --ours --all muse checkout --theirs --all # Check remaining conflicts muse status # Complete the merge muse commit -m "merge: resolve conflicts" JSON output (``--format json`` or ``--json``):: { "action": "created|switched|detached|already_on", "branch": " | null", "commit_id": "", "from_branch": "" } Exit codes:: 0 — success 1 — invalid branch name, branch not found, dirty working tree (without --force) 3 — internal error (missing snapshot or object) """ from __future__ import annotations import argparse import json import logging import pathlib import sys from muse.core.errors import ExitCode from muse.core.merge_engine import read_merge_state, write_merge_state from muse.core.object_store import has_object, read_object, restore_object from muse.core.repo import read_repo_id, require_repo from muse.core.snapshot import diff_workdir_vs_snapshot from muse.core.store import ( get_head_commit_id, get_head_snapshot_id, get_head_snapshot_manifest, read_commit, read_current_branch, read_snapshot, resolve_commit_ref, write_branch_ref, write_head_branch, write_head_commit, write_text_atomic, ) from muse.core.reflog import append_reflog from muse.core.validation import contain_path, sanitize_display, validate_branch_name from muse.cli.guard import require_clean_workdir from muse.cli.commands.stash import _stash_push_programmatic, _stash_pop_programmatic from muse.core.snapshot import directories_from_manifest from muse.core.cohen_transform import three_way_merge_lines from muse.domain import SnapshotManifest from muse.plugins.registry import read_domain, resolve_plugin from muse.core._types import Manifest logger = logging.getLogger(__name__) _CHECKOUT_HEAD_FILENAME = "CHECKOUT_HEAD" def _apply_autostash(root: pathlib.Path, fmt: str) -> None: """Pop the autostash and emit a git-idiomatic status line. Called unconditionally from the ``finally`` block in :func:`run` when an autostash was created. On conflict the stash is left intact at index 0 so the user can resolve and ``muse stash pop`` manually. Args: root: Repository root containing ``.muse/``. fmt: Output format (``"text"`` or ``"json"``). In JSON mode all autostash messages go to **stderr** so stdout stays machine- readable. """ try: entry = _stash_pop_programmatic(root) files_n = len(entry["delta"]) + len(entry["deleted"]) if fmt == "text": # Mirrors git: "Applied autostash." print(f"Applied autostash ({files_n} file(s) restored).") else: # Keep stdout clean; emit to stderr so scripts can detect it. print(f"autostash: applied ({files_n} file(s) restored)", file=sys.stderr) except ValueError as exc: # Empty stash or bad index — shouldn't happen since we just pushed, # but handle gracefully. msg = f"⚠️ autostash pop failed — stash may be empty: {exc}" print(msg, file=sys.stderr) except RuntimeError as exc: # Missing objects in store. Stash is still there at index 0. print( f"⚠️ autostash: could not restore changes — object store error: {exc}\n" f" Your changes are safe in stash@{{0}} — run 'muse stash pop' to restore them.", file=sys.stderr, ) def checkout_head_path(root: pathlib.Path) -> pathlib.Path: """Return the path to ``.muse/CHECKOUT_HEAD``. This file exists only while a checkout is in progress. Its presence after a crash means the working tree may be partially mutated and ``muse status`` should warn the user. """ return root / ".muse" / _CHECKOUT_HEAD_FILENAME def read_checkout_head(root: pathlib.Path) -> str | None: """Return the target branch recorded in ``.muse/CHECKOUT_HEAD``, or ``None``. ``None`` means no checkout is in progress (normal state). A non-``None`` value means a previous checkout was interrupted and the working tree may be partially mutated. """ p = checkout_head_path(root) try: return p.read_text(encoding="utf-8").strip() or None except FileNotFoundError: return None def _checkout_snapshot( root: pathlib.Path, target_snapshot_id: str, current_snapshot_id: str | None, *, target_branch: str | None = None, ) -> None: """Incrementally update the working tree from the current to the target snapshot. Uses the domain plugin to compute the delta between the two snapshots and only touches files that actually changed — removing deleted paths and restoring added/modified ones from the object store. Calls ``plugin.apply()`` as the domain-level post-checkout hook. Pre-flight integrity guarantee Before mutating any file on disk, every object that must be restored is verified to exist in the local store via :func:`has_object`. If any object is missing the function prints a diagnostic and raises ``SystemExit`` **without touching the working tree**. CHECKOUT_HEAD marker ``.muse/CHECKOUT_HEAD`` is written with *target_branch* (or the snapshot ID when no branch name is available) before the first file mutation. It is deleted atomically on success. If the process is killed mid-checkout ``muse status`` detects the marker and warns the user, preventing silent data loss from being misread as uncommitted deletions. Raises ``SystemExit(ExitCode.INTERNAL_ERROR)`` if the target snapshot or any required object is missing from the local store. """ plugin = resolve_plugin(root) domain = read_domain(root) target_snap_rec = read_snapshot(root, target_snapshot_id) if target_snap_rec is None: print( f"❌ Snapshot {target_snapshot_id[:8]} not found in object store.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) target_snap = SnapshotManifest( files=target_snap_rec.manifest, domain=domain, directories=list(target_snap_rec.directories), ) cur_rec = None if current_snapshot_id is not None: cur_rec = read_snapshot(root, current_snapshot_id) current_snap = ( SnapshotManifest( files=cur_rec.manifest, domain=domain, directories=list(cur_rec.directories), ) if cur_rec else SnapshotManifest(files={}, domain=domain, directories=[]) ) else: current_snap = SnapshotManifest(files={}, domain=domain, directories=[]) delta = plugin.diff(current_snap, target_snap) cur_manifest: Manifest = cur_rec.manifest if cur_rec else {} # ── Gap 1: Pre-flight object existence check ────────────────────────────── # Build the list of file paths that need to be restored (insert / replace / # patch ops whose address is a real file in the target manifest, not a # directory entry). Then verify every required object exists in the local # store BEFORE touching a single file on disk. If anything is missing we # abort here — zero mutations to the working tree. to_restore = [ op["address"] for op in delta["ops"] if op["op"] in ("insert", "replace", "patch") and op["address"] in target_snap_rec.manifest ] preflight_missing = [ rel_path for rel_path in to_restore if not has_object(root, target_snap_rec.manifest[rel_path]) ] if preflight_missing: for rel_path in preflight_missing: object_id = target_snap_rec.manifest[rel_path] print( f"❌ Object {object_id[:8]} for '{sanitize_display(rel_path)}' " "is missing from the local object store.", file=sys.stderr, ) print( f"❌ Checkout aborted: {len(preflight_missing)} object(s) not in local " "store. Fetch missing objects from the remote before retrying.\n" " Working tree was NOT modified.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # ── Gap 2: Write CHECKOUT_HEAD before first mutation ───────────────────── # Record the intended target so `muse status` can detect an interrupted # checkout and warn the user rather than silently reporting missing files # as uncommitted deletions. marker_path = checkout_head_path(root) marker_label = target_branch or target_snapshot_id write_text_atomic(marker_path, marker_label + "\n") # ── Remove files that no longer exist in the target snapshot ───────────── # Skip directory-level delete ops — only actual files (addresses present in # the current manifest) are unlinked here. removed = [ op["address"] for op in delta["ops"] if op["op"] == "delete" and op["address"] in cur_manifest ] for rel_path in removed: fp = root / rel_path if fp.exists() and fp.is_file(): fp.unlink() # ── Restore added and modified files from the content-addressed store ───── # All objects were verified present above; restore_object should not return # False here. A second-layer check is kept as a defensive catch for races # (e.g. another process pruning the store mid-checkout). restore_failures: list[str] = [] for rel_path in to_restore: object_id = target_snap_rec.manifest[rel_path] try: safe_dest = contain_path(root, rel_path) except ValueError as exc: logger.warning("⚠️ Skipping unsafe manifest path %r: %s", rel_path, exc) continue if not restore_object(root, object_id, safe_dest): restore_failures.append(rel_path) if restore_failures: for rel_path in restore_failures: object_id = target_snap_rec.manifest[rel_path] print( f"❌ Object {object_id[:8]} for '{sanitize_display(rel_path)}' " "disappeared from the store mid-checkout (store was modified " "concurrently?).", file=sys.stderr, ) print( f"❌ Checkout incomplete: {len(restore_failures)} object(s) could not " "be restored. Run `muse checkout ` to retry.", file=sys.stderr, ) # CHECKOUT_HEAD remains on disk — muse status will detect the # interrupted state and guide the user. raise SystemExit(ExitCode.INTERNAL_ERROR) # ── Domain-level post-checkout hook ─────────────────────────────────────── plugin.apply(delta, root) # ── Clear CHECKOUT_HEAD on success ──────────────────────────────────────── marker_path.unlink(missing_ok=True) def _checkout_with_merge( root: pathlib.Path, target: str, current_branch: str, repo_id: str, *, fmt: str, dry_run: bool, ) -> dict[str, object]: """Perform a branch switch carrying uncommitted changes via three-way merge. Implements ``muse checkout -m`` (the Cohen Transform checkout): 1. Snapshot the dirty working-tree files (modified + deleted tracked files). 2. Apply the target-branch snapshot to the working tree normally. 3. For each dirty file, three-way merge: base = HEAD version (what we last committed) ours = working-tree version (your uncommitted edit) theirs = target-branch version 4. Clean merges: write merged content silently. 5. Conflicts: write diff3 + Manyana-labeled markers inline; record in MERGE_STATE.json (same format as ``muse merge`` conflicts so ``muse checkout --ours/--theirs`` works identically). Binary files with conflicts are left at the target-branch version and listed as conflicts; the user must resolve them manually. Args: root: Repository root. target: Target branch name. current_branch: Current branch name. repo_id: Repository ID string. fmt: Output format (``'text'`` or ``'json'``). dry_run: When ``True``, report what would happen without writing. Returns: A result dict suitable for JSON serialisation with keys: ``action``, ``branch``, ``from_branch``, ``commit_id``, ``clean_merges``, ``conflicts``, ``dry_run``. """ # ── Capture current dirty state ─────────────────────────────────────────── current_snapshot_id = get_head_snapshot_id(root, repo_id, current_branch) head_manifest: dict[str, str] = {} if current_snapshot_id: snap_rec = read_snapshot(root, current_snapshot_id) if snap_rec: head_manifest = dict(snap_rec.manifest) _added, modified, deleted, _untracked, _ad, _dd = diff_workdir_vs_snapshot( root, head_manifest ) dirty_paths = modified | deleted # Read the working-tree content for every modified file BEFORE the # checkout overwrites the working tree. workdir_content: dict[str, bytes] = {} for rel_path in dirty_paths: fp = root / rel_path if fp.is_file(): workdir_content[rel_path] = fp.read_bytes() else: workdir_content[rel_path] = b"" # deleted if dry_run: target_commit_id = get_head_commit_id(root, target) or "" return { "dry_run": True, "action": "switched", "branch": target, "from_branch": current_branch, "commit_id": target_commit_id, "dirty_paths": sorted(dirty_paths), "clean_merges": [], "conflicts": [], } # ── Apply the target snapshot (switches the branch normally) ───────────── target_snapshot_id = get_head_snapshot_id(root, repo_id, target) if target_snapshot_id: _checkout_snapshot(root, target_snapshot_id, current_snapshot_id, target_branch=target) target_commit_id = get_head_commit_id(root, target) or "" write_head_branch(root, target) append_reflog( root, target, old_id=get_head_commit_id(root, current_branch) or None, new_id=target_commit_id or ("0" * 64), author="user", operation=( f"checkout -m: carrying changes from {sanitize_display(current_branch)} " f"to {sanitize_display(target)}" ), ) if not dirty_paths: return { "dry_run": False, "action": "switched", "branch": target, "from_branch": current_branch, "commit_id": target_commit_id, "clean_merges": [], "conflicts": [], } # ── Three-way merge each dirty file (the Cohen Transform) ───────────────── # Read the target snapshot manifest so we know theirs content. target_manifest: dict[str, str] = {} if target_snapshot_id: target_snap_rec = read_snapshot(root, target_snapshot_id) if target_snap_rec: target_manifest = dict(target_snap_rec.manifest) clean_merges: list[str] = [] conflicts: list[str] = [] ours_label = sanitize_display(current_branch) theirs_label = sanitize_display(target) for rel_path in sorted(dirty_paths): # base = HEAD version (from the object store) base_oid = head_manifest.get(rel_path) base_bytes = read_object(root, base_oid) if base_oid else b"" base_bytes = base_bytes or b"" # ours = working-tree content captured before checkout ours_bytes = workdir_content.get(rel_path, b"") # theirs = target-branch version (now on disk after checkout) theirs_oid = target_manifest.get(rel_path) theirs_bytes = read_object(root, theirs_oid) if theirs_oid else b"" theirs_bytes = theirs_bytes or b"" # Detect binary files: skip line-merge, treat as conflict. def _is_binary(b: bytes) -> bool: return b"\x00" in b[:8000] if _is_binary(base_bytes) or _is_binary(ours_bytes) or _is_binary(theirs_bytes): # Binary conflict: leave theirs (already on disk), record conflict. conflicts.append(rel_path) continue # Line-level three-way merge via the Cohen Transform. base_lines = base_bytes.decode("utf-8", errors="replace").splitlines(keepends=True) ours_lines = ours_bytes.decode("utf-8", errors="replace").splitlines(keepends=True) theirs_lines = theirs_bytes.decode("utf-8", errors="replace").splitlines(keepends=True) merged_lines, has_conflict = three_way_merge_lines( base_lines, ours_lines, theirs_lines, label_ours=f"{ours_label} (uncommitted)", label_base="base (HEAD)", label_theirs=f"{theirs_label}", ) dest = root / rel_path dest.parent.mkdir(parents=True, exist_ok=True) write_text_atomic(dest, "".join(merged_lines)) if has_conflict: conflicts.append(rel_path) else: clean_merges.append(rel_path) # ── Record MERGE_STATE when conflicts remain ─────────────────────────────── if conflicts: ours_commit_id = get_head_commit_id(root, current_branch) or "" write_merge_state( root, base_commit=ours_commit_id, ours_commit=ours_commit_id, theirs_commit=target_commit_id, conflict_paths=conflicts, other_branch=target, ) return { "dry_run": False, "action": "switched", "branch": target, "from_branch": current_branch, "commit_id": target_commit_id, "clean_merges": clean_merges, "conflicts": conflicts, } def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse checkout`` subcommand and all its flags.""" parser = subparsers.add_parser( "checkout", help="Switch branches or restore working tree from a commit.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "target", nargs="?", help="Branch name, commit ID, or (with --ours/--theirs) file path to resolve.", ) parser.add_argument( "-b", "--create", action="store_true", help="Create a new branch at HEAD and switch to it.", ) parser.add_argument( "--force", "-f", action="store_true", help="Discard uncommitted changes and force the switch.", ) parser.add_argument( "--dry-run", "-n", action="store_true", dest="dry_run", help=( "Show what would happen without making any changes. " "Exits 0 if the switch would succeed, 1 if it would be blocked." ), ) parser.add_argument( "--format", default="text", dest="fmt", help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.add_argument( "--ours", dest="resolve_ours", action="store_true", help=( "Resolve conflict(s) by restoring from our branch (HEAD at merge time). " "Use with a file path to resolve one path, or --all to resolve every conflict." ), ) parser.add_argument( "--theirs", dest="resolve_theirs", action="store_true", help=( "Resolve conflict(s) by restoring from the branch being merged in. " "Use with a file path to resolve one path, or --all to resolve every conflict." ), ) parser.add_argument( "--all", dest="resolve_all", action="store_true", help=( "Resolve ALL unresolved conflicts at once using --ours or --theirs. " "Requires exactly one of --ours or --theirs." ), ) parser.add_argument( "--merge", "-m", dest="merge", action="store_true", help=( "Carry uncommitted working-tree changes into the target branch using " "a three-way merge (the Cohen Transform). " "Files that merge cleanly are updated in-place; files that conflict " "receive diff3-style markers with Manyana action labels " "(e.g. ``<<<<<<< ours [deleted]``) and the merge is recorded in " "MERGE_STATE.json for resolution with " "``muse checkout --ours/--theirs``. " "Inspired by ``git checkout -m`` and Bram Cohen's Manyana project." ), ) parser.add_argument( "--autostash", dest="autostash", action="store_true", help=( "Automatically stash uncommitted changes before switching branches " "and pop them back onto the working tree after the switch succeeds. " "Equivalent to running ``muse stash`` before checkout and " "``muse stash pop`` after. " "Mutually exclusive with --force and --merge." ), ) parser.set_defaults(func=run, merge=False, autostash=False) def run(args: argparse.Namespace) -> None: """Switch branches or restore working tree from a commit. All error messages are written to **stderr**; stdout is reserved for structured output only. Agents should pass ``--format json`` to receive a machine-readable result:: { "action": "created|switched|detached|already_on", "branch": " | null", "commit_id": "", "from_branch": "" } ``--dry-run`` (``-n``) is agent-safe: it exits 0 when the switch would succeed and 1 when it would be blocked (dirty tree, missing snapshot, …). When combined with ``--format json`` the JSON payload includes ``"dry_run": true``. Conflict resolution:: # resolve one path muse checkout --ours src/billing.py muse checkout --theirs src/billing.py # resolve every conflict at once muse checkout --ours --all muse checkout --theirs --all Exit codes:: 0 — success 1 — invalid branch name, branch not found, dirty working tree (without --force) 3 — internal error (missing snapshot or object) """ target: str | None = args.target create: bool = args.create force: bool = args.force dry_run: bool = args.dry_run merge: bool = getattr(args, "merge", False) autostash: bool = getattr(args, "autostash", False) fmt: str = args.fmt resolve_ours: bool = args.resolve_ours resolve_theirs: bool = args.resolve_theirs resolve_all: bool = args.resolve_all if fmt not in ("text", "json"): print( f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if autostash and force: print("❌ --autostash and --force are mutually exclusive.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if autostash and merge: print("❌ --autostash and --merge are mutually exclusive.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Conflict resolution mode ────────────────────────────────────────────── if resolve_ours or resolve_theirs: if resolve_ours and resolve_theirs: print("❌ Cannot specify both --ours and --theirs.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not resolve_all and not target: print( "❌ Specify a file path, or use --all to resolve every conflict.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() merge_state = read_merge_state(root) if merge_state is None: print( "❌ No merge in progress. --ours/--theirs only work during a conflicted merge.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) side_label = "ours" if resolve_ours else "theirs" source_commit_id = merge_state.ours_commit if resolve_ours else merge_state.theirs_commit if not source_commit_id: print(f"❌ MERGE_STATE has no {side_label} commit recorded.", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) source_commit = read_commit(root, source_commit_id) if source_commit is None: print( f"❌ Source commit {source_commit_id[:8]} not found in object store.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) snap = read_snapshot(root, source_commit.snapshot_id) if snap is None: print( f"❌ Snapshot for commit {source_commit_id[:8]} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # ── Determine which conflict paths to resolve ───────────────────── if resolve_all: paths_to_resolve = list(merge_state.conflict_paths) if not paths_to_resolve: if fmt == "json": print(json.dumps({ "action": "conflict_resolved_all", "side": side_label, "resolved_count": 0, "remaining_conflicts": 0, })) else: print("✅ No conflicts to resolve — working tree is already clean.") return else: file_path = (target or "").replace("\\", "/") matching = [ p for p in merge_state.conflict_paths if p == file_path or p.startswith(f"{file_path}::") ] if not matching: print( f"ℹ️ '{sanitize_display(file_path)}' is not in the conflict list " f"— may already be resolved." ) return paths_to_resolve = matching # ── Apply resolutions ───────────────────────────────────────────── resolved: list[str] = [] for conflict_path in paths_to_resolve: # Extract the bare file path (strip symbol qualifier if present). bare_file = conflict_path.split("::")[0] if "::" in conflict_path else conflict_path if bare_file in snap.manifest: object_id = snap.manifest[bare_file] try: safe_dest = contain_path(root, bare_file) except ValueError as exc: print( f"❌ Unsafe path '{sanitize_display(bare_file)}': {exc}", file=sys.stderr, ) continue if not restore_object(root, object_id, safe_dest): print( f"❌ Object {object_id[:8]} for '{sanitize_display(bare_file)}' " f"not in local store.", file=sys.stderr, ) continue else: # File doesn't exist on the chosen side — delete it. target_path = root / bare_file if target_path.exists(): target_path.unlink() resolved.append(conflict_path) # ── Update MERGE_STATE ──────────────────────────────────────────── remaining = [p for p in merge_state.conflict_paths if p not in resolved] write_merge_state( root, base_commit=merge_state.base_commit or "", ours_commit=merge_state.ours_commit or "", theirs_commit=merge_state.theirs_commit or "", conflict_paths=remaining, other_branch=merge_state.other_branch, ) if resolve_all: if fmt == "json": print(json.dumps({ "action": "conflict_resolved_all", "side": side_label, "resolved_count": len(resolved), "remaining_conflicts": len(remaining), })) else: print(f"✅ Resolved {len(resolved)} conflict(s) using {side_label}.") if remaining: print( f" {len(remaining)} could not be resolved automatically " f"— run 'muse conflicts' to inspect." ) else: print(" All conflicts resolved. Run 'muse commit' to complete the merge.") else: file_path = (target or "").replace("\\", "/") if fmt == "json": print(json.dumps({ "action": "conflict_resolved", "file": file_path, "side": side_label, "remaining_conflicts": len(remaining), })) else: print(f"✅ Resolved '{sanitize_display(file_path)}' using {side_label}.") if remaining: print( f" {len(remaining)} conflict(s) remaining — " f"run 'muse conflicts' to see them." ) else: print(" All conflicts resolved. Run 'muse commit' to complete the merge.") return # ── End conflict resolution mode ───────────────────────────────────────── if target is None: print( "❌ Specify a branch, commit ID, or use --ours/--theirs with a file path.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() repo_id = read_repo_id(root) current_branch = read_current_branch(root) muse_dir = root / ".muse" # ── --merge: carry uncommitted changes via the Cohen Transform ──────────── # Bypass the clean-workdir guard and instead three-way merge each dirty # file into the target branch context. Only valid for branch switches # (not -b/--create, not detached-HEAD checkouts). if merge and not create and target is not None: try: validate_branch_name(target) except ValueError as exc: print( f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) ref_file = muse_dir / "refs" / "heads" / target if not ref_file.exists(): print( f"❌ Branch '{sanitize_display(target)}' not found. " f"--merge only works with existing branches.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if target == current_branch: print(f"Already on '{sanitize_display(target)}'") return result = _checkout_with_merge( root, target, current_branch, repo_id, fmt=fmt, dry_run=dry_run, ) conflicts: list[str] = result.get("conflicts", []) # type: ignore[assignment] clean_merges: list[str] = result.get("clean_merges", []) # type: ignore[assignment] if fmt == "json": print(json.dumps(result)) else: if dry_run: print(f"Would switch to branch '{sanitize_display(target)}' carrying changes") else: print(f"Switched to branch '{sanitize_display(target)}'") if clean_merges: print(f" Merged cleanly: {len(clean_merges)} file(s)") for p in clean_merges: print(f" ✔ {sanitize_display(p)}") if conflicts: print(f" Conflicts: {len(conflicts)} file(s)", file=sys.stderr) for p in conflicts: print(f" CONFLICT (both modified): {sanitize_display(p)}", file=sys.stderr) print( "\nFix conflicts then run 'muse commit' to complete.\n" "Run 'muse diff --conflict' to see what each side changed.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) return # ── End --merge mode ────────────────────────────────────────────────────── # ── Main checkout flow ──────────────────────────────────────────────────── # try/finally ensures the autostash is always popped — whether the switch # succeeds, fails validation, or raises SystemExit — so the user's work is # never stranded in the stash. autostash_entry = None try: # Creating a new branch starts at the current HEAD — no files change, # so dirty tracked files cannot be overwritten. The guard only fires # when switching to an *existing* branch whose snapshot differs. if not create: if autostash and not dry_run: # Git idiom: "WIP on : autostash" is the canonical # stash message. The push makes the tree clean so the switch # proceeds; the finally block pops unconditionally after. autostash_entry = _stash_push_programmatic( root, message=f"WIP on {sanitize_display(current_branch)}: autostash", ) if autostash_entry is not None and fmt == "text": print("Stashing local changes before checkout…") elif not autostash: # Normal path: block if dirty (unless --force bypasses). require_clean_workdir(root, "checkout", force=force) # autostash + dry_run: skip both — assume stash would succeed and # the switch would proceed, so the dry-run sees a clean state. current_snapshot_id = get_head_snapshot_id(root, repo_id, current_branch) if create: try: validate_branch_name(target) except ValueError as exc: print( f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) ref_file = muse_dir / "refs" / "heads" / target if ref_file.exists(): print( f"❌ Branch '{sanitize_display(target)}' already exists. " f"Use 'muse checkout {sanitize_display(target)}' to switch to it.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if dry_run: if fmt == "json": print(json.dumps({ "dry_run": True, "action": "created", "branch": target, "from_branch": current_branch, "commit_id": get_head_commit_id(root, current_branch) or "", })) else: print(f"Would create and switch to branch '{sanitize_display(target)}'") return current_commit = get_head_commit_id(root, current_branch) or "" if current_commit: write_branch_ref(root, target, current_commit) else: write_text_atomic(ref_file, "") write_head_branch(root, target) append_reflog( root, target, old_id=None, new_id=current_commit or ("0" * 64), author="user", operation=f"branch: created from {sanitize_display(current_branch)}", ) if fmt == "json": print(json.dumps({ "dry_run": False, "action": "created", "branch": target, "from_branch": current_branch, "commit_id": current_commit, })) else: print(f"Switched to a new branch '{sanitize_display(target)}'") return try: validate_branch_name(target) except ValueError as exc: print( f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) ref_file = muse_dir / "refs" / "heads" / target if ref_file.exists(): if target == current_branch: if force and current_snapshot_id: # --force on the current branch restores the working tree # to HEAD, overwriting modified/deleted tracked files. # Pass None as current_snapshot_id so _checkout_snapshot # diffs empty→HEAD, producing insert ops for every tracked # file. restore_object skips files already correct. if not dry_run: _checkout_snapshot(root, current_snapshot_id, None, target_branch=target) if fmt == "json": print(json.dumps({ "dry_run": dry_run, "action": "restored", "branch": target, "from_branch": current_branch, "commit_id": current_snapshot_id, })) else: if dry_run: print(f"Would restore working tree to HEAD of '{sanitize_display(target)}'") else: print(f"HEAD of '{sanitize_display(target)}' restored to working tree") return if fmt == "json": print(json.dumps({ "dry_run": dry_run, "action": "already_on", "branch": target, "from_branch": current_branch, "commit_id": get_head_commit_id(root, target) or "", })) else: print(f"Already on '{sanitize_display(target)}'") return target_commit_id = get_head_commit_id(root, target) or "" current_commit_id = get_head_commit_id(root, current_branch) or "" if dry_run: if fmt == "json": print(json.dumps({ "dry_run": True, "action": "switched", "branch": target, "from_branch": current_branch, "commit_id": target_commit_id, })) else: print(f"Would switch to branch '{sanitize_display(target)}'") return target_snapshot_id = get_head_snapshot_id(root, repo_id, target) if target_snapshot_id: _checkout_snapshot( root, target_snapshot_id, current_snapshot_id, target_branch=target, ) write_head_branch(root, target) append_reflog( root, target, old_id=current_commit_id or None, new_id=target_commit_id or ("0" * 64), author="user", operation=( f"checkout: moving from {sanitize_display(current_branch)} " f"to {sanitize_display(target)}" ), ) if fmt == "json": print(json.dumps({ "dry_run": False, "action": "switched", "branch": target, "from_branch": current_branch, "commit_id": target_commit_id, })) else: print(f"Switched to branch '{sanitize_display(target)}'") return commit = resolve_commit_ref(root, repo_id, current_branch, target) if commit is None: print( f"❌ '{sanitize_display(target)}' is not a branch or commit ID.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) current_commit_id = get_head_commit_id(root, current_branch) or "" if dry_run: if fmt == "json": print(json.dumps({ "dry_run": True, "action": "detached", "branch": None, "from_branch": current_branch, "commit_id": commit.commit_id, })) else: print( f"Would detach HEAD at {commit.commit_id[:8]} " f"{sanitize_display(commit.message or '')}" ) return _checkout_snapshot(root, commit.snapshot_id, current_snapshot_id) # Detached HEAD — no branch name; marker cleared by _checkout_snapshot. write_head_commit(root, commit.commit_id) append_reflog( root, current_branch, old_id=current_commit_id or None, new_id=commit.commit_id, author="user", operation=f"checkout: detaching HEAD at {commit.commit_id[:12]}", ) if fmt == "json": print(json.dumps({ "dry_run": False, "action": "detached", "branch": None, "from_branch": current_branch, "commit_id": commit.commit_id, })) else: print(f"HEAD is now at {commit.commit_id[:8]} {sanitize_display(commit.message or '')}") finally: # ── Autostash pop ───────────────────────────────────────────────────── # Mirrors git's "Applied autostash." message. Runs on every exit path # — success, validation error, SystemExit — so the user's work is # never stranded in the stash. if autostash_entry is not None: _apply_autostash(root, fmt)