"""``muse cherry-pick`` — apply a specific commit's changes on top of HEAD. Cherry-pick computes the *delta* introduced by a commit (its snapshot vs its parent's snapshot), then applies that delta on top of the current HEAD via a three-way merge. The result is a new commit that replays the same change in a different context. Usage:: muse cherry-pick — apply and commit immediately muse cherry-pick --no-commit — apply to working tree only muse cherry-pick --dry-run — simulate without writing anything JSON output (``--format json`` or ``--json``):: { "status": "picked | applied | conflict | dry_run", "commit_id": " | null", "branch": "", "ref": "", "source_commit_id": "", "snapshot_id": " | null", "message": "", "no_commit": false, "dry_run": false, "conflicts": [] } The schema is identical across all code paths (success, ``--no-commit``, ``--dry-run``, conflict). Exit codes:: 0 — success (picked, applied to workdir, or dry-run) 1 — ref not found, conflict, invalid format, invalid branch name 3 — internal error (unreadable target snapshot) """ from __future__ import annotations import argparse import datetime import json import logging import sys from muse.core.errors import ExitCode from muse.core.merge_engine import write_merge_state 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, compute_snapshot_id, directories_from_manifest from muse.core.store import ( CommitRecord, SnapshotRecord, get_head_commit_id, get_head_snapshot_manifest, read_commit, read_current_branch, read_snapshot, resolve_commit_ref, write_branch_ref, write_commit, write_snapshot, ) 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 from muse.domain import SnapshotManifest from muse.plugins.registry import read_domain, resolve_plugin from muse.core._types import Manifest logger = logging.getLogger(__name__) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse cherry-pick`` subcommand and all its flags.""" parser = subparsers.add_parser( "cherry-pick", help="Apply a specific commit's changes on top of HEAD.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("ref", help="Commit ID (full or prefix) to apply.") parser.add_argument( "-m", "--message", default=None, help="Override the cherry-pick commit message (default: re-uses the source message).", ) parser.add_argument( "-n", "--no-commit", action="store_true", dest="no_commit", help="Apply the changes 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 cherry-pick without writing anything. " "Reports what would be applied and whether conflicts would arise." ), ) 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: """Apply a specific commit's changes on top of HEAD. Computes the delta between ``ref`` and its parent, then applies that delta to the current HEAD snapshot via a three-way merge. 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": "picked | applied | conflict | dry_run", "commit_id": " | null", "branch": "", "ref": "", "source_commit_id": "", "snapshot_id": " | null", "message": "", "no_commit": false, "dry_run": false, "conflicts": [] } Pass ``--dry-run`` to detect conflicts before committing. The working tree and branch pointer are never modified. Pass ``--no-commit`` to stage the result in the working tree for a subsequent ``muse commit``. Pass ``-m`` / ``--message`` to override the cherry-picked commit message. Exit codes:: 0 — success (picked, applied to workdir, or dry-run) 1 — ref not found, conflict, invalid format, invalid branch name 3 — internal error (unreadable target snapshot) """ 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, "cherry-pick", 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) domain = read_domain(root) plugin = resolve_plugin(root) 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) # Validate the target snapshot before touching the working tree. target_snap_rec = read_snapshot(root, target.snapshot_id) if target_snap_rec is None: print( f"❌ Snapshot {target.snapshot_id[:8]} for commit {target.commit_id[:8]} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) target_manifest = target_snap_rec.manifest # Build the base manifest (the parent of the cherry-picked commit). # Fail fast if the parent commit is recorded but its data is missing — # that indicates object-store corruption, not a legitimate root commit. base_manifest: Manifest = {} if target.parent_commit_id: 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 — " "object store may be corrupted.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) parent_snap = read_snapshot(root, parent_commit.snapshot_id) if parent_snap is None: print( f"❌ Parent snapshot {parent_commit.snapshot_id[:8]} not found — " "object store may be corrupted.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_manifest = parent_snap.manifest ours_manifest = get_head_snapshot_manifest(root, repo_id, branch) or {} base_snap = SnapshotManifest(files=base_manifest, domain=domain, directories=directories_from_manifest(base_manifest)) ours_snap = SnapshotManifest(files=ours_manifest, domain=domain, directories=directories_from_manifest(ours_manifest)) target_snap = SnapshotManifest(files=target_manifest, domain=domain, directories=directories_from_manifest(target_manifest)) result = plugin.merge(base_snap, ours_snap, target_snap) # Sanitize the source commit message before embedding it in any stored commit. safe_message = sanitize_display(target.message.splitlines()[0]) commit_message = message or safe_message if not result.is_clean: # Write merge state so `muse conflicts` and `muse checkout --ours/--theirs` # can inspect the conflict without re-running cherry-pick. write_merge_state( root, base_commit=target.parent_commit_id or "", ours_commit=get_head_commit_id(root, branch) or "", theirs_commit=target.commit_id, conflict_paths=result.conflicts, ) if fmt == "json": print(json.dumps({ "status": "conflict", "commit_id": None, "branch": branch, "ref": ref, "source_commit_id": target.commit_id, "snapshot_id": None, "message": commit_message, "no_commit": no_commit, "dry_run": False, "conflicts": sorted(result.conflicts), })) else: print( f"❌ Cherry-pick conflict in {len(result.conflicts)} file(s):", file=sys.stderr, ) for p in sorted(result.conflicts): print(f" CONFLICT (both modified): {sanitize_display(p)}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) merged_manifest = result.merged["files"] # Dry-run: all validation passed — report and exit without writes. if dry_run: snapshot_id = compute_snapshot_id(merged_manifest, directories_from_manifest(merged_manifest)) if fmt == "json": print(json.dumps({ "status": "dry_run", "commit_id": None, "branch": branch, "ref": ref, "source_commit_id": target.commit_id, "snapshot_id": snapshot_id, "message": commit_message, "no_commit": no_commit, "dry_run": True, "conflicts": [], })) else: print( f"[dry-run] Would cherry-pick '{sanitize_display(ref)}' " f"({target.commit_id[:8]}) on '{sanitize_display(branch)}'" ) return if no_commit: apply_manifest(root, merged_manifest) if fmt == "json": print(json.dumps({ "status": "applied", "commit_id": None, "branch": branch, "ref": ref, "source_commit_id": target.commit_id, "snapshot_id": None, "message": commit_message, "no_commit": True, "dry_run": False, "conflicts": [], })) else: print( f"Applied {target.commit_id[:8]} to working tree. " f"Run 'muse commit' to record." ) return # Correct write ordering for atomicity: # 1. Compute snapshot_id and commit_id (pure computation — no I/O). # 2. write_snapshot (idempotent — crash here leaves workdir unchanged). # 3. write_commit (idempotent — crash here leaves workdir unchanged). # 4. apply_manifest (workdir modified only after both objects are durable). # 5. write_branch_ref (branch pointer advances last — visible to others only when complete). # 6. append_reflog (non-critical audit trail — never blocks success). head_commit_id = get_head_commit_id(root, branch) manifest = merged_manifest manifest_dirs = directories_from_manifest(manifest) snapshot_id = compute_snapshot_id(manifest, manifest_dirs) 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=commit_message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=manifest, directories=manifest_dirs)) write_commit(root, CommitRecord( commit_id=commit_id, repo_id=repo_id, branch=branch, snapshot_id=snapshot_id, message=commit_message, committed_at=committed_at, parent_commit_id=head_commit_id, )) apply_manifest(root, 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"cherry-pick: {sanitize_display(ref)} -> {commit_id[:12]}", ) if fmt == "json": print(json.dumps({ "status": "picked", "commit_id": commit_id, "branch": branch, "ref": ref, "source_commit_id": target.commit_id, "snapshot_id": snapshot_id, "message": commit_message, "no_commit": False, "dry_run": False, "conflicts": [], })) else: print( f"[{sanitize_display(branch)} {commit_id[:8]}] " f"{sanitize_display(commit_message)}" )