"""``muse apply-patch `` — apply a Muse patch to the working tree. Applies a ``.mpatch`` file produced by ``muse format-patch``. Before touching any file on disk, the command: 1. Verifies the ``patch_id`` — recomputes the SHA-256 over the canonical JSON and rejects the patch if the IDs don't match. 2. Checks applicability — the ``requires_snapshot`` field must match the current HEAD snapshot ID (bypassed with ``--force``). 3. Restores all added and modified files from the repo's object store. 4. Deletes files listed in ``files_deleted``. The result is a dirty working tree ready for ``muse commit``. Flags ----- ``--dry-run`` Report what would change without writing anything to disk. ``--check`` Report applicability only (does not apply the patch). ``--force`` Bypass the ``requires_snapshot`` applicability check. Use when you know the patch applies cleanly despite a snapshot mismatch. ``--json`` Emit a JSON result object to stdout. Output (JSON, ``--json``):: { "patch_id": "sha256:<64hex>", "files_applied": ["hello.py"], "files_deleted": [], "dry_run": false, "applicable": true, "duration_ms": 0.003412, "exit_code": 0 } Exit codes:: 0 — success 1 — user error: bad patch file, patch_id mismatch, not applicable 2 — not a Muse repository 3 — I/O error reading patch or restoring files Examples:: muse apply-patch /tmp/patches/feat-add-hello.mpatch muse apply-patch patch.mpatch --dry-run muse apply-patch patch.mpatch --check --json muse apply-patch patch.mpatch --force --json """ import argparse import base64 import json as _json import logging import pathlib import sys from muse.core.types import NULL_LONG_ID, long_id, short_id from muse.core.errors import ExitCode from muse.core.object_store import restore_object, write_object from muse.core.patch_record import PatchRecord, compute_patch_id, deserialize_patch from muse.core.repo import require_repo from muse.core.refs import read_current_branch from muse.core.commits import get_head_snapshot_id from muse.core.validation import sanitize_display from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.timing import start_timer from typing import TypedDict logger = logging.getLogger(__name__) class _CheckJson(EnvelopeJson): """JSON output for ``muse apply-patch --check-only --json``.""" patch_id: str applicable: bool requires_snapshot: str class _ApplyPatchJson(EnvelopeJson): """JSON output for ``muse apply-patch --json``.""" patch_id: str files_applied: list[str] files_deleted: list[str] dry_run: bool applicable: bool # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``muse apply-patch`` subcommand.""" parser = subparsers.add_parser( "apply-patch", help="Apply a Muse .mpatch file to the working tree.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "patch_file", metavar="FILE", help="Path to the .mpatch file to apply.", ) parser.add_argument( "--dry-run", "-n", action="store_true", dest="dry_run", help="Report what would change without touching disk.", ) parser.add_argument( "--check", action="store_true", dest="check_only", help="Report applicability only; do not apply the patch.", ) parser.add_argument( "--force", "-f", action="store_true", dest="force", help="Bypass the requires_snapshot applicability check.", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit a JSON result object to stdout.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Run # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Apply a Muse ``.mpatch`` file to the working tree. Verifies patch integrity (``patch_id`` SHA-256), checks applicability against the current HEAD snapshot, restores added/modified files from the object store, and deletes files listed as removed. The result is a dirty working tree ready for ``muse commit``. Agent quickstart ---------------- :: muse apply-patch patch.mpatch --json muse apply-patch patch.mpatch --dry-run --json # preview only muse apply-patch patch.mpatch --check --json # applicability only muse apply-patch patch.mpatch --force --json # skip snapshot check JSON fields ----------- patch_id SHA-256 patch identifier (``sha256:``). files_applied List of file paths restored or created. files_deleted List of file paths removed. dry_run ``true`` when ``--dry-run`` was passed (no writes occurred). applicable ``true`` if the current snapshot matches ``requires_snapshot``. With ``--check``, only ``patch_id``, ``applicable``, and ``requires_snapshot`` are returned. Exit codes ---------- 0 Patch applied successfully (or dry-run / check passed). 1 Patch integrity failure, not applicable, or missing file. 2 Not inside a Muse repository. 3 I/O error restoring or deleting files. """ elapsed = start_timer() patch_path = pathlib.Path(args.patch_file) dry_run: bool = args.dry_run check_only: bool = args.check_only force: bool = args.force json_out: bool = args.json_out # ── Load patch file ─────────────────────────────────────────────────────── if not patch_path.exists(): print( f"❌ Patch file not found: {sanitize_display(str(patch_path))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) try: raw = patch_path.read_bytes() except OSError as exc: print(f"❌ Could not read patch file: {exc}", file=sys.stderr) raise SystemExit(ExitCode.IO_ERROR) try: record: PatchRecord = deserialize_patch(raw) except Exception as exc: print(f"❌ Invalid patch file: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Verify patch_id integrity ───────────────────────────────────────────── expected_id = compute_patch_id(record) if record.patch_id != expected_id: print( f"❌ Patch integrity check failed.\n" f" Stored: {sanitize_display(record.patch_id)}\n" f" Expected: {sanitize_display(expected_id)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() # ── Applicability check ─────────────────────────────────────────────────── applicable = True requires_snapshot = record.applicability.get("requires_snapshot", "") if not force and requires_snapshot: try: branch = read_current_branch(root) current_snapshot = get_head_snapshot_id(root, branch) except Exception: current_snapshot = "" # Initial-commit sentinel (all zeros) is always applicable _sentinel = NULL_LONG_ID if requires_snapshot != _sentinel and current_snapshot != requires_snapshot: applicable = False if check_only: if json_out: print(_json.dumps(_CheckJson( **make_envelope(elapsed), patch_id=record.patch_id, applicable=applicable, requires_snapshot=requires_snapshot, ), separators=(",", ":"))) else: status = "applicable" if applicable else "not applicable" print(f"{'✅' if applicable else '❌'} Patch {status}") return if not applicable: print( f"❌ Patch is not applicable: current snapshot does not match " f"requires_snapshot.\n" f" Use --force to bypass this check.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Restore / delete files ──────────────────────────────────────────────── files_applied: list[str] = [] files_deleted_applied: list[str] = [] to_manifest = record.to_manifest files_deleted = record.files_deleted if not dry_run: # Seed the local object store with embedded blobs from the patch. for oid, b64_content in record.blobs.items(): try: content = base64.b64decode(b64_content) write_object(root, oid, content) except Exception as exc: logger.debug("apply-patch: could not seed object %s: %s", short_id(oid), exc) # Restore added and modified files from object store. for rel_path, object_id in to_manifest.items(): dest = root / rel_path dest.parent.mkdir(parents=True, exist_ok=True) try: restore_object(root, object_id, dest) files_applied.append(rel_path) except Exception as exc: print( f"❌ Could not restore {sanitize_display(rel_path)}: {exc}", file=sys.stderr, ) raise SystemExit(ExitCode.IO_ERROR) # Delete files no longer in the target manifest. for rel_path in files_deleted: dest = root / rel_path if dest.exists(): try: dest.unlink() files_deleted_applied.append(rel_path) except OSError as exc: logger.debug("apply-patch: could not delete %s: %s", rel_path, exc) else: files_applied = sorted(to_manifest.keys()) files_deleted_applied = list(files_deleted) if json_out: print(_json.dumps(_ApplyPatchJson( **make_envelope(elapsed), patch_id=record.patch_id, files_applied=files_applied, files_deleted=files_deleted_applied, dry_run=dry_run, applicable=applicable, ), separators=(",", ":"))) else: if files_applied or files_deleted_applied: prefix = "[dry-run] " if dry_run else "" for p in files_applied: print(f"{prefix}✅ {sanitize_display(p)}") for p in files_deleted_applied: print(f"{prefix}🗑 {sanitize_display(p)}") else: print("✅ Patch applied (no file changes)")