"""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. """ import argparse import json import logging import pathlib import sys from collections.abc import Callable from typing import Literal, TypedDict from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.types import Manifest from muse.core.refs import read_current_branch from muse.core.commits import resolve_commit_ref from muse.core.snapshots import get_commit_snapshot_manifest from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.timing import start_timer 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 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 class _ChangedSymbolDict(_PublicSymbolDict, total=False): change: str breaking: bool class _ListJson(EnvelopeJson): """JSON envelope for list mode (no --diff).""" commit_id: str language_filter: str | None file_filter: str | None total: int results: list[_PublicSymbolDict] class _DiffJson(EnvelopeJson): """JSON envelope for diff mode (--diff REF).""" commit_id: str base_commit_id: str language_filter: str | None file_filter: str | None semver_impact: str stability_pct: int breaking_count: int added: list[_PublicSymbolDict] removed: list[_PublicSymbolDict] changed: list[_ChangedSymbolDict] 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", "-j", action="store_true", dest="json_out", help="Emit results as JSON (agent-friendly; -j is a shorthand alias).", ) 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. Without ``--diff``, lists all public symbols at the given ref. With ``--diff``, shows added, removed, and changed symbols between two refs. Use ``--breaking`` (requires ``--diff``) to show only breaking changes. Agent quickstart ---------------- :: muse api-surface --json # surface at HEAD muse api-surface --diff main --json # changes vs main muse api-surface --diff main --breaking --json # breaking changes only JSON fields ----------- symbols List of public symbol records at the target ref (without --diff). added Symbols added since --diff ref. removed Symbols removed since --diff ref. changed Symbols whose signature changed since --diff ref. breaking Subset of ``changed`` that are backward-incompatible. total Total count of public symbols. Exit codes ---------- 0 Success. 1 ``--breaking`` without ``--diff``, or commit not found. 2 Not inside a Muse repository. """ elapsed = start_timer() 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 json_out: bool = args.json_out if breaking_only and diff_ref is None: print("❌ --breaking requires --diff REF.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() branch = read_current_branch(root) cache = load_symbol_cache(root) commit = resolve_commit_ref(root, 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, json_out, elapsed, ) cache.save() return base_commit = resolve_commit_ref(root, 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, json_out, base_surface, elapsed, ) else: _run_diff_mode( commit.commit_id, base_commit.commit_id, language, file_filter, added, removed, changed, count_only, json_out, base_surface, elapsed, ) 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, elapsed: Callable[[], float], ) -> 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(_ListJson( **make_envelope(elapsed), commit_id=commit_id, language_filter=language, file_filter=file_filter, total=len(entries), results=[e.to_dict() for e in entries], ))) return print(f"\nPublic API surface — {commit_id}") 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, elapsed: Callable[[], float], ) -> 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(_DiffJson( **make_envelope(elapsed), 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()) ], ))) return print(f"\nPublic API surface — {commit_id} vs {base_commit_id}") 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)")