"""``muse switch`` — focused branch switcher. A cleaner, more intentional alternative to ``muse checkout`` for branch operations. It delegates to checkout's internal machinery so all snapshot application, reflog, and dirty-tree logic stays in one place. Differences from ``muse checkout`` ------------------------------------ - **Branch-only**: no file restoring, no conflict resolution. - **``-c``** creates a new branch (like checkout ``-b``). - **``-C``** force-creates: resets the ref if the branch already exists. - **``switch -``**: switches to the previously-checked-out branch, stored in ``.muse/PREV_BRANCH``. - **``--discard-changes``**: explicit name for ``--force``; discards dirty working-tree changes before switching. - **``--detach``**: detach HEAD at a commit (like ``checkout `` without a branch argument). - **``--intent``**: annotate a newly-created branch with a human/agent description of its purpose (only valid with ``-c``). - **``--resumable``**: mark a newly-created branch as a resumable agent checkpoint (only valid with ``-c``). JSON schema (``--json``):: { "action": "switched" | "created" | "detached" | "already_on" | "reset", "branch": " | null", "from_branch": "", "commit_id": "", "dry_run": true | false, "duration_ms": 1.234, "exit_code": 0 } JSON error schema (``--json``, always to stdout so agents can parse failures):: { "error": "", "message": "", "duration_ms": 0.3, "exit_code": 1 } Exit codes:: 0 — success (or would-succeed in dry-run) 1 — user error: branch not found, dirty tree, bad name, no previous branch 2 — not a Muse repository 3 — internal error Examples:: muse switch feat # switch to existing branch muse switch -c task/new-thing # create and switch muse switch -c task/new --intent "implement feature X" --resumable muse switch -C feat # force-reset feat to HEAD then switch muse switch - # back to previous branch muse switch --detach abc123 # detach HEAD at commit muse switch feat --dry-run --json # preview in JSON muse switch feat --discard-changes # force switch over dirty tree muse switch feat --autoshelf # shelf, switch, pop muse switch feat --merge # carry changes via 3-way merge """ import argparse import contextlib import io import json as _json import logging import pathlib import sys from muse.core.types import NULL_COMMIT_ID from muse.core.paths import prev_branch_path as _prev_branch_path, ref_path as _ref_path from muse.core.errors import ExitCode from muse.core.reflog import append_reflog from muse.core.repo import require_repo from muse.core.refs import read_ref from muse.core.io import write_text_atomic from muse.core.refs import ( get_head_commit_id, read_current_branch, read_head, write_branch_ref, write_head_branch, ) from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.validation import sanitize_display, validate_branch_name from muse.core.timing import start_timer logger = logging.getLogger(__name__) class _SwitchJson(EnvelopeJson): action: str branch: str | None from_branch: str commit_id: str dry_run: bool # Path of the previous-branch file, relative to the repo's .muse directory. _PREV_BRANCH_FILE = "PREV_BRANCH" # --------------------------------------------------------------------------- # PREV_BRANCH helpers # --------------------------------------------------------------------------- def _read_prev_branch(root: pathlib.Path) -> str | None: """Return the previously-switched-to branch, or ``None`` if not recorded.""" path = _prev_branch_path(root) try: val = path.read_text(encoding="utf-8").strip() return val if val else None except (FileNotFoundError, PermissionError, OSError): return None def _write_prev_branch(root: pathlib.Path, branch: str) -> None: """Atomically record *branch* as the previous branch.""" write_text_atomic(_prev_branch_path(root), f"{branch}\n") # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``muse switch`` subcommand.""" parser = subparsers.add_parser( "switch", help="Switch branches (focused alternative to checkout).", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "target", nargs="?", metavar="BRANCH | COMMIT | -", help=( "Branch to switch to, commit ID (with --detach), or '-' to " "return to the previously checked-out branch." ), ) parser.add_argument( "-c", "--create", action="store_true", dest="create", help="Create a new branch at HEAD and switch to it.", ) parser.add_argument( "-C", "--force-create", action="store_true", dest="force_create", help=( "Force-create: create the branch if it doesn't exist, or reset " "its tip to the current HEAD if it does. Then switch to it." ), ) parser.add_argument( "--detach", action="store_true", help="Detach HEAD at BRANCH or COMMIT rather than switching branches.", ) parser.add_argument( "--discard-changes", "-f", action="store_true", dest="discard_changes", help="Discard uncommitted working-tree changes before switching.", ) parser.add_argument( "--merge", "-m", action="store_true", dest="merge", help=( "Carry uncommitted changes into the target branch via a " "three-way merge (the Cohen Transform). Mutually exclusive " "with --discard-changes and --autoshelf." ), ) parser.add_argument( "--autoshelf", action="store_true", dest="autoshelf", help=( "Shelf uncommitted changes, switch, then pop them back. " "Mutually exclusive with --discard-changes and --merge." ), ) parser.add_argument( "--intent", default=None, metavar="TEXT", help=( "Annotate the new branch with an intent description. " "Only valid with -c/--create." ), ) parser.add_argument( "--resumable", action="store_true", default=False, help=( "Mark the new branch as a resumable agent checkpoint " "(discoverable via ``muse branch --resumable``). " "Only valid with -c/--create." ), ) parser.add_argument( "-n", "--dry-run", action="store_true", dest="dry_run", help="Preview what would happen without making any changes.", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON on stdout.", ) parser.set_defaults( func=run, create=False, force_create=False, detach=False, discard_changes=False, merge=False, autoshelf=False, intent=None, resumable=False, dry_run=False, json_out=False, ) # --------------------------------------------------------------------------- # Run # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Switch branches, delegating to checkout's internal machinery. All snapshot application, reflog, autoshelf, and merge logic lives in checkout so there is exactly one implementation. switch translates its flags into a compatible Namespace then calls checkout.run(). When ``--json`` is set, every error is emitted as a JSON object to stdout so agents can parse failures without inspecting the exit code first. Agent quickstart:: muse switch feat --json muse switch -c task/new-thing --json muse switch -c task/new --intent "implement X" --resumable --json muse switch --dry-run feat --json JSON fields:: action ``"switched"``, ``"created"``, ``"detached"``, ``"already_on"``, or ``"reset"``. branch Target branch name; ``null`` on detached HEAD. from_branch Branch name at the time the command was invoked. commit_id SHA-256 commit ID at the new HEAD. dry_run ``true`` when ``--dry-run`` was passed. muse_version Muse release that produced this output. schema Envelope schema version (int). exit_code ``0`` on success, ``1`` on user error. duration_ms Wall-clock milliseconds for the command. timestamp ISO-8601 UTC timestamp of command completion. warnings List of non-fatal advisory messages. Exit codes:: 0 Success (or would-succeed in dry-run). 1 User error (branch not found, dirty tree, bad name, no prev branch). 2 Not a Muse repository. 3 Internal error. """ # Lazy import to avoid a circular dependency at module load time. from muse.cli.commands import checkout as checkout_mod elapsed = start_timer() target: str | None = args.target create: bool = args.create force_create: bool = args.force_create detach: bool = args.detach discard_changes: bool = args.discard_changes merge: bool = getattr(args, "merge", False) autoshelf: bool = getattr(args, "autoshelf", False) intent: str | None = getattr(args, "intent", None) resumable: bool = getattr(args, "resumable", False) dry_run: bool = args.dry_run json_out: bool = args.json_out def _emit_error( msg: str, code: int, error_key: str = "switch_failed", **extra: str, ) -> None: """Emit a structured error to stdout (JSON) or stderr (text) then exit.""" if json_out: payload = { **make_envelope(elapsed, exit_code=code), "error": error_key, "message": msg, } payload.update(extra) print(_json.dumps(payload)) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(code) # ── Mutual-exclusion guards ─────────────────────────────────────────────── if create and force_create: _emit_error( "-c and -C are mutually exclusive.", ExitCode.USER_ERROR, "mutual_exclusion", ) if create and detach: _emit_error( "--create and --detach are mutually exclusive.", ExitCode.USER_ERROR, "mutual_exclusion", ) if discard_changes and merge: _emit_error( "--discard-changes and --merge are mutually exclusive.", ExitCode.USER_ERROR, "mutual_exclusion", ) if discard_changes and autoshelf: _emit_error( "--discard-changes and --autoshelf are mutually exclusive.", ExitCode.USER_ERROR, "mutual_exclusion", ) if merge and autoshelf: _emit_error( "--merge and --autoshelf are mutually exclusive.", ExitCode.USER_ERROR, "mutual_exclusion", ) # ── Require a target ───────────────────────────────────────────────────── if target is None: _emit_error( "Specify a branch, '-' for previous, or a commit with --detach.", ExitCode.USER_ERROR, "missing_target", ) root = require_repo() # ── switch - (previous branch) ─────────────────────────────────────────── if target == "-": prev = _read_prev_branch(root) if not prev: _emit_error( "No previous branch recorded. " "Switch to a branch first, then use 'muse switch -'.", ExitCode.USER_ERROR, "no_prev_branch", ) target = prev # ── -C / force-create ──────────────────────────────────────────────────── if force_create: _run_force_create( root, target, dry_run=dry_run, fmt="json" if json_out else "text", elapsed_fn=elapsed, ) return # ── Record current branch for "switch -" before delegating ─────────────── current_branch = read_current_branch(root) # ── Same-commit fast-path: no files change, dirty tree is safe ─────────── # If the target branch already points to the same commit as HEAD, switching # is a pure HEAD ref update — apply_manifest is a no-op and nothing in the # working tree will be touched. Skip the dirty guard and delegate entirely, # matching git switch behaviour. # Only applies when not creating a new branch (create=True means the branch # doesn't exist yet so we can't compare tips). if not create and not dry_run: try: _target_commit = get_head_commit_id(root, target) _current_commit = get_head_commit_id(root, current_branch) except (ValueError, OSError): _target_commit = None _current_commit = None if _target_commit and _target_commit == _current_commit and target != current_branch: # Pure ref switch: just update HEAD, write reflog, done. write_head_branch(root, target) append_reflog( root, target, old_id=_current_commit, new_id=_current_commit, author="user", operation=f"switch: from {current_branch} (same commit)", ) _write_prev_branch(root, current_branch) if json_out: print(_json.dumps(_SwitchJson( **make_envelope(elapsed), action="switched", branch=target, from_branch=current_branch, commit_id=_current_commit, dry_run=False, ))) else: print(f"Switched to branch '{sanitize_display(target)}'") return # ── Build a compatible Namespace for checkout.run() ────────────────────── checkout_args = argparse.Namespace( target=target, create=create, # -b force=discard_changes, # --force dry_run=dry_run, merge=merge, autoshelf=autoshelf, # Switch owns the JSON payload; always run checkout in text mode so # only one structured payload is emitted to stdout. json_out=False, resolve_ours=False, resolve_theirs=False, resolve_all=False, intent=intent, resumable=resumable, ) if json_out: # Suppress checkout's text output; switch owns the JSON payload. _discard = io.StringIO() try: with contextlib.redirect_stdout(_discard), \ contextlib.redirect_stderr(_discard): checkout_mod.run(checkout_args) except SystemExit as exc: raw = exc.code code = int(raw.value) if hasattr(raw, "value") else (int(raw) if raw is not None else 1) print(_json.dumps({ **make_envelope(elapsed, exit_code=code), "error": "switch_failed", "message": _discard.getvalue().strip() or "switch failed", })) raise SystemExit(code) except Exception as exc: # Checkout raised a non-SystemExit exception (e.g. ValueError from # validate_branch_name on an internal code path). Wrap it as a # structured JSON error so agents always get parseable output. print(_json.dumps({ **make_envelope(elapsed, exit_code=ExitCode.USER_ERROR), "error": "switch_failed", "message": sanitize_display(str(exc)), })) raise SystemExit(ExitCode.USER_ERROR) else: checkout_mod.run(checkout_args) # ── Write PREV_BRANCH (only on a real, successful branch switch) ───────── if not dry_run: try: new_branch = read_current_branch(root) if new_branch != current_branch: _write_prev_branch(root, current_branch) except ValueError: _write_prev_branch(root, current_branch) # ── Emit switch's own JSON payload ─────────────────────────────────────── if json_out: if create: action = "created" branch_out = target commit_id = get_head_commit_id(root, branch_out) or "" elif dry_run: action = "switched" branch_out = target commit_id = get_head_commit_id(root, branch_out) or "" else: head = read_head(root) if head["kind"] == "commit": # Detached HEAD after --detach. action = "detached" branch_out = None commit_id = head["commit_id"] else: new_branch = head["branch"] branch_out = new_branch commit_id = get_head_commit_id(root, new_branch) or "" if new_branch == current_branch: action = "already_on" else: action = "switched" print(_json.dumps(_SwitchJson( **make_envelope(elapsed), action=action, branch=sanitize_display(branch_out) if branch_out else None, from_branch=sanitize_display(current_branch), commit_id=commit_id, dry_run=dry_run, ))) # --------------------------------------------------------------------------- # Internal: force-create (-C) # --------------------------------------------------------------------------- def _run_force_create( root: pathlib.Path, target: str, *, dry_run: bool, fmt: str, elapsed_fn: "() -> float", ) -> None: """Implement ``muse switch -C ``. Creates *target* if it doesn't exist, or resets its tip to the current HEAD if it does. Then switches to it. Args: root: Absolute repo root. target: Branch name to create or reset. dry_run: If True, validate and report without writing anything. fmt: Output format: ``"text"`` or ``"json"``. elapsed_fn: Callable returning milliseconds elapsed since the command started. Used to populate ``duration_ms`` in JSON output. """ try: validate_branch_name(target) except ValueError as exc: if fmt == "json": print(_json.dumps({ **make_envelope(elapsed_fn, exit_code=ExitCode.USER_ERROR), "error": "invalid_branch_name", "message": sanitize_display(str(exc)), })) else: print( f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) current_branch = read_current_branch(root) current_commit = get_head_commit_id(root, current_branch) or "" ref_file = _ref_path(root, target) branch_existed = ref_file.exists() if dry_run: action = "reset" if branch_existed else "created" if fmt == "json": print(_json.dumps(_SwitchJson( **make_envelope(elapsed_fn), action=action, branch=sanitize_display(target), from_branch=sanitize_display(current_branch), commit_id=current_commit, dry_run=True, ))) else: verb = "Reset" if branch_existed else "Create" print( f"[dry-run] Would {verb.lower()} branch " f"'{sanitize_display(target)}' at " f"{current_commit} and switch to it." ) return # Write the ref (create or overwrite). if current_commit: write_branch_ref(root, target, current_commit) else: write_text_atomic(ref_file, "") # Switch HEAD. write_head_branch(root, target) append_reflog( root, target, old_id=None, new_id=current_commit or NULL_COMMIT_ID, author="user", operation=( f"switch -C: {'reset' if branch_existed else 'created'} " f"from {sanitize_display(current_branch)}" ), ) # Record previous branch for "switch -". _write_prev_branch(root, current_branch) action = "reset" if branch_existed else "created" if fmt == "json": print(_json.dumps(_SwitchJson( **make_envelope(elapsed_fn), action=action, branch=sanitize_display(target), from_branch=sanitize_display(current_branch), commit_id=current_commit, dry_run=False, ))) else: verb = "Reset and switched" if branch_existed else "Switched to a new branch" print(f"{verb} '{sanitize_display(target)}'")