"""muse code checkout-symbol — restore a historical version of a specific symbol. Extracts a single named symbol from a historical committed snapshot and writes it back into the current working-tree file, replacing the current version of that symbol. This is a **surgical** operation: only the target symbol's lines change. All surrounding code — other symbols, comments, imports, blank lines outside the symbol boundary — is left untouched. Why this matters ---------------- Git's ``checkout`` restores entire files. If you need to roll back a single function while keeping everything else current, you need to manually cherry- pick lines. ``muse code checkout-symbol`` does this atomically against Muse's content-addressed symbol index. No-op detection: if the symbol body at the target commit is identical to the current working-tree version (same body hash), the file is not written and ``"changed": false`` is reported. Security note: the file path component of ADDRESS is validated via ``contain_path()`` before any disk access. Paths that escape the repo root (e.g. ``../../etc/passwd::foo``) are rejected with exit 1. Usage:: muse code checkout-symbol "src/billing.py::compute_invoice_total" --commit HEAD~3 muse code checkout-symbol "src/auth.py::validate_token" --commit abc12345 --dry-run muse code checkout-symbol "src/billing.py::compute_invoice_total" --commit HEAD~3 --json Output (without --dry-run):: Restoring: src/billing.py::compute_invoice_total from commit: abc12345 (2026-02-15) lines 42–67 → replaced with 31 historical line(s) ✅ Written to src/billing.py Output (no-op — symbol already matches):: ✅ src/billing.py::compute_invoice_total already matches commit abc12345 — nothing to do. Output (with --dry-run):: Dry run — no files will be written. Restoring: src/billing.py::compute_invoice_total from commit: abc12345 (2026-02-15) --- current +++ historical @@ -42,26 +42,20 @@ def compute_invoice_total(...): - ...current body... + ...historical body... JSON output (``--json``):: { "schema_version": "0.1.5", "address": "src/billing.py::compute_invoice_total", "file": "src/billing.py", "branch": "main", "restored_from": "abc12345", "dry_run": false, "changed": true, "appended": false, "current_start": 42, "current_end": 67, "historical_line_count": 31, "diff_lines": [] } Flags: ``--commit, -c REF`` Required. Commit to restore from. ``--dry-run`` Print the diff without writing anything. ``--json`` Emit result as JSON for agent consumption. In dry-run mode the JSON includes a ``diff_lines`` list so agents can inspect what would change. """ from __future__ import annotations import argparse import difflib import json import logging import pathlib from muse._version import __version__ from muse.core.errors import ExitCode from muse.core.object_store import read_object from muse.core.repo import read_repo_id, require_repo from muse.core.store import get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref from muse.core.validation import contain_path, sanitize_display from muse.plugins.code.ast_parser import SymbolRecord, SymbolTree, parse_symbols logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _extract_lines(source: bytes, lineno: int, end_lineno: int) -> list[str]: """Extract lines *lineno*..*end_lineno* (1-indexed, inclusive). Returns an empty list and logs a warning if the range is out of bounds rather than silently truncating. """ all_lines = source.decode("utf-8", errors="replace").splitlines(keepends=True) total = len(all_lines) if lineno < 1 or end_lineno < lineno or end_lineno > total: logger.warning( "Line range %d–%d out of bounds for %d-line source — returning empty", lineno, end_lineno, total, ) return [] return all_lines[lineno - 1:end_lineno] def _find_symbol_in_source( source: bytes, file_rel: str, address: str, ) -> SymbolRecord | None: """Parse *source* as *file_rel* and return the record for *address*, or None.""" # file_rel must be the repo-relative path so that parse_symbols builds # address keys that match the caller's address format (e.g. "src/a.py::fn"). tree: SymbolTree = parse_symbols(source, file_rel) return tree.get(address) # --------------------------------------------------------------------------- # CLI registration and entry point # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the checkout-symbol subcommand.""" parser = subparsers.add_parser( "checkout-symbol", help="Restore a historical version of a specific symbol into the working tree.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "address", metavar="ADDRESS", help='Symbol address, e.g. "src/billing.py::compute_invoice_total".', ) parser.add_argument( "--commit", "-c", required=True, metavar="REF", dest="ref", help="Commit to restore the symbol from (required).", ) parser.add_argument( "--dry-run", action="store_true", dest="dry_run", help="Print the diff without writing anything.", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit result as JSON for agent consumption.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Restore a historical version of a specific symbol into the working tree. Extracts the symbol body from the given historical commit and splices it into the current working-tree file at the symbol's current location. Only the target symbol's lines change; everything else is left untouched. If the symbol body already matches the historical version, no file is written and ``changed=false`` is reported — a safe no-op. If the symbol does not exist at ``--commit``, the command exits with an error. If the symbol does not exist in the current working tree, the historical version is appended to the end of the file. The file path component of ADDRESS is validated against the repo root — path-traversal addresses (e.g. ``../../etc/passwd::foo``) are rejected. """ address: str = args.address ref: str = args.ref dry_run: bool = args.dry_run as_json: bool = args.as_json root = require_repo() try: repo_id = read_repo_id(root) except (FileNotFoundError, json.JSONDecodeError, KeyError) as exc: logger.error("Cannot read repo identity: %s", exc) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc branch = read_current_branch(root) if "::" not in address: logger.error("ADDRESS must be a symbol address like 'src/billing.py::func'.") raise SystemExit(ExitCode.USER_ERROR) file_rel, _sym_name = address.split("::", 1) # Validate the file path stays inside the repo root. try: contain_path(root, file_rel) except ValueError as exc: logger.error("%s", exc) raise SystemExit(ExitCode.USER_ERROR) from exc commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: logger.error("Commit %r not found.", ref) raise SystemExit(ExitCode.USER_ERROR) # ------------------------------------------------------------------ # Load the historical blob and locate the symbol inside it. # ------------------------------------------------------------------ manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} obj_id = manifest.get(file_rel) if obj_id is None: logger.error("'%s' is not in snapshot %s.", file_rel, commit.commit_id[:8]) raise SystemExit(ExitCode.USER_ERROR) historical_raw = read_object(root, obj_id) if historical_raw is None: logger.error("Blob %s missing from object store.", obj_id[:8]) raise SystemExit(ExitCode.USER_ERROR) hist_rec = _find_symbol_in_source(historical_raw, file_rel, address) if hist_rec is None: logger.error("Symbol '%s' not found in commit %s.", address, commit.commit_id[:8]) raise SystemExit(ExitCode.USER_ERROR) historical_lines = _extract_lines( historical_raw, hist_rec["lineno"], hist_rec["end_lineno"] ) # Guard against a corrupted snapshot producing empty content. Writing # an empty replacement would silently delete the symbol — hard to debug. if not historical_lines: logger.error( "Symbol '%s' at commit %s produced no extractable lines — " "snapshot may be corrupted. Aborting without writing.", address, commit.commit_id[:8], ) raise SystemExit(ExitCode.INTERNAL_ERROR) # ------------------------------------------------------------------ # Load the current working-tree file. # Read once — reuse bytes for both line-list and symbol lookup. # ------------------------------------------------------------------ working_file = root / file_rel if working_file.exists(): current_raw = working_file.read_bytes() current_lines = current_raw.decode("utf-8", errors="replace").splitlines(keepends=True) # Pass file_rel (not the absolute path) so parse_symbols builds # addresses matching the address format used throughout Muse. cur_rec = _find_symbol_in_source(current_raw, file_rel, address) else: current_raw = b"" current_lines = [] cur_rec = None # ------------------------------------------------------------------ # No-op detection: bail out early when the bodies already match. # Only skip when NOT in dry-run mode — a dry-run explicitly requests # "show me what would happen," so it must go through the full diff # path even when the result would be unchanged. Skipping early in # dry-run mode would omit verified_preview from the JSON, breaking # agent pipelines that always pass --dry-run before writing. # ------------------------------------------------------------------ already_current = ( cur_rec is not None and cur_rec["body_hash"] == hist_rec["body_hash"] ) if already_current and not dry_run: assert cur_rec is not None # narrowed by already_current check above if as_json: print(json.dumps({ "schema_version": __version__, "address": address, "file": file_rel, "branch": branch, "restored_from": commit.commit_id[:8], "dry_run": False, "changed": False, "appended": False, "current_start": cur_rec["lineno"], "current_end": cur_rec["end_lineno"], "historical_line_count": len(historical_lines), "diff_lines": [], "verified": True, }, indent=2)) else: print( f"✅ {address} already matches commit {commit.commit_id[:8]}" " — nothing to do." ) return # ------------------------------------------------------------------ # Compute the patched file content. # ------------------------------------------------------------------ appended = cur_rec is None if cur_rec is not None: cur_start, cur_end = cur_rec["lineno"], cur_rec["end_lineno"] new_lines = current_lines[:cur_start - 1] + historical_lines + current_lines[cur_end:] else: cur_start = cur_end = 0 new_lines = current_lines + ["\n"] + historical_lines # ------------------------------------------------------------------ # Dry run — report without writing. # ------------------------------------------------------------------ if dry_run: diff_lines = list(difflib.unified_diff( current_lines, new_lines, fromfile="current", tofile="historical", lineterm="", )) # Preview whether the resulting file would be parseable — lets agents # detect malformed output before committing to a write. verified_preview = _find_symbol_in_source( "".join(new_lines).encode("utf-8"), file_rel, address ) is not None if as_json: print(json.dumps({ "schema_version": __version__, "address": address, "file": file_rel, "branch": branch, "restored_from": commit.commit_id[:8], "dry_run": True, "changed": not already_current, "appended": appended, "current_start": cur_start, "current_end": cur_end, "historical_line_count": len(historical_lines), "diff_lines": diff_lines, "verified_preview": verified_preview, }, indent=2)) else: print("Dry run — no files will be written.\n") print(f"Restoring: {sanitize_display(address)}") print(f" from commit: {commit.commit_id[:8]} ({commit.committed_at.date()})") if not verified_preview: print(" ⚠️ Warning: result would not be parseable at this address.") print("\n" + "".join(diff_lines)) return # ------------------------------------------------------------------ # Write the patched file. # ------------------------------------------------------------------ written_content = "".join(new_lines) working_file.write_text(written_content, encoding="utf-8") # Post-write verification: confirm the symbol is parseable at its address # in the content we just wrote. Uses the in-memory string to avoid a # second disk read. Catches splice edge-cases (encoding artifacts, AST # parse failures) that would silently produce a broken file. verified = _find_symbol_in_source( written_content.encode("utf-8"), file_rel, address ) is not None if not verified: logger.warning( "Post-write verification failed: '%s' not found after restore. " "The file was written but the symbol may not be at the expected " "address. Inspect with: muse code symbols --file %s", address, file_rel, ) if as_json: print(json.dumps({ "schema_version": __version__, "address": address, "file": file_rel, "branch": branch, "restored_from": commit.commit_id[:8], "dry_run": False, "changed": True, "appended": appended, "current_start": cur_start, "current_end": cur_end, "historical_line_count": len(historical_lines), "diff_lines": [], "verified": verified, }, indent=2)) else: print(f"Restoring: {sanitize_display(address)}") print(f" from commit: {commit.commit_id[:8]} ({commit.committed_at.date()})") if not appended: print( f" lines {cur_start}–{cur_end} → replaced with " f"{len(historical_lines)} historical line(s)" ) else: print(" symbol not found in working tree — appending at end of file") if verified: print(f"✅ Written to {file_rel}") else: print(f"⚠️ Written to {file_rel} — post-write verification failed")