"""Version-history and changelog generation for ``muse code docs``. Reads the symbol history index (built by ``muse code index rebuild``) and the commit/tag store to answer three questions: * *When was this symbol first introduced?* :func:`get_symbol_version_events` * *Which release first shipped it?* :func:`infer_since_version` * *What changed between v1.0 and v2.0?* :func:`generate_changelog` * *Is the docstring stale?* :func:`detect_stale_docstring` All reads are from the content-addressed object store and the msgpack index. No working-tree files are modified and no subprocesses are spawned. Performance ----------- The most expensive operation is :func:`generate_changelog`, which walks the commit graph between two refs. On a 10,000-commit repo this takes <200 ms because ``read_commit`` is a single file read and the symbol history index provides O(1) address lookup. Security -------- * All commit refs are resolved through :func:`~muse.core.store.resolve_commit_ref`, which strips glob metacharacters and validates the ref string before any file I/O. * The commit-graph walk is bounded by *max_commits* to prevent runaway traversal of a malformed or adversarially crafted repository. * No user-supplied data is evaluated or executed. """ from __future__ import annotations import logging import pathlib from typing import Literal, NotRequired, TypedDict from muse.core.indices import SymbolHistoryEntry, load_symbol_history from muse.core.store import ( MsgpackValue, TagRecord, get_all_tags, read_commit, read_current_branch, resolve_commit_ref, walk_commits_between, _str_val, ) from muse.plugins.code.ast_parser import SymbolKind type _StrMap = dict[str, str] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Public type definitions # --------------------------------------------------------------------------- class SymbolVersionEvent(TypedDict): """One entry in a symbol's version history timeline.""" commit_id: str """The commit in which this event occurred.""" committed_at: str """ISO-8601 timestamp from the commit record.""" op: str """Operation: ``"insert"`` | ``"replace"`` | ``"delete"``.""" version: str | None """Tag name that maps directly to this commit, or ``None``.""" sem_ver_bump: str | None """Semantic version bump recorded on the commit (``"major"`` | ``"minor"`` | ``"patch"`` | ``"none"``), or ``None`` when the commit is absent.""" breaking: bool """``True`` when the commit's ``breaking_changes`` list is non-empty.""" class ChangelogEntry(TypedDict): """One symbol's entry in a documentation changelog.""" address: str """Fully-qualified symbol address, e.g. ``"muse/core/store.py::read_commit"``.""" name: str """Bare symbol name.""" kind: SymbolKind """Symbol kind (``"function"``, ``"class"``, etc.).""" file: str """Source file path, workspace-relative.""" class ChangelogReport(TypedDict): """Structured changelog between two commits or version tags.""" from_ref: str """Starting reference (exclusive). May be a commit ID or tag name.""" to_ref: str """Ending reference (inclusive). May be a commit ID or tag name.""" added: list[ChangelogEntry] """Symbols that first appeared (``"insert"`` op) within the range.""" removed: list[ChangelogEntry] """Symbols that were deleted within the range.""" changed: list[ChangelogEntry] """Symbols modified non-breakingly within the range.""" breaking: list[ChangelogEntry] """Symbols involved in a breaking change within the range.""" class StaleInfo(TypedDict): """Docstring staleness assessment for one symbol.""" is_stale: bool """``True`` when the implementation changed more recently than the body (which includes the docstring).""" last_doc_commit: str | None """Commit ID of the last event that changed the body (heuristic for docstring edit). ``None`` when insufficient history exists.""" last_impl_commit: str | None """Commit ID of the last event that changed the signature. ``None`` when there has been no signature change.""" signature_changed: bool """``True`` when the signature changed *after* the last body-only change.""" body_changed: bool """``True`` when the body changed *after* the last signature change (reversed drift — implementation regressed toward original, rare but possible).""" # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _build_commit_to_version_map( root: pathlib.Path, repo_id: str ) -> _StrMap: """Return ``{commit_id: tag_name}`` for every tag recorded in the repository. When a single commit carries multiple tags, the last-sorted tag wins (stable, deterministic output regardless of filesystem iteration order). """ tags = get_all_tags(root, repo_id) result: _StrMap = {} for tag in sorted(tags, key=lambda t: t.tag): result[tag.commit_id] = tag.tag return result def _resolve_ref_to_commit_id( root: pathlib.Path, repo_id: str, branch: str, ref: str, ) -> str | None: """Resolve *ref* to a commit ID, accepting tag names, commit IDs, and HEAD notation.""" # First try the standard resolve path (handles HEAD, HEAD~N, SHA prefixes). commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is not None: return commit.commit_id # Fallback: look up *ref* as a tag name. for tag in get_all_tags(root, repo_id): if tag.tag == ref: return tag.commit_id return None # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def get_symbol_version_events( root: pathlib.Path, repo_id: str, address: str, ) -> list[SymbolVersionEvent]: """Return the full version history timeline for *address*. Events are ordered oldest-first (insertion order in the index). Returns an empty list when the index does not contain the address or has not been built yet. Args: root: Repository root directory. repo_id: Repository UUID — used to map commits to version tags. address: Symbol address e.g. ``"muse/core/store.py::read_commit"``. """ index = load_symbol_history(root) entries = index.get(address, []) if not entries: return [] commit_version_map = _build_commit_to_version_map(root, repo_id) events: list[SymbolVersionEvent] = [] for entry in entries: commit = read_commit(root, entry.commit_id) sem_ver_bump: str | None = None breaking = False if commit is not None: raw_bump = commit.sem_ver_bump sem_ver_bump = raw_bump if raw_bump != "none" else None breaking = bool(commit.breaking_changes) events.append( SymbolVersionEvent( commit_id=entry.commit_id, committed_at=entry.committed_at, op=entry.op, version=commit_version_map.get(entry.commit_id), sem_ver_bump=sem_ver_bump, breaking=breaking, ) ) return events def infer_since_version(events: list[SymbolVersionEvent]) -> str | None: """Return the version tag of the commit that first introduced this symbol. Scans *events* (oldest-first) for an ``"insert"`` op with a non-``None`` ``version`` field. If no tagged insertion exists, falls back to the first event of any op that carries a version. Args: events: Output of :func:`get_symbol_version_events`, oldest-first. Returns: The tag name string (e.g. ``"v1.2.0"``) or ``None`` when no tagged commit is found. """ for event in events: if event["op"] == "insert" and event["version"] is not None: return event["version"] for event in events: if event["version"] is not None: return event["version"] return None def infer_last_changed_version(events: list[SymbolVersionEvent]) -> str | None: """Return the version tag of the most recent modifying event. Scans *events* in newest-first order for a ``"replace"`` or ``"delete"`` op that carries a version tag. Args: events: Output of :func:`get_symbol_version_events`, oldest-first. """ for event in reversed(events): if event["op"] in ("replace", "delete") and event["version"] is not None: return event["version"] return None def detect_stale_docstring( root: pathlib.Path, address: str, ) -> StaleInfo: """Determine whether a symbol's docstring is potentially stale. Compares the ``signature_id`` and ``body_hash`` timelines. When the signature changed *after* the last body-only event, the docstring may not reflect the current API — hence "stale." This is necessarily a heuristic: the index tracks content hashes, not which specific lines inside the body changed. A one-line body edit that is *not* a docstring update will produce the same signal as a pure docstring update. When the index is absent or there are fewer than two events, returns a :class:`StaleInfo` with ``is_stale=False`` (optimistic default — no evidence of staleness). Args: root: Repository root directory. address: Symbol address. """ index = load_symbol_history(root) entries = index.get(address, []) if len(entries) < 2: return StaleInfo( is_stale=False, last_doc_commit=None, last_impl_commit=None, signature_changed=False, body_changed=False, ) last_sig_change_idx: int | None = None last_body_change_idx: int | None = None prev = entries[0] for i, entry in enumerate(entries[1:], start=1): if entry.signature_id != prev.signature_id: last_sig_change_idx = i if entry.body_hash != prev.body_hash: last_body_change_idx = i prev = entry if last_body_change_idx is None: return StaleInfo( is_stale=False, last_doc_commit=None, last_impl_commit=None, signature_changed=False, body_changed=False, ) sig_changed_after = ( last_sig_change_idx is not None and last_sig_change_idx > last_body_change_idx ) body_changed_after = ( last_sig_change_idx is not None and last_body_change_idx > last_sig_change_idx ) last_body_entry = entries[last_body_change_idx] last_impl_commit: str | None = ( entries[last_sig_change_idx].commit_id if last_sig_change_idx is not None else None ) return StaleInfo( is_stale=sig_changed_after or body_changed_after, last_doc_commit=last_body_entry.commit_id, last_impl_commit=last_impl_commit, signature_changed=sig_changed_after, body_changed=body_changed_after, ) def generate_changelog( root: pathlib.Path, repo_id: str, from_ref: str, to_ref: str, max_commits: int = 10_000, ) -> ChangelogReport: """Build a structured changelog between *from_ref* and *to_ref*. Walks the commit graph from *to_ref* back to *from_ref* (exclusive), then uses the symbol history index to classify each symbol address touched in that range as added, removed, changed, or breaking. Args: root: Repository root directory. repo_id: Repository UUID. from_ref: Exclusive start — tag name, commit ID, or HEAD notation. to_ref: Inclusive end — tag name, commit ID, or HEAD notation. max_commits: Safety cap on the number of commits to walk. Returns: A fully-typed :class:`ChangelogReport`. """ branch_raw = (root / ".muse" / "HEAD").read_text(encoding="utf-8").strip() if branch_raw.startswith("ref: refs/heads/"): branch = branch_raw[len("ref: refs/heads/"):] else: branch = "main" from_commit_id = _resolve_ref_to_commit_id(root, repo_id, branch, from_ref) to_commit_id = _resolve_ref_to_commit_id(root, repo_id, branch, to_ref) if to_commit_id is None: logger.warning("⚠️ Cannot resolve to_ref '%s' — returning empty changelog", to_ref) return ChangelogReport( from_ref=from_ref, to_ref=to_ref, added=[], removed=[], changed=[], breaking=[], ) commits = walk_commits_between( root, to_commit_id=to_commit_id, from_commit_id=from_commit_id, max_commits=max_commits, ) # Collect the set of commit IDs in this range for fast lookup. range_commit_ids: frozenset[str] = frozenset(c.commit_id for c in commits) # Build a breaking-commit set for the range. breaking_commits: frozenset[str] = frozenset( c.commit_id for c in commits if c.breaking_changes ) # Walk the symbol history index to classify each address. index = load_symbol_history(root) added: list[ChangelogEntry] = [] removed: list[ChangelogEntry] = [] changed: list[ChangelogEntry] = [] breaking: list[ChangelogEntry] = [] seen_addresses: set[str] = set() for address, entries in index.items(): relevant = [e for e in entries if e.commit_id in range_commit_ids] if not relevant: continue if address in seen_addresses: continue seen_addresses.add(address) # Parse the address to extract file and name. if "::" in address: file_part, sym_part = address.split("::", 1) else: file_part = address sym_part = address # Determine the dominant op for this address in the range. ops = {e.op for e in relevant} has_breaking = any(e.commit_id in breaking_commits for e in relevant) # Build a minimal ChangelogEntry without requiring the full SymbolRecord. entry: ChangelogEntry = ChangelogEntry( address=address, name=sym_part.split(".")[-1], kind="function", # best-effort default; callers may enrich file=file_part, ) if has_breaking: breaking.append(entry) elif "insert" in ops and "delete" not in ops: added.append(entry) elif "delete" in ops: removed.append(entry) else: changed.append(entry) return ChangelogReport( from_ref=from_ref, to_ref=to_ref, added=sorted(added, key=lambda e: e["address"]), removed=sorted(removed, key=lambda e: e["address"]), changed=sorted(changed, key=lambda e: e["address"]), breaking=sorted(breaking, key=lambda e: e["address"]), )