"""``muse reset`` — move HEAD to a prior commit. Modes:: --soft Move the branch pointer only; working tree is untouched. --hard Move the branch pointer AND restore the working tree from the target snapshot. Requires a clean working tree unless ``--force`` is also passed. Usage:: muse reset — soft reset (branch pointer only) muse reset --hard — hard reset (branch + working tree) muse reset --dry-run — simulate without writing anything JSON output (``--format json`` or ``--json``):: { "branch": "", "ref": "", "old_commit_id": " | null", "new_commit_id": "", "snapshot_id": "", "mode": "soft | hard", "dry_run": false } Exit codes:: 0 — success (or dry-run with no problems found) 1 — ref not found, invalid branch name, invalid format 3 — internal error (snapshot missing from object store) """ from __future__ import annotations import argparse import json import logging import sys from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_head_commit_id, read_current_branch, read_snapshot, resolve_commit_ref, write_branch_ref, ) from muse.core.reflog import append_reflog from muse.core.validation import sanitize_display, validate_branch_name from muse.core.workdir import apply_manifest from muse.cli.guard import require_clean_workdir logger = logging.getLogger(__name__) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse reset`` subcommand and all its flags.""" parser = subparsers.add_parser( "reset", help="Move HEAD to a prior commit.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("ref", help="Commit ID or ref to reset to (e.g. HEAD~1, a3f2c9).") parser.add_argument( "--hard", action="store_true", help="Reset branch pointer AND restore the working tree from the target snapshot.", ) parser.add_argument( "--soft", action="store_true", help="Reset branch pointer only, leaving the working tree unchanged (default).", ) parser.add_argument( "--force", action="store_true", help="With --hard: proceed even if the working tree has uncommitted changes.", ) parser.add_argument( "--dry-run", action="store_true", dest="dry_run", help=( "Simulate the reset without writing anything. " "Reports what would change and validates the target ref." ), ) parser.add_argument( "--format", "-f", 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.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Move HEAD to a prior commit. All error messages are written to **stderr**; stdout is reserved for structured output only. Agents should pass ``--format json`` for a stable, machine-readable result:: { "branch": "", "ref": "", "old_commit_id": " | null", "new_commit_id": "", "snapshot_id": "", "mode": "soft | hard", "dry_run": false } Pass ``--dry-run`` to validate the target ref and preview the reset without writing anything to disk. Exit 0 if the reset would succeed. Exit codes:: 0 — success (or successful dry-run) 1 — ref not found, invalid branch name, invalid format 3 — internal error (target snapshot missing from object store) """ ref: str = args.ref hard: bool = args.hard dry_run: bool = getattr(args, "dry_run", False) force: bool = args.force fmt: str = args.fmt 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) root = require_repo() # Only --hard modifies the working tree; --soft just moves the branch pointer. # Dry-run never touches the working tree. if hard and not dry_run: require_clean_workdir(root, "reset --hard", force=force) repo_id = read_repo_id(root) branch = read_current_branch(root) try: validate_branch_name(branch) except ValueError as exc: print( f"❌ Current branch name is invalid: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ '{sanitize_display(ref)}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # For --hard, validate the snapshot exists BEFORE writing anything to disk. # This prevents the critical ordering bug where write_branch_ref() succeeds # but apply_manifest() fails, leaving the branch ref pointing at the new # commit while the working tree is only partially restored. snapshot_id = commit.snapshot_id if hard: snapshot = read_snapshot(root, snapshot_id) if snapshot is None: print( f"❌ Snapshot {snapshot_id[:8]} not found in object store " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) else: snapshot = None mode = "hard" if hard else "soft" old_commit_id = get_head_commit_id(root, branch) # Dry-run: report what would happen without writing anything. if dry_run: if fmt == "json": print(json.dumps({ "branch": branch, "ref": ref, "old_commit_id": old_commit_id, "new_commit_id": commit.commit_id, "snapshot_id": snapshot_id, "mode": mode, "dry_run": True, })) else: label = f"[dry-run] Would reset '{sanitize_display(branch)}'" msg_snip = sanitize_display(commit.message.splitlines()[0][:60]) print(f"{label} ({mode}) to {commit.commit_id[:8]} {msg_snip}") return # Write the branch ref first for --soft (no snapshot dependency). # For --hard: snapshot was already validated above — write ref then restore. write_branch_ref(root, branch, commit.commit_id) append_reflog( root, branch, old_id=old_commit_id, new_id=commit.commit_id, author="user", operation=f"reset ({mode}): moving to {commit.commit_id[:12]}", ) if hard and snapshot is not None: apply_manifest(root, snapshot.manifest) if fmt == "json": print(json.dumps({ "branch": branch, "ref": ref, "old_commit_id": old_commit_id, "new_commit_id": commit.commit_id, "snapshot_id": snapshot_id, "mode": mode, "dry_run": False, })) elif hard: print( f"HEAD is now at {commit.commit_id[:8]} " f"{sanitize_display(commit.message.splitlines()[0])}" ) else: print(f"Moved {sanitize_display(branch)} to {commit.commit_id[:8]}")