"""muse code compare — semantic comparison between any two historical snapshots. ``muse diff`` compares the working tree to HEAD. ``muse code compare`` compares any two historical commits — a full semantic diff between a release tag and the current HEAD, between the start and end of a sprint, between two branches. Unlike a line-level diff, ``muse code compare`` tracks: - Functions / classes **added** or **removed** - Symbols **renamed** or **moved** across files (not just deleted + re-added) - **Signature changes** vs **implementation-only** changes Pass ``--semver`` to get an automatic MAJOR / MINOR / PATCH recommendation based on public API changes: removed symbols → MAJOR, added symbols → MINOR, implementation-only changes → PATCH. Usage:: muse code compare HEAD~10 HEAD (use commit SHAs, not ~ notation) muse code compare a3f2c9 cb4afa muse code compare a3f2c9 cb4afa --kind function muse code compare a3f2c9 cb4afa --file billing.py muse code compare a3f2c9 cb4afa --semver muse code compare a3f2c9 cb4afa --stat muse code compare a3f2c9 cb4afa --json Output:: Semantic comparison From: a3f2c9e1 "Add billing module" To: cb4afaed "Merge: release v1.0" src/billing.py added compute_invoice_total (renamed from calculate_total) modified Invoice.to_dict (signature changed) moved validate_amount → src/validation.py src/validation.py (new file) added validate_amount (moved from src/billing.py) 7 symbol change(s) across 3 file(s) SemVer impact: MINOR """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import TypedDict from muse.core._types import Manifest from muse.core.errors import ExitCode 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.symbol_cache import load_symbol_cache from muse.domain import DomainOp from muse.plugins.code._query import _SUFFIX_LANG, language_of, symbols_for_snapshot from muse.plugins.code.symbol_diff import build_diff_ops from muse.core.validation import sanitize_display from muse.plugins.registry import read_domain logger = logging.getLogger(__name__) # Language normalisation (case-insensitive --language matching). _LANG_CANONICAL: Manifest = {lang.lower(): lang for lang in set(_SUFFIX_LANG.values())} def _normalise_language(lang: str) -> str: return _LANG_CANONICAL.get(lang.strip().lower(), lang.strip()) class _OpSummary(TypedDict): op: str address: str detail: str class _CommitRef(TypedDict): commit_id: str message: str class _CompareFilters(TypedDict): kind: str | None file: str | None language: str | None class _CompareStat(TypedDict): files_changed: int symbols_added: int symbols_removed: int symbols_modified: int semver_impact: str #: JSON payload for ``muse code compare`` — functional form because ``from`` is a keyword. _CompareJson = TypedDict("_CompareJson", { "from": _CommitRef, "to": _CommitRef, "filters": _CompareFilters, "stat": _CompareStat, "ops": list[_OpSummary], }) # --------------------------------------------------------------------------- # Knowtation-specific types and helpers (Phase 2.6) # --------------------------------------------------------------------------- class _KnowtationFrontmatterDiff(TypedDict, total=False): """Per-note frontmatter delta summary for the Knowtation JSON shape.""" tags_added: list[str] tags_removed: list[str] entity_added: list[str] entity_removed: list[str] scalar_changes: dict[str, object] class _KnowtationSectionsDiff(TypedDict, total=False): """Per-note section delta summary for the Knowtation JSON shape.""" added: list[str] removed: list[str] modified: list[str] moved: list[str] class _KnowtationNoteOp(TypedDict, total=False): """A single note-level change in the Knowtation compare JSON output.""" op: str # "insert" | "delete" | "modify" path: str title: str frontmatter: _KnowtationFrontmatterDiff sections: _KnowtationSectionsDiff body_lines_changed: int class _KnowtationStat(TypedDict): """Aggregate statistics for a Knowtation vault comparison.""" notes_added: int notes_removed: int notes_modified: int frontmatter_changes: int section_changes: int tag_changes: int #: Functional TypedDict for Knowtation ``muse code compare --json`` output. _KnowtationCompareJson = TypedDict("_KnowtationCompareJson", { "from": _CommitRef, "to": _CommitRef, "domain": str, "stat": _KnowtationStat, "ops": list[_KnowtationNoteOp], }) def _note_title(content: bytes) -> str: """Extract the title from note bytes (frontmatter > h1 > filename).""" try: from muse.plugins.knowtation.parser import parse_frontmatter fm = parse_frontmatter(content) if fm and fm.title: return fm.title except Exception: pass # Fall back to first H1 heading. try: text = content.decode("utf-8", errors="replace") for line in text.splitlines(): stripped = line.strip() if stripped.startswith("# "): return stripped[2:].strip() except Exception: pass return "" def _summarise_note_delta( delta: "object", ) -> tuple[_KnowtationFrontmatterDiff, _KnowtationSectionsDiff, int]: """Extract a human-readable summary from a ``StructuredDelta``. Returns (frontmatter_diff, sections_diff, body_lines_changed). """ from muse.domain import DeleteOp, InsertOp, MoveOp, ReplaceOp fm_diff: _KnowtationFrontmatterDiff = {} sec_diff: _KnowtationSectionsDiff = {} body_lines = 0 if not isinstance(delta, dict): return fm_diff, sec_diff, body_lines ops: list[dict] = delta.get("ops") or [] tags_added: list[str] = [] tags_removed: list[str] = [] entity_added: list[str] = [] entity_removed: list[str] = [] scalar_changes: dict[str, object] = {} sec_added: list[str] = [] sec_removed: list[str] = [] sec_modified: list[str] = [] sec_moved: list[str] = [] for op in ops: op_type: str = op.get("op", "") id_val: str = op.get("address", "") # Frontmatter set ops — id like "tags::alpha" or "entity::person" if id_val.startswith("tags::"): tag = id_val[len("tags::"):] if op_type == "insert": tags_added.append(tag) elif op_type == "delete": tags_removed.append(tag) elif id_val.startswith("entity::"): ent = id_val[len("entity::"):] if op_type == "insert": entity_added.append(ent) elif op_type == "delete": entity_removed.append(ent) # Scalar frontmatter replace — id like "fm::project" elif id_val.startswith("fm::") and op_type == "replace": field = id_val[len("fm::"):] scalar_changes[field] = op.get("new_content", "") # Body line ops — address like "section:1:Background#0::line:3" # Must be checked before the plain section: prefix. elif "::line:" in id_val: if op_type in ("insert", "delete", "replace"): body_lines += 1 # Section ops — address like "section:1:Background#0" elif id_val.startswith("section:"): section_name = id_val if op_type == "insert": sec_added.append(section_name) elif op_type == "delete": sec_removed.append(section_name) elif op_type == "move": sec_moved.append(section_name) if tags_added: fm_diff["tags_added"] = sorted(tags_added) if tags_removed: fm_diff["tags_removed"] = sorted(tags_removed) if entity_added: fm_diff["entity_added"] = sorted(entity_added) if entity_removed: fm_diff["entity_removed"] = sorted(entity_removed) if scalar_changes: fm_diff["scalar_changes"] = scalar_changes if sec_added: sec_diff["added"] = sec_added if sec_removed: sec_diff["removed"] = sec_removed if sec_modified: sec_diff["modified"] = sec_modified if sec_moved: sec_diff["moved"] = sec_moved return fm_diff, sec_diff, body_lines def _run_knowtation_compare( root: pathlib.Path, commit_a: "object", commit_b: "object", manifest_a: Manifest, manifest_b: Manifest, as_json: bool, stat_only: bool, ) -> None: """Knowtation-specific compare path for ``muse code compare``. Reads note content from the object store for modified notes, calls ``diff_notes`` on each pair, and emits a vault-aware JSON/text comparison. Args: root: Repository root. commit_a: Resolved CommitRecord for the base (older) commit. commit_b: Resolved CommitRecord for the target (newer) commit. manifest_a: Content-hash manifest for the base commit. manifest_b: Content-hash manifest for the target commit. as_json: When True, emit JSON; otherwise emit human-readable text. stat_only: When True, emit only aggregate statistics. """ from muse.core.object_store import read_object from muse.plugins.knowtation.differ import diff_notes from muse.plugins.knowtation._query import is_note all_paths: set[str] = set(manifest_a) | set(manifest_b) note_paths: set[str] = {p for p in all_paths if is_note(p)} ops: list[_KnowtationNoteOp] = [] notes_added = notes_removed = notes_modified = 0 frontmatter_changes = section_changes = tag_changes = 0 for path in sorted(note_paths): hash_a = manifest_a.get(path) hash_b = manifest_b.get(path) if hash_a is None and hash_b is not None: # Added in B notes_added += 1 content_b = read_object(root, hash_b) or b"" op: _KnowtationNoteOp = { "op": "insert", "path": path, "title": _note_title(content_b), } ops.append(op) elif hash_a is not None and hash_b is None: # Removed in B notes_removed += 1 content_a = read_object(root, hash_a) or b"" op = { "op": "delete", "path": path, "title": _note_title(content_a), } ops.append(op) elif hash_a != hash_b and hash_a is not None and hash_b is not None: # Modified notes_modified += 1 content_a = read_object(root, hash_a) or b"" content_b = read_object(root, hash_b) or b"" try: delta = diff_notes(content_a, content_b) except Exception as exc: logger.debug("diff_notes failed for %s: %s", path, exc) op = {"op": "modify", "path": path, "title": _note_title(content_b)} ops.append(op) continue fm_diff, sec_diff, body_lines = _summarise_note_delta(delta) # Accumulate aggregate stats. n_tag_changes = len(fm_diff.get("tags_added", [])) + len(fm_diff.get("tags_removed", [])) n_fm_changes = n_tag_changes + len(fm_diff.get("entity_added", [])) + \ len(fm_diff.get("entity_removed", [])) + len(fm_diff.get("scalar_changes", {})) n_sec_changes = len(sec_diff.get("added", [])) + len(sec_diff.get("removed", [])) + \ len(sec_diff.get("modified", [])) + len(sec_diff.get("moved", [])) frontmatter_changes += n_fm_changes section_changes += n_sec_changes tag_changes += n_tag_changes note_op: _KnowtationNoteOp = { "op": "modify", "path": path, "title": _note_title(content_b), "body_lines_changed": body_lines, } if fm_diff: note_op["frontmatter"] = fm_diff if sec_diff: note_op["sections"] = sec_diff ops.append(note_op) stat: _KnowtationStat = { "notes_added": notes_added, "notes_removed": notes_removed, "notes_modified": notes_modified, "frontmatter_changes": frontmatter_changes, "section_changes": section_changes, "tag_changes": tag_changes, } msg_a = _first_line(commit_a.message) msg_b = _first_line(commit_b.message) if as_json: out = _KnowtationCompareJson(**{ "from": _CommitRef(commit_id=commit_a.commit_id, message=msg_a), "to": _CommitRef(commit_id=commit_b.commit_id, message=msg_b), "domain": "knowtation", "stat": stat, "ops": ops, }) print(json.dumps(out, indent=2)) return # ── Human-readable vault comparison ───────────────────────────────────── total_notes = notes_added + notes_removed + notes_modified print("\nVault comparison (domain: knowtation)") print(f' From: {commit_a.commit_id[:8]} "{sanitize_display(msg_a)}"') print(f' To: {commit_b.commit_id[:8]} "{sanitize_display(msg_b)}"') if stat_only: print(f"\n Notes added: {notes_added}") print(f" Notes removed: {notes_removed}") print(f" Notes modified: {notes_modified}") print(f" Frontmatter changes: {frontmatter_changes}") print(f" Section changes: {section_changes}") print(f" Tag changes: {tag_changes}") return if not ops: print("\n (no note changes between these two commits)") return for note_op in ops: op_label = note_op["op"] note_path = note_op["path"] note_title = note_op.get("title", "") title_str = f' "{sanitize_display(note_title)}"' if note_title else "" if op_label == "insert": print(f"\n added {sanitize_display(note_path)}{title_str}") elif op_label == "delete": print(f"\n removed {sanitize_display(note_path)}{title_str}") else: print(f"\n modified {sanitize_display(note_path)}{title_str}") fm = note_op.get("frontmatter", {}) sec = note_op.get("sections", {}) body = note_op.get("body_lines_changed", 0) if fm.get("tags_added"): print(f" tags+: {', '.join(fm['tags_added'])}") if fm.get("tags_removed"): print(f" tags-: {', '.join(fm['tags_removed'])}") if sec.get("added"): print(f" §+: {', '.join(sec['added'])}") if sec.get("removed"): print(f" §-: {', '.join(sec['removed'])}") if body: print(f" ~{body} line(s) changed") print(f"\n{total_notes} note change(s) across {notes_added + notes_removed + notes_modified} note(s)") def _first_line(message: str) -> str: """Return the first non-empty line of a commit message.""" for line in message.splitlines(): stripped = line.strip() if stripped: return stripped return message.strip() def _is_public_name(address: str) -> bool: """Return True if the symbol is a public, non-import API symbol. Excludes: - Import pseudo-symbols (``::import::*``) — dependency management, not API. - Private names starting with ``_``. """ if "::import::" in address: return False name = address.split("::")[-1] if "::" in address else address return not name.startswith("_") def _assess_semver(ops: list[DomainOp]) -> tuple[str, list[str]]: """Return (bump_level, reasons) from the diff ops. Rules applied to public symbols only (names not starting with ``_``): MAJOR — symbol removed, or signature / rename / move change. MINOR — symbol added (no MAJOR triggers). PATCH — only implementation changes. NONE — empty diff. """ reasons: list[str] = [] has_major = False has_minor = False has_patch = False for op in ops: if op["op"] == "patch": for child in op["child_ops"]: addr: str = child["address"] if not _is_public_name(addr): continue if child["op"] == "delete": has_major = True reasons.append(f"removed: {addr}") elif child["op"] == "insert": has_minor = True elif child["op"] == "replace": ns: str = child["new_summary"] if any(kw in ns for kw in ("signature", "renamed", "moved")): has_major = True reasons.append(f"breaking: {addr} — {ns}") else: has_patch = True elif op["op"] == "delete": addr2: str = op["address"] if _is_public_name(addr2): has_major = True reasons.append(f"file removed: {addr2}") elif op["op"] == "insert": if _is_public_name(op["address"]): has_minor = True if has_major: return "MAJOR", reasons if has_minor: return "MINOR", [] if has_patch: return "PATCH", [] return "NONE", [] def _count_ops(ops: list[DomainOp]) -> tuple[int, int, int, int]: """Return (added, removed, modified, files_changed) from diff ops.""" added = removed = modified = 0 files: set[str] = set() for op in ops: if op["op"] == "patch": if op["child_ops"]: files.add(op["address"]) for child in op["child_ops"]: if child["op"] == "insert": added += 1 elif child["op"] == "delete": removed += 1 elif child["op"] == "replace": modified += 1 elif op["op"] == "insert": files.add(op["address"]) added += 1 elif op["op"] == "delete": files.add(op["address"]) removed += 1 elif op["op"] == "replace": files.add(op["address"]) modified += 1 return added, removed, modified, len(files) def _format_child_op(op: DomainOp) -> str: """Return a compact one-line description of a symbol-level op.""" addr: str = op["address"] name = addr.split("::")[-1] if "::" in addr else addr if op["op"] == "insert": summary: str = op["content_summary"] moved = ( f" (moved from {summary.split('moved from')[-1].strip()})" if "moved from" in summary else "" ) return f" added {name}{moved}" if op["op"] == "delete": summary2: str = op["content_summary"] moved2 = ( f" (moved to {summary2.split('moved to')[-1].strip()})" if "moved to" in summary2 else "" ) return f" removed {name}{moved2}" if op["op"] == "replace": ns: str = op["new_summary"] detail = f" ({ns})" if ns else "" return f" modified {name}{detail}" return f" changed {name}" def _flatten_ops(ops: list[DomainOp]) -> list[_OpSummary]: """Flatten all ops to a serialisable summary list.""" result: list[_OpSummary] = [] for op in ops: if op["op"] == "patch": for child in op["child_ops"]: if child["op"] == "insert": detail: str = child["content_summary"] elif child["op"] == "delete": detail = child["content_summary"] elif child["op"] == "replace": detail = child["new_summary"] else: detail = "" result.append(_OpSummary( op=child["op"], address=child["address"], detail=detail, )) elif op["op"] == "insert": result.append(_OpSummary(op="insert", address=op["address"], detail=op["content_summary"])) elif op["op"] == "delete": result.append(_OpSummary(op="delete", address=op["address"], detail=op["content_summary"])) elif op["op"] == "replace": result.append(_OpSummary(op="replace", address=op["address"], detail=op["new_summary"])) else: result.append(_OpSummary(op=op["op"], address=op["address"], detail="")) return result def _filter_manifest_by_language( manifest: Manifest, language: str ) -> Manifest: return {p: h for p, h in manifest.items() if language_of(p) == language} def _filter_ops_by_file(ops: list[DomainOp], file_filter: str) -> list[DomainOp]: """Keep only ops whose address is or ends with *file_filter*.""" return [ op for op in ops if op.get("address", "") == file_filter or op.get("address", "").endswith("/" + file_filter) ] def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the compare subcommand.""" parser = subparsers.add_parser( "compare", help="Deep semantic comparison between any two historical snapshots.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("ref_a", metavar="REF-A", help="Base commit (older).") parser.add_argument("ref_b", metavar="REF-B", help="Target commit (newer).") parser.add_argument( "--kind", "-k", default=None, metavar="KIND", dest="kind_filter", help="Restrict to symbols of this kind (function, class, method, …).", ) parser.add_argument( "--file", "-f", default=None, metavar="FILE", dest="file_filter", help="Only show changes in this file (accepts a path suffix, e.g. 'billing.py').", ) parser.add_argument( "--language", "-l", default=None, metavar="LANG", dest="language_filter", help="Only show changes in files of this language (case-insensitive).", ) parser.add_argument( "--stat", action="store_true", help="Print a summary (counts + SemVer impact) instead of the full listing.", ) parser.add_argument( "--semver", action="store_true", help=( "Append a MAJOR / MINOR / PATCH recommendation based on public API changes: " "symbol removed → MAJOR, symbol added → MINOR, impl-only → PATCH." ), ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as JSON.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Deep semantic comparison between any two historical snapshots. ``muse code compare`` reads both commits from the object store, parses AST symbol trees for all semantic files, and produces a full symbol-level delta: which functions were added, removed, renamed, moved, and modified between these two points. Use it to understand the semantic scope of a release, a sprint, or a branch divergence — at the function level, not the line level. """ ref_a: str = args.ref_a ref_b: str = args.ref_b kind_filter: str | None = args.kind_filter file_filter: str | None = args.file_filter language_filter: str | None = args.language_filter stat_only: bool = args.stat show_semver: bool = args.semver as_json: bool = args.as_json if language_filter is not None: language_filter = _normalise_language(language_filter) if kind_filter is not None: kind_filter = kind_filter.strip().lower() root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) commit_a = resolve_commit_ref(root, repo_id, branch, ref_a) if commit_a is None: print(f"❌ Commit '{ref_a}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) commit_b = resolve_commit_ref(root, repo_id, branch, ref_b) if commit_b is None: print(f"❌ Commit '{ref_b}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest_a: Manifest = get_commit_snapshot_manifest(root, commit_a.commit_id) or {} manifest_b: Manifest = get_commit_snapshot_manifest(root, commit_b.commit_id) or {} # Knowtation domain: use vault-aware note comparison instead of symbol diff. try: domain = read_domain(root) except Exception: domain = "code" if domain == "knowtation": _run_knowtation_compare( root=root, commit_a=commit_a, commit_b=commit_b, manifest_a=manifest_a, manifest_b=manifest_b, as_json=as_json, stat_only=stat_only, ) return # Apply language filter at the manifest level so cross-file move detection # in build_diff_ops only considers the requested language. if language_filter is not None: manifest_a = _filter_manifest_by_language(manifest_a, language_filter) manifest_b = _filter_manifest_by_language(manifest_b, language_filter) # Load the symbol cache once and share it across both snapshot loads. shared_cache = load_symbol_cache(root) trees_a = symbols_for_snapshot( root, manifest_a, kind_filter=kind_filter, cache=shared_cache ) trees_b = symbols_for_snapshot( root, manifest_b, kind_filter=kind_filter, cache=shared_cache ) ops = build_diff_ops(manifest_a, manifest_b, trees_a, trees_b) # Post-filter by --file (applied after build_diff_ops so cross-file move # detection still runs on the full op set). if file_filter is not None: ops = _filter_ops_by_file(ops, file_filter) semver_bump, semver_reasons = _assess_semver(ops) added, removed, modified, files_changed = _count_ops(ops) total_symbols = added + removed + modified msg_a = _first_line(commit_a.message) msg_b = _first_line(commit_b.message) if as_json: out = _CompareJson(**{ "from": _CommitRef(commit_id=commit_a.commit_id, message=msg_a), "to": _CommitRef(commit_id=commit_b.commit_id, message=msg_b), "filters": _CompareFilters(kind=kind_filter, file=file_filter, language=language_filter), "stat": _CompareStat( files_changed=files_changed, symbols_added=added, symbols_removed=removed, symbols_modified=modified, semver_impact=semver_bump, ), "ops": list(_flatten_ops(ops)), }) print(json.dumps(out, indent=2)) return # ── Human-readable ──────────────────────────────────────────────────────── print("\nSemantic comparison") print(f' From: {commit_a.commit_id[:8]} "{sanitize_display(msg_a)}"') print(f' To: {commit_b.commit_id[:8]} "{sanitize_display(msg_b)}"') if stat_only: print(f"\n Files changed: {files_changed}") print(f" Symbols added: {added}") print(f" Symbols removed: {removed}") print(f" Symbols modified: {modified}") print(f" Net change: {added - removed:+d} symbols") print(f" SemVer impact: {semver_bump}") if semver_reasons: for r in semver_reasons[:5]: print(f" → {r}") return if not ops: print("\n (no semantic changes between these two commits)") return for op in ops: if op["op"] == "patch": fp = op["address"] child_ops = op.get("child_ops", []) if not child_ops: continue is_new = fp not in manifest_a is_gone = fp not in manifest_b suffix = " (new file)" if is_new else (" (removed)" if is_gone else "") print(f"\n{sanitize_display(fp)}{suffix}") for child in child_ops: print(_format_child_op(child)) else: fp = op["address"] if op["op"] == "insert": print(f"\n{sanitize_display(fp)} (new file)") print(f" added {sanitize_display(fp)} (file)") elif op["op"] == "delete": print(f"\n{sanitize_display(fp)} (removed)") print(f" removed {sanitize_display(fp)} (file)") else: print(f"\n{sanitize_display(fp)}") print(f" modified {sanitize_display(fp)} (file)") print(f"\n{total_symbols} symbol change(s) across {files_changed} file(s)") if show_semver or semver_bump in ("MAJOR",): print(f"SemVer impact: {semver_bump}") for r in semver_reasons[:5]: print(f" → {r}")