"""``muse revert`` — create a new commit that undoes a prior commit. Revert is the safe undo: instead of rewriting history (as ``muse reset`` does), revert appends a new commit whose snapshot matches the target commit's parent snapshot. The branch history stays linear and auditable. Usage:: muse revert — revert and commit immediately muse revert --no-commit — apply the revert to the working tree only muse revert --dry-run — simulate without writing anything JSON output (``--format json`` or ``--json``):: { "status": "reverted | applied", "commit_id": " | null", "branch": "", "ref": "", "reverted_commit_id": "", "snapshot_id": "", "message": "", "no_commit": false, "dry_run": false } The schema is identical for all paths (normal, ``--no-commit``, ``--dry-run``); only ``status``, ``commit_id``, ``no_commit``, and ``dry_run`` vary. Exit codes:: 0 — success (reverted, applied to workdir, or dry-run) 1 — ref not found, root commit, invalid format 3 — internal error (parent commit or snapshot missing) """ from __future__ import annotations import argparse import datetime import json import logging import sys from muse.core.errors import ExitCode 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 from muse.core.store import ( CommitRecord, get_head_commit_id, read_commit, read_current_branch, read_snapshot, resolve_commit_ref, write_branch_ref, write_commit, ) 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 revert`` subcommand and all its flags.""" parser = subparsers.add_parser( "revert", help="Create a new commit that undoes a prior commit.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("ref", help="Commit to revert (ID, HEAD, HEAD~N).") parser.add_argument( "-m", "--message", default=None, help="Override the revert commit message.", ) parser.add_argument( "--no-commit", "-n", action="store_true", dest="no_commit", help="Apply the revert to the working tree without creating a commit.", ) parser.add_argument( "--force", action="store_true", help="Proceed even if the working tree has uncommitted changes.", ) parser.add_argument( "--dry-run", action="store_true", dest="dry_run", help=( "Simulate the revert without writing anything. " "Validates the target ref and its parent snapshot." ), ) 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: """Create a new commit that undoes 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 with a consistent schema across all paths:: { "status": "reverted | applied", "commit_id": " | null", "branch": "", "ref": "", "reverted_commit_id": "", "snapshot_id": "", "message": "", "no_commit": false, "dry_run": false } Pass ``--dry-run`` to validate the target ref and its parent snapshot without touching the working tree or creating any commit. Pass ``--no-commit`` to apply the revert to the working tree and stage it for a subsequent ``muse commit``. Exit codes:: 0 — success (reverted, applied, or dry-run) 1 — ref not found, root commit, invalid format 3 — internal error (parent commit or snapshot missing) """ ref: str = args.ref message: str | None = args.message no_commit: bool = args.no_commit force: bool = args.force dry_run: bool = getattr(args, "dry_run", False) 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() # Dry-run never touches the working tree. if not dry_run: require_clean_workdir(root, "revert", 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) target = resolve_commit_ref(root, repo_id, branch, ref) if target is None: print( f"❌ Commit '{sanitize_display(ref)}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if target.parent_commit_id is None: print( "❌ Cannot revert the root commit (no parent to restore).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) parent_commit = read_commit(root, target.parent_commit_id) if parent_commit is None: print( f"❌ Parent commit {target.parent_commit_id[:8]} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) target_snapshot = read_snapshot(root, parent_commit.snapshot_id) if target_snapshot is None: print( f"❌ Snapshot {parent_commit.snapshot_id[:8]} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # Sanitize the original commit message before embedding it in the revert # commit message, which is stored permanently on disk. safe_original_message = sanitize_display(target.message.splitlines()[0]) revert_message = message or f'Revert "{safe_original_message}"' # The parent snapshot is already content-addressed in the object store — # reuse its snapshot_id directly rather than re-scanning the workdir. snapshot_id = parent_commit.snapshot_id # Dry-run: validate succeeded — report what would happen and exit. if dry_run: if fmt == "json": print(json.dumps({ "status": "reverted", "commit_id": None, "branch": branch, "ref": ref, "reverted_commit_id": target.commit_id, "snapshot_id": snapshot_id, "message": revert_message, "no_commit": no_commit, "dry_run": True, })) else: print( f"[dry-run] Would revert '{sanitize_display(ref)}' " f"({target.commit_id[:8]}) on '{sanitize_display(branch)}'" ) return head_commit_id = get_head_commit_id(root, branch) if no_commit: apply_manifest(root, target_snapshot.manifest) if fmt == "json": print(json.dumps({ "status": "applied", "commit_id": None, "branch": branch, "ref": ref, "reverted_commit_id": target.commit_id, "snapshot_id": snapshot_id, "message": revert_message, "no_commit": True, "dry_run": False, })) else: print( f"Revert of {target.commit_id[:8]} applied to working tree. " f"Run 'muse commit' to record." ) return # Correct write ordering for atomicity: # 1. Compute commit record (all data validated above — no I/O failures possible here). # 2. write_commit (idempotent — crash here leaves the workdir unchanged). # 3. apply_manifest (workdir is modified only after the commit is durably stored). # 4. write_branch_ref (branch pointer advances last — visible to others only when complete). # 5. append_reflog (non-critical audit trail — never blocks success). committed_at = datetime.datetime.now(datetime.timezone.utc) commit_id = compute_commit_id( parent_ids=[head_commit_id] if head_commit_id else [], snapshot_id=snapshot_id, message=revert_message, committed_at_iso=committed_at.isoformat(), ) write_commit(root, CommitRecord( commit_id=commit_id, repo_id=repo_id, branch=branch, snapshot_id=snapshot_id, message=revert_message, committed_at=committed_at, parent_commit_id=head_commit_id, )) apply_manifest(root, target_snapshot.manifest) write_branch_ref(root, branch, commit_id) append_reflog( root, branch, old_id=head_commit_id, new_id=commit_id, author="user", operation=f"revert: {sanitize_display(ref)} → {commit_id[:12]}", ) if fmt == "json": print(json.dumps({ "status": "reverted", "commit_id": commit_id, "branch": branch, "ref": ref, "reverted_commit_id": target.commit_id, "snapshot_id": snapshot_id, "message": revert_message, "no_commit": False, "dry_run": False, })) else: print( f"[{sanitize_display(branch)} {commit_id[:8]}] " f"{sanitize_display(revert_message)}" )