"""``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 and stage changes for the next commit muse revert --dry-run — simulate without writing anything Ref formats accepted:: HEAD — most recent commit on the current branch HEAD~1 — one commit before HEAD (if supported by resolve_commit_ref) sha256:… — full content-addressed commit ID — unambiguous prefix (min 7 chars) of a commit ID 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, "files_added": ["path/restored.py", ...], "files_modified": ["path/changed.py", ...], "files_removed": ["path/deleted.py", ...], "duration_ms": 12.3, "exit_code": 0 } The schema is **identical** for all paths (normal, ``--no-commit``, ``--dry-run``); only ``status``, ``commit_id``, ``no_commit``, ``dry_run``, and the file-diff lists vary. ``duration_ms`` and ``exit_code`` are always present — including on error responses — so agents can parse the outcome without inspecting the shell exit status. File-diff fields ---------------- ``files_added`` — paths present in the parent snapshot but absent from the current HEAD snapshot; these files will be restored on disk. ``files_modified`` — paths in both snapshots but with different object IDs; these files will be overwritten with the older content. ``files_removed`` — paths in the current HEAD snapshot but absent from the parent snapshot; these files will be deleted from disk. Agent recipe:: result = muse revert HEAD --json if result["exit_code"] != 0: handle_error(result) log(f"reverted {result['reverted_commit_id'][:20]} in {result['duration_ms']:.1f}ms") log(f"+{len(result['files_added'])} -{len(result['files_removed'])} ~{len(result['files_modified'])}") Exit codes:: 0 — success (reverted, applied to workdir, or dry-run) 1 — ref not found, root commit, invalid format 2 — not a Muse repository 3 — internal error (parent commit or snapshot missing) """ import argparse import datetime import json import logging import pathlib import sys from muse.core.types import short_id from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.reflog import append_reflog from muse.core.symlog import _write_symlogs from muse.core.repo import require_repo from muse.cli.config import get_config_value from muse.core.validation import sanitize_provenance from muse.core.ids import hash_commit from muse.core.types import Manifest from muse.core.refs import ( RefConflictError, get_head_commit_id, read_current_branch, write_branch_ref, ) from muse.core.commits import ( CommitRecord, read_commit, resolve_commit_ref, write_commit, ) from muse.core.snapshots import read_snapshot from muse.core.validation import sanitize_display, validate_branch_name from muse.core.workdir import apply_manifest from muse.core.timing import start_timer from muse.cli.guard import require_clean_workdir from muse.plugins.code.stage import StagedFileMap, make_entry, read_stage, write_stage logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # JSON wire format # --------------------------------------------------------------------------- class _RevertJson(EnvelopeJson, total=False): """JSON output schema — identical across all paths including errors.""" status: str # "reverted" | "applied" | "error" commit_id: str | None # null on --no-commit / --dry-run / error branch: str ref: str reverted_commit_id: str | None snapshot_id: str | None message: str no_commit: bool dry_run: bool files_added: list[str] # restored from parent snapshot files_modified: list[str] # content changed relative to current HEAD files_removed: list[str] # deleted to match parent snapshot # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _compute_diff( current_manifest: Manifest, target_manifest: Manifest, ) -> tuple[list[str], list[str], list[str]]: """Return (files_added, files_modified, files_removed) for the revert. The *revert* moves the working tree FROM *current_manifest* TO *target_manifest* (the parent snapshot). - ``files_added`` — in target but not in current (will be restored) - ``files_modified`` — in both but different object IDs (will be overwritten) - ``files_removed`` — in current but not in target (will be deleted) """ current_keys = set(current_manifest) target_keys = set(target_manifest) files_added = sorted(target_keys - current_keys) files_removed = sorted(current_keys - target_keys) files_modified = sorted( p for p in current_keys & target_keys if current_manifest[p] != target_manifest[p] ) return files_added, files_modified, files_removed def _apply_manifest_safe( root: pathlib.Path, prev_manifest: Manifest, target_manifest: Manifest, ) -> None: """Apply *target_manifest* to the working tree, handling the empty-target case. ``apply_manifest`` guards against accidentally applying an empty manifest when there are previously tracked files. In the revert context an empty target is always intentional — the caller validated that we're reverting to a known parent snapshot. We handle the empty case by removing all files that were in *prev_manifest*. """ if not target_manifest: from muse.core.workdir import _prune_empty_directories removed_paths: list[str] = [] for rel_posix in prev_manifest: fp = root / rel_posix if fp.exists(): fp.unlink() removed_paths.append(rel_posix) if removed_paths: _prune_empty_directories(root, removed_paths, frozenset()) else: from muse.core.snapshot import directories_from_manifest apply_manifest( root, prev_manifest, target_manifest, target_directories=directories_from_manifest(target_manifest), ) def _stage_revert( root: pathlib.Path, current_manifest: Manifest, target_manifest: Manifest, files_added: list[str], files_modified: list[str], files_removed: list[str], ) -> None: """Update the stage index to reflect the reverted state. Called by the ``--no-commit`` path so that ``muse status`` shows staged changes and ``muse commit`` can record them without a separate ``muse code add`` step. - Paths in *files_removed* → staged as ``D`` (deleted). - Paths in *files_added* → staged as ``A`` (added, with the target object ID). - Paths in *files_modified* → staged as ``M`` (modified, with the target object ID). """ stage: StagedFileMap = dict(read_stage(root)) for path in files_removed: if path in current_manifest: # Was in HEAD → mark as deleted. stage[path] = make_entry(object_id="", mode="D") else: stage.pop(path, None) for path in files_added: stage[path] = make_entry(object_id=target_manifest[path], mode="A") for path in files_modified: stage[path] = make_entry(object_id=target_manifest[path], mode="M") write_stage(root, stage) # --------------------------------------------------------------------------- # Command registration # --------------------------------------------------------------------------- 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", action="store_true", dest="no_commit", help=( "Apply the revert to the working tree and stage the changes " "without creating a commit. Run 'muse commit' afterwards." ), ) parser.add_argument( "--force", action="store_true", help="Proceed even if the working tree has uncommitted changes.", ) parser.add_argument( "--dry-run", "-n", action="store_true", dest="dry_run", help=( "Simulate the revert without writing anything. " "Validates the target ref and its parent snapshot." ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON instead of human text.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Main handler # --------------------------------------------------------------------------- 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. ``files_added`` / ``files_modified`` / ``files_removed`` are populated on all three paths (normal, ``--no-commit``, ``--dry-run``). Agent quickstart:: muse revert HEAD --json muse revert HEAD --no-commit --json muse revert HEAD --dry-run --json muse revert sha256:abc123 --json JSON fields:: status str "reverted" | "applied" | "error" commit_id str|null New revert commit ID; null for --no-commit/--dry-run branch str Current branch name ref str Ref as passed by the caller reverted_commit_id str|null Commit ID that was reverted snapshot_id str|null Snapshot ID restored by the revert message str Revert commit message no_commit bool True when --no-commit was given dry_run bool True when no writes were made files_added list[str] Paths restored from the parent snapshot files_modified list[str] Paths overwritten with older content files_removed list[str] Paths deleted to match the parent snapshot Exit codes:: 0 Success (reverted, applied, or dry-run). 1 Ref not found, root commit, or invalid format. 2 Not a Muse repository. 3 Internal error (parent commit or snapshot missing). """ elapsed = start_timer() 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) json_out: bool = args.json_out def _emit_error(exit_code: int, error_key: str, msg: str, **extra: str) -> None: """Emit a JSON error envelope on stdout when json_out is True.""" if json_out: payload = { **make_envelope(elapsed, exit_code=exit_code), "status": "error", "commit_id": None, "branch": "", "ref": ref, "reverted_commit_id": None, "snapshot_id": None, "message": msg, "no_commit": no_commit, "dry_run": dry_run, "files_added": [], "files_modified": [], "files_removed": [], "error": error_key, **extra, } print(json.dumps(payload)) root = require_repo() # Dry-run never touches the working tree. if not dry_run: require_clean_workdir(root, "revert", force=force, json_out=json_out) branch = read_current_branch(root) try: validate_branch_name(branch) except ValueError as exc: _emit_error(ExitCode.INTERNAL_ERROR, "invalid_branch", str(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, branch, ref) if target is None: _emit_error( ExitCode.USER_ERROR, "commit_not_found", f"commit '{ref}' not found", branch=branch, ) print( f"❌ Commit '{sanitize_display(ref)}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if target.parent_commit_id is None: _emit_error( ExitCode.USER_ERROR, "root_commit", "cannot revert the root commit (no parent to restore)", branch=branch, reverted_commit_id=target.commit_id, ) 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: _emit_error( ExitCode.INTERNAL_ERROR, "parent_not_found", f"parent commit {target.parent_commit_id} not found", branch=branch, reverted_commit_id=target.commit_id, ) print( f"❌ Parent commit {target.parent_commit_id} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) target_snapshot = read_snapshot(root, parent_commit.snapshot_id) if target_snapshot is None: _emit_error( ExitCode.INTERNAL_ERROR, "snapshot_missing", f"snapshot {parent_commit.snapshot_id} not found", branch=branch, reverted_commit_id=target.commit_id, ) print( f"❌ Snapshot {parent_commit.snapshot_id} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # Read the current HEAD snapshot to compute the file-level diff. head_commit_id = get_head_commit_id(root, branch) current_manifest: Manifest = {} if head_commit_id: head_commit = read_commit(root, head_commit_id) if head_commit: head_snap = read_snapshot(root, head_commit.snapshot_id) if head_snap: current_manifest = dict(head_snap.manifest) # Compute the file-level diff: current → parent (target of revert). files_added, files_modified, files_removed = _compute_diff( current_manifest, dict(target_snapshot.manifest) ) # 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 json_out: print(json.dumps(_RevertJson( **make_envelope(elapsed), 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, files_added=files_added, files_modified=files_modified, files_removed=files_removed, ))) else: print( f"[dry-run] Would revert '{sanitize_display(ref)}' " f"({target.commit_id}) on '{sanitize_display(branch)}'" ) return if no_commit: _apply_manifest_safe(root, current_manifest, dict(target_snapshot.manifest)) _stage_revert( root, current_manifest, dict(target_snapshot.manifest), files_added, files_modified, files_removed, ) if json_out: print(json.dumps(_RevertJson( **make_envelope(elapsed), 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, files_added=files_added, files_modified=files_modified, files_removed=files_removed, ))) else: print( f"Revert of {target.commit_id} 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) revert_author = sanitize_provenance(get_config_value("user.handle", root) or "") commit_id = hash_commit( parent_ids=[head_commit_id] if head_commit_id else [], snapshot_id=snapshot_id, message=revert_message, committed_at_iso=committed_at.isoformat(), author=revert_author, ) write_commit(root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snapshot_id, message=revert_message, committed_at=committed_at, parent_commit_id=head_commit_id, author=revert_author, )) _apply_manifest_safe(root, current_manifest, dict(target_snapshot.manifest)) try: write_branch_ref(root, branch, commit_id, expected_id=head_commit_id) except RefConflictError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) append_reflog( root, branch, old_id=head_commit_id, new_id=commit_id, author="user", operation=f"revert: {sanitize_display(ref)} → {commit_id}", ) try: _write_symlogs( root, parent_commit_id=head_commit_id, new_snapshot_id=snapshot_id, new_commit_id=commit_id, author=revert_author or "unknown", commit_message=revert_message, ) except Exception as _symlog_exc: logger.warning( "⚠️ symlog write failed for revert commit %s — symbol journal may be incomplete: %s", commit_id, _symlog_exc, ) if json_out: print(json.dumps(_RevertJson( **make_envelope(elapsed), 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, files_added=files_added, files_modified=files_modified, files_removed=files_removed, ))) else: print( f"[{sanitize_display(branch)} {short_id(commit_id)}] " f"{sanitize_display(revert_message)}" )