"""muse code api-surface — public API surface tracking. Shows which symbols in a snapshot are part of the public API, and how the public API changed between two commits. A symbol is **public** when all of the following hold: * ``kind`` is one of: ``function``, ``async_function``, ``class``, ``method``, ``async_method`` * ``name`` does not start with ``_`` (Python convention for private/internal) * ``kind`` is not ``import`` Muse answers "what changed in the public API between v1.0 and v1.1?" in O(1) against committed snapshots — no checkout required, no working-tree parsing. The diff output also produces a ``semver_impact`` field (``MAJOR``/``MINOR``/ ``PATCH``) that feeds directly into ``muse commit``'s bump proposal. Usage:: muse code api-surface muse code api-surface --commit HEAD~5 muse code api-surface --diff main muse code api-surface --diff main --breaking muse code api-surface --language Python muse code api-surface --file src/billing.py muse code api-surface --count muse code api-surface --json With ``--diff REF``, shows a three-section report:: Public API surface — commit a1b2c3d4 vs commit e5f6a7b8 ────────────────────────────────────────────────────────────── Added (3): + src/billing.py::compute_tax function + src/auth.py::refresh_token function + src/models.py::User.to_json method Removed (1): - src/billing.py::compute_total function ⚠ BREAKING Changed (2): ~ src/billing.py::Invoice.pay method (signature_change) ⚠ BREAKING ~ src/auth.py::validate_token function (impl_only) semver impact: MAJOR · stability: 75% · 3 breaking change(s) Flags: ``--commit, -c REF`` Show or compare from this commit (default: HEAD). ``--diff REF`` Compare the commit from ``--commit`` against this ref. ``--breaking`` In diff mode, show only breaking changes (removed + signature_change/ signature+impl). Exits non-zero when any breaking changes exist. ``--language LANG`` Filter to symbols in files of this language. ``--file PATH`` Filter to symbols in this file path (substring match). ``--count`` Print only the total symbol count (or change count in diff mode). Scriptable — exits non-zero when breaking changes exist with ``--diff``. ``--json`` Emit results as JSON with ``commit_id`` (full SHA), ``semver_impact``, ``stability_pct``, and ``breaking_count`` fields. """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import Literal from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( Manifest, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.plugins.code._query import language_of, symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolRecord from muse.core.validation import sanitize_display from typing import TypedDict logger = logging.getLogger(__name__) type FlatSymbolMap = dict[str, SymbolRecord] # address → symbol type ChangedSymbolMap = dict[str, tuple[SymbolRecord, SymbolRecord, str]] # address → (base, cur, summary) class _PublicSymbolDict(TypedDict): """JSON-serialisable form of a :class:`_PublicSymbol`.""" address: str kind: str name: str qualified_name: str language: str content_id: str signature_id: str body_hash: str SemverImpact = Literal["MAJOR", "MINOR", "PATCH", "NONE"] _PUBLIC_KINDS: frozenset[str] = frozenset({ "function", "async_function", "class", "method", "async_method", }) _BREAKING_CHANGES: frozenset[str] = frozenset({ "signature_change", "signature+impl", }) # --------------------------------------------------------------------------- # Domain helpers # --------------------------------------------------------------------------- def _is_public(name: str, kind: str) -> bool: """Return True when the symbol meets the public-API criteria.""" return kind in _PUBLIC_KINDS and not name.split(".")[-1].startswith("_") def _public_symbols( root: pathlib.Path, manifest: Manifest, language_filter: str | None, file_filter: str | None, cache: SymbolCache, ) -> FlatSymbolMap: """Return all public symbols from *manifest* as a flat ``address → SymbolRecord`` dict. Shares *cache* with the caller — no extra disk I/O. """ result: FlatSymbolMap = {} sym_map = symbols_for_snapshot( root, manifest, language_filter=language_filter, cache=cache, ) for file_path, tree in sym_map.items(): if file_filter and file_filter not in file_path: continue for address, rec in tree.items(): if _is_public(rec["name"], rec["kind"]): result[address] = rec return result def _classify_change(old: SymbolRecord, new: SymbolRecord) -> str: """Classify what changed between two versions of the same public symbol.""" if old["content_id"] == new["content_id"]: return "unchanged" if old["signature_id"] != new["signature_id"]: if old["body_hash"] != new["body_hash"]: return "signature+impl" return "signature_change" return "impl_only" def _semver_impact( added: FlatSymbolMap, removed: FlatSymbolMap, changed: ChangedSymbolMap, ) -> SemverImpact: """Infer the minimum semver bump required by the observed API changes. Rules (in priority order): - Any removal → MAJOR - Any signature_change or signature+impl → MAJOR - Any addition → MINOR - Any impl_only change → PATCH - No changes → NONE """ if removed: return "MAJOR" for _, (_, _, cls) in changed.items(): if cls in _BREAKING_CHANGES: return "MAJOR" if added: return "MINOR" if changed: return "PATCH" return "NONE" def _stability_pct( base_size: int, removed: FlatSymbolMap, changed: ChangedSymbolMap, ) -> int: """Percentage of the base API surface that survived unchanged.""" if base_size == 0: return 100 disturbed = len(removed) + len(changed) return round((base_size - disturbed) / base_size * 100) # --------------------------------------------------------------------------- # Output helper # --------------------------------------------------------------------------- class _ApiEntry: """Wraps one public symbol for display or JSON serialisation.""" __slots__ = ("address", "rec", "language") def __init__(self, address: str, rec: SymbolRecord, language: str) -> None: self.address = address self.rec = rec self.language = language def to_dict(self) -> _PublicSymbolDict: """Return a JSON-serialisable dict with full (untruncated) IDs.""" return { "address": self.address, "kind": self.rec["kind"], "name": self.rec["name"], "qualified_name": self.rec["qualified_name"], "language": self.language, "content_id": self.rec["content_id"], "signature_id": self.rec["signature_id"], "body_hash": self.rec["body_hash"], } # --------------------------------------------------------------------------- # CLI registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the api-surface subcommand.""" parser = subparsers.add_parser( "api-surface", help="Show the public API surface and how it changed between two commits.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help="Show surface at this commit (default: HEAD).", ) parser.add_argument( "--diff", default=None, metavar="REF", dest="diff_ref", help="Compare HEAD (or --commit) against this ref.", ) parser.add_argument( "--breaking", action="store_true", dest="breaking_only", help="Show only breaking changes; exit non-zero when any exist.", ) parser.add_argument( "--language", "-l", default=None, metavar="LANG", dest="language", help="Filter to this language (Python, Go, Rust, …).", ) parser.add_argument( "--file", default=None, metavar="PATH", dest="file_filter", help="Filter to symbols in this file (substring match).", ) parser.add_argument( "--count", action="store_true", dest="count_only", help="Print only the total symbol count (scriptable).", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Show the public API surface and how it changed between two commits.""" ref: str | None = args.ref diff_ref: str | None = args.diff_ref language: str | None = args.language file_filter: str | None = args.file_filter breaking_only: bool = args.breaking_only count_only: bool = args.count_only as_json: bool = args.as_json if breaking_only and diff_ref is None: print("❌ --breaking requires --diff REF.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) cache = load_symbol_cache(root) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} current_surface = _public_symbols(root, manifest, language, file_filter, cache) if diff_ref is None: _run_list_mode( root, commit.commit_id, current_surface, language, file_filter, count_only, as_json, ) cache.save() return base_commit = resolve_commit_ref(root, repo_id, branch, diff_ref) if base_commit is None: print(f"❌ Diff ref '{diff_ref}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) base_manifest = get_commit_snapshot_manifest(root, base_commit.commit_id) or {} base_surface = _public_symbols(root, base_manifest, language, file_filter, cache) cache.save() added = {a: r for a, r in current_surface.items() if a not in base_surface} removed = {a: r for a, r in base_surface.items() if a not in current_surface} changed: ChangedSymbolMap = {} for addr in current_surface: if addr in base_surface: cls = _classify_change(base_surface[addr], current_surface[addr]) if cls != "unchanged": changed[addr] = (base_surface[addr], current_surface[addr], cls) if breaking_only: added = {} removed_breaking = removed # all removals are breaking changed = {a: v for a, v in changed.items() if v[2] in _BREAKING_CHANGES} _run_diff_mode( commit.commit_id, base_commit.commit_id, language, file_filter, added, removed_breaking, changed, count_only, as_json, base_surface, ) else: _run_diff_mode( commit.commit_id, base_commit.commit_id, language, file_filter, added, removed, changed, count_only, as_json, base_surface, ) has_breaking = bool(removed) or any( v[2] in _BREAKING_CHANGES for v in changed.values() ) if has_breaking: raise SystemExit(ExitCode.USER_ERROR if breaking_only else 0) def _run_list_mode( root: pathlib.Path, commit_id: str, surface: FlatSymbolMap, language: str | None, file_filter: str | None, count_only: bool, as_json: bool, ) -> None: """Render the simple list (no --diff) output.""" entries = [ _ApiEntry(addr, rec, language_of(addr.split("::")[0])) for addr, rec in sorted(surface.items()) ] if count_only and not as_json: print(len(entries)) return if as_json: print(json.dumps( { "commit_id": commit_id, "language_filter": language, "file_filter": file_filter, "total": len(entries), "results": [e.to_dict() for e in entries], }, indent=2, )) return print(f"\nPublic API surface — {commit_id[:8]}") if language: print(f" (language: {language})") if file_filter: print(f" (file: {file_filter})") print("─" * 62) if not entries: print(" (no public symbols found)") return max_addr = max(len(e.address) for e in entries) for e in entries: print(f" {sanitize_display(e.address):<{max_addr}} {e.rec['kind']}") print(f"\n {len(entries)} public symbol(s)") def _run_diff_mode( commit_id: str, base_commit_id: str, language: str | None, file_filter: str | None, added: FlatSymbolMap, removed: FlatSymbolMap, changed: ChangedSymbolMap, count_only: bool, as_json: bool, base_surface: FlatSymbolMap, ) -> None: """Render the diff output.""" impact = _semver_impact(added, removed, changed) stability = _stability_pct(len(base_surface), removed, changed) breaking_count = len(removed) + sum( 1 for _, (_, _, cls) in changed.items() if cls in _BREAKING_CHANGES ) total_changes = len(added) + len(removed) + len(changed) if count_only and not as_json: print(total_changes) return if as_json: print(json.dumps( { "commit_id": commit_id, "base_commit_id": base_commit_id, "language_filter": language, "file_filter": file_filter, "semver_impact": impact, "stability_pct": stability, "breaking_count": breaking_count, "added": [ _ApiEntry(a, r, language_of(a.split("::")[0])).to_dict() for a, r in sorted(added.items()) ], "removed": [ _ApiEntry(a, r, language_of(a.split("::")[0])).to_dict() for a, r in sorted(removed.items()) ], "changed": [ { **_ApiEntry(a, new, language_of(a.split("::")[0])).to_dict(), "change": cls, "breaking": cls in _BREAKING_CHANGES, } for a, (_, new, cls) in sorted(changed.items()) ], }, indent=2, )) return print(f"\nPublic API surface — {commit_id[:8]} vs {base_commit_id[:8]}") if language: print(f" (language: {language})") if file_filter: print(f" (file: {file_filter})") print("─" * 62) all_addrs = sorted(set(list(added) + list(removed) + list(changed))) max_addr = max((len(a) for a in all_addrs), default=40) if added: print(f"\nAdded ({len(added)}):") for addr, rec in sorted(added.items()): print(f" + {sanitize_display(addr):<{max_addr}} {rec['kind']}") if removed: print(f"\nRemoved ({len(removed)}):") for addr, rec in sorted(removed.items()): print(f" - {sanitize_display(addr):<{max_addr}} {rec['kind']} ⚠ BREAKING") if changed: print(f"\nChanged ({len(changed)}):") for addr, (_, new, cls) in sorted(changed.items()): breaking_tag = " ⚠ BREAKING" if cls in _BREAKING_CHANGES else "" print(f" ~ {sanitize_display(addr):<{max_addr}} {new['kind']} ({cls}){breaking_tag}") if not added and not removed and not changed: print("\n ✅ No public API changes detected.") else: print(f"\n semver impact: {impact} · stability: {stability}% · {breaking_count} breaking change(s)")