"""muse code semantic-cherry-pick — cherry-pick specific symbols, not files. Extracts named symbols from a source commit and applies them to the current working tree, replacing only those symbols. All other code is left untouched. This is the semantic counterpart to ``git cherry-pick``, which operates at the file-hunk level. ``muse code semantic-cherry-pick`` operates at the symbol level: you name the exact functions, classes, or methods you want to bring forward. Multiple symbols can be cherry-picked in a single invocation. They are applied left-to-right. Failures are recorded and processing continues with the remaining symbols — the full result set is always returned. Security note: every file path extracted from ADDRESS arguments is validated via ``contain_path()`` before any disk access or directory creation. Paths that escape the repo root (e.g. ``../../etc/shadow::foo``) are rejected and the symbol is recorded as ``not_found`` with an appropriate detail message. Usage:: muse code semantic-cherry-pick "src/billing.py::compute_total" --from abc12345 muse code semantic-cherry-pick \\ "src/auth.py::validate_token" \\ "src/auth.py::refresh_token" \\ --from feature-branch muse code semantic-cherry-pick "src/core.py::hash_content" --from HEAD~5 --dry-run muse code semantic-cherry-pick "src/billing.py::Invoice.pay" --from v1.0 --json Output:: Semantic cherry-pick from commit abc12345 ────────────────────────────────────────────────────────────── ✅ src/auth.py::validate_token applied (lines 12–34 → 29 lines) ✅ src/auth.py::refresh_token applied (lines 36–58 → 17 lines) ❌ src/billing.py::compute_total not found in source commit 2 applied, 1 failed Flags: ``--from REF`` Required. Commit or branch to cherry-pick from. ``--dry-run`` Print what would change without writing anything. Each result still includes ``diff_lines`` and ``verified`` so agents can gate on output quality before committing to a write. ``--json`` Emit per-symbol results as JSON (includes ``diff_lines`` and ``verified``). """ from __future__ import annotations import argparse import difflib import json import logging import pathlib from typing import Literal, TypedDict from muse 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 parse_symbols from muse.core.store import Manifest type _BlobCache = dict[str, bytes] logger = logging.getLogger(__name__) ApplyStatus = Literal["applied", "not_found", "file_missing", "parse_error", "already_current"] class _PickResultDict(TypedDict): """JSON schema for one cherry-pick result.""" address: str status: str detail: str old_lines: int new_lines: int diff_lines: list[str] verified: bool class _PickResult: """Result for one cherry-picked symbol.""" def __init__( self, address: str, status: ApplyStatus, detail: str = "", old_lines: int = 0, new_lines: int = 0, diff_lines: list[str] | None = None, verified: bool = True, ) -> None: self.address = address self.status = status self.detail = detail self.old_lines = old_lines self.new_lines = new_lines self.diff_lines: list[str] = diff_lines if diff_lines is not None else [] self.verified = verified def to_dict(self) -> _PickResultDict: return { "address": self.address, "status": self.status, "detail": self.detail, "old_lines": self.old_lines, "new_lines": self.new_lines, "diff_lines": self.diff_lines, "verified": self.verified, } def _verify_symbol(working_file: pathlib.Path, file_rel: str, address: str) -> bool: """Re-parse *working_file* and confirm *address* is locatable after a write. Returns ``False`` if the write produced syntactically invalid output or if the symbol is no longer addressable — a useful signal to agents that the splice needs human review. """ try: tree = parse_symbols(working_file.read_bytes(), file_rel) return tree.get(address) is not None except Exception: return False def _apply_symbol( root: pathlib.Path, address: str, src_manifest: Manifest, dry_run: bool, src_cache: _BlobCache, ) -> _PickResult: """Apply one symbol from *src_manifest* to the working tree. *src_cache* is a content-addressed blob cache keyed by object ID. Passing the same dict across calls for a single invocation ensures that multiple addresses targeting the same source file only fetch the blob once. """ if "::" not in address: return _PickResult(address, "not_found", "address has no '::' separator") file_rel = address.split("::")[0] # Validate the file path stays inside the repo root before any I/O. try: working_file = contain_path(root, file_rel) except ValueError as exc: return _PickResult(address, "not_found", str(exc)) # Read historical blob — use src_cache so repeated addresses targeting # the same source file pay the fetch + decode cost only once. obj_id = src_manifest.get(file_rel) if obj_id is None: return _PickResult(address, "file_missing", f"'{file_rel}' not in source snapshot") if obj_id not in src_cache: raw = read_object(root, obj_id) if raw is None: return _PickResult(address, "file_missing", f"blob {obj_id[:8]} missing from object store") src_cache[obj_id] = raw src_raw = src_cache[obj_id] try: src_tree = parse_symbols(src_raw, file_rel) except Exception as exc: return _PickResult(address, "parse_error", str(exc)) src_rec = src_tree.get(address) if src_rec is None: return _PickResult(address, "not_found", "symbol not found in source commit") src_text = src_raw.decode("utf-8", errors="replace") src_lines_all = src_text.splitlines(keepends=True) src_symbol_lines = src_lines_all[src_rec["lineno"] - 1 : src_rec["end_lineno"]] # Read current working tree. if not working_file.exists(): diff_lines = list(difflib.unified_diff( [], src_symbol_lines, fromfile="current", tofile="historical", lineterm="", )) if not dry_run: working_file.parent.mkdir(parents=True, exist_ok=True) working_file.write_text("".join(src_symbol_lines), encoding="utf-8") verified = _verify_symbol(working_file, file_rel, address) else: # Dry-run: simulate verification in memory. try: sim_tree = parse_symbols("".join(src_symbol_lines).encode(), file_rel) verified = sim_tree.get(address) is not None except Exception: verified = False return _PickResult(address, "applied", "created file", 0, len(src_symbol_lines), diff_lines, verified) current_text = working_file.read_text(encoding="utf-8", errors="replace") current_lines = current_text.splitlines(keepends=True) current_raw = current_text.encode("utf-8") try: current_tree = parse_symbols(current_raw, file_rel) except Exception as exc: return _PickResult(address, "parse_error", f"current file: {exc}") current_rec = current_tree.get(address) if current_rec is not None: if current_rec["content_id"] == src_rec["content_id"]: return _PickResult(address, "already_current", "content identical", 0, 0) old_start = current_rec["lineno"] - 1 old_end = current_rec["end_lineno"] current_symbol_lines = current_lines[old_start:old_end] new_lines = current_lines[:old_start] + src_symbol_lines + current_lines[old_end:] detail = f"lines {current_rec['lineno']}–{current_rec['end_lineno']} → {len(src_symbol_lines)} lines" else: # Symbol not in current tree — append at end. current_symbol_lines = [] new_lines = current_lines + ["\n"] + src_symbol_lines detail = "appended at end (symbol not found in current tree)" diff_lines = list(difflib.unified_diff( current_symbol_lines, src_symbol_lines, fromfile="current", tofile="historical", lineterm="", )) if not dry_run: working_file.write_text("".join(new_lines), encoding="utf-8") verified = _verify_symbol(working_file, file_rel, address) else: # Dry-run: simulate verification in memory. try: sim_tree = parse_symbols("".join(new_lines).encode(), file_rel) verified = sim_tree.get(address) is not None except Exception: verified = False return _PickResult(address, "applied", detail, len(current_symbol_lines), len(src_symbol_lines), diff_lines, verified) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the semantic-cherry-pick subcommand.""" parser = subparsers.add_parser( "semantic-cherry-pick", help="Cherry-pick specific named symbols from a historical commit.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "addresses", nargs="+", metavar="ADDRESS", help='Symbol addresses to cherry-pick, e.g. "src/auth.py::validate_token".', ) parser.add_argument( "--from", dest="from_ref", required=True, metavar="REF", help="Commit or branch to cherry-pick symbols from (required).", ) parser.add_argument( "--dry-run", action="store_true", help="Print what would change without writing anything.", ) parser.add_argument( "--json", dest="as_json", action="store_true", help="Emit per-symbol results as JSON (includes diff_lines and verified).", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Cherry-pick specific named symbols from a historical commit. Extracts each listed symbol from the source commit and splices it into the current working-tree file at the symbol's current location. Only the target symbol's lines change; all surrounding code is preserved. If the symbol does not exist in the current working tree, the historical version is appended to the end of the file. Failures are recorded and processing continues with remaining symbols; the full result set is always returned. ``--dry-run`` shows what would change without writing anything. ``--json`` emits per-symbol results for machine consumption. """ addresses: list[str] = args.addresses from_ref: str = args.from_ref dry_run: bool = args.dry_run as_json: bool = args.as_json root = require_repo() try: repo_id = read_repo_id(root) except (OSError, json.JSONDecodeError, KeyError) as exc: logger.error("❌ Could not read repo identity: %s", exc) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc try: branch = read_current_branch(root) except Exception as exc: logger.error("❌ Could not read current branch: %s", exc) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc if not addresses: logger.error("❌ At least one ADDRESS is required.") raise SystemExit(ExitCode.USER_ERROR) from_commit = resolve_commit_ref(root, repo_id, branch, from_ref) if from_commit is None: logger.error("❌ --from ref '%s' not found.", from_ref) raise SystemExit(ExitCode.USER_ERROR) src_manifest = get_commit_snapshot_manifest(root, from_commit.commit_id) if src_manifest is None: logger.error( "❌ Snapshot for commit %s is missing from the object store.", from_commit.commit_id[:8], ) raise SystemExit(ExitCode.INTERNAL_ERROR) # src_cache avoids re-fetching the same blob when multiple addresses # target the same source file within a single invocation. src_cache: _BlobCache = {} results: list[_PickResult] = [] for address in addresses: result = _apply_symbol(root, address, src_manifest, dry_run, src_cache) results.append(result) n_applied = sum(1 for r in results if r.status == "applied") n_already = sum(1 for r in results if r.status == "already_current") n_failed = sum(1 for r in results if r.status not in ("applied", "already_current")) unverified = [r.address for r in results if r.status == "applied" and not r.verified] if as_json: print(json.dumps( { "schema_version": __version__, "branch": branch, "from_commit": from_commit.commit_id[:8], "dry_run": dry_run, "results": [r.to_dict() for r in results], "applied": n_applied, "already_current": n_already, "failed": n_failed, "unverified": unverified, }, indent=2, )) return action = "Dry-run" if dry_run else "Semantic cherry-pick" print(f"\n{action} from commit {from_commit.commit_id[:8]}") print("─" * 62) max_addr = max(len(r.address) for r in results) for r in results: if r.status == "applied": icon = "✅" label = f"applied ({r.detail})" if not r.verified: label += " ⚠️ unverified — re-parse failed after write" elif r.status == "already_current": icon = "ℹ️ " label = "already current — no change needed" else: icon = "❌" label = f"{r.status} ({r.detail})" print(f"\n {icon} {sanitize_display(r.address):<{max_addr}} {label}") print(f"\n {n_applied} applied, {n_failed} failed") if n_already: print(f" {n_already} already current") if dry_run: print(" (dry run — no files were written)") if unverified: logger.warning( "⚠️ %d symbol(s) could not be verified after write: %s", len(unverified), ", ".join(unverified), )