"""Symbol-aware documentation extraction for ``muse code docs``. Assembles :class:`SymbolDoc` records by combining four data sources: 1. **Symbol graph** — :class:`~muse.core.symbol_cache.SymbolCache` gives the full set of symbols in the committed snapshot (kind, name, line ranges). 2. **Call graph** — :mod:`muse.plugins.code._callgraph` supplies ``ForwardGraph`` (caller → callees) and the reverse mapping (callee → callers), so every doc page shows who calls a symbol and what it calls. 3. **Version history** — :mod:`muse.core.doc_history` resolves when the symbol first appeared and when it was last changed, mapped to release tags. 4. **Test linkage** — a BFS through the call graph links each production symbol to the test functions that transitively call it. Docstring extraction -------------------- :class:`~muse.plugins.code.ast_parser.SymbolRecord` deliberately does not store docstrings as a dedicated field — they are included in the body hash. Extraction is therefore on-the-fly: the raw source bytes for each file are read once (from the content-addressed object store), the AST is parsed with ``ast.get_docstring``, and the result is cached per ``(file_path, content_hash)`` pair. Health scoring -------------- Every symbol receives a ``doc_health`` score in the range ``[0.0, 1.0]``: +0.30 has a docstring +0.20 docstring ≥ 40 chars (substantive) +0.20 at least one linked test exists +0.15 ``since_version`` was inferred +0.15 docstring is not stale (impl unchanged since last body edit) The ``doc_debt_score`` on :class:`DocSummary` is ``1.0 − caller-weighted average health`` for public symbols, so highly-called, undocumented symbols contribute more debt than leaf utilities with no callers. Security -------- * Source files are read as raw bytes from the SHA-256–verified object store. ``ast.parse`` is used only to extract docstrings — it never executes code. * No subprocess is spawned. No working-tree file is written. * All path lookups use workspace-relative strings validated by the snapshot manifest — no path traversal is possible. Performance ----------- * AST module parses are cached per ``(file_path, content_hash)`` for the duration of one :func:`extract_docs` call, eliminating duplicate parses when multiple symbols share a file. * The ``SymbolCache`` is loaded once and shared across all helpers. * Call-graph BFS is bounded by *depth* (default 3) and snapshot size. * On a 400-file Python codebase a warm-cache full extraction completes in <150 ms. """ from __future__ import annotations import ast import datetime import hashlib import logging import pathlib from collections import deque from typing import Literal, NotRequired, TypedDict from muse.core.doc_history import ( StaleInfo, detect_stale_docstring, get_symbol_version_events, infer_last_changed_version, infer_since_version, ) from muse.core.object_store import read_object from muse.core.validation import MAX_AST_BYTES from muse.core.store import ( Manifest, get_head_commit_id, get_commit_snapshot_manifest, read_commit, ) type SymbolIndex = dict[str, "SymbolRecord"] type NameAddrMap = dict[str, list[str]] type CoverageMap = dict[str, set[str]] from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.plugins.code._callgraph import ( ForwardGraph, ReverseGraph, build_forward_graph, build_reverse_graph, ) from muse.plugins.code._query import symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolKind, SymbolRecord logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Public type definitions # --------------------------------------------------------------------------- DocHealthReason = Literal[ "no_docstring", "docstring_too_short", "no_tests", "no_version_annotation", "stale_impl", ] """Reasons why a symbol's doc health score is below 1.0.""" class SymbolDoc(TypedDict): """Fully-assembled documentation record for one symbol. Every field is populated from committed data only — the working tree is never consulted, making the output reproducible given the same commit. """ address: str """Canonical symbol address, e.g. ``"muse/core/store.py::read_commit"``.""" name: str """Bare symbol name.""" qualified_name: str """Qualified name within the file, e.g. ``"MyClass.my_method"``.""" kind: SymbolKind """Symbol kind: ``"function"``, ``"class"``, ``"method"``, etc.""" file: str """Workspace-relative source file path.""" lineno: int """1-based line number where the symbol definition begins.""" end_lineno: int """1-based line number where the symbol definition ends.""" signature: str """First non-decorator line of the definition (best-effort extraction).""" docstring: str | None """Extracted docstring, or ``None`` when absent.""" callers: list[str] """Symbol addresses that directly call this symbol, sorted lexicographically.""" callees: list[str] """Bare callee names this symbol calls, sorted lexicographically.""" since_commit: str | None """Commit ID in which this symbol first appeared.""" since_version: str | None """Version tag of the first release containing this symbol.""" last_changed_commit: str | None """Commit ID of the most recent modification.""" last_changed_version: str | None """Version tag of the most recent release that modified this symbol.""" breaking_changes: list[str] """Breaking change descriptions collected from the commit history.""" linked_tests: list[str] """Pytest node IDs of test functions that transitively call this symbol.""" doc_health: float """Documentation health score in ``[0.0, 1.0]``.""" doc_health_reasons: list[DocHealthReason] """Why the score is below 1.0 (empty when ``doc_health == 1.0``).""" class MissingDocEntry(TypedDict): """Summary entry for a public symbol that lacks a docstring.""" address: str name: str kind: SymbolKind file: str caller_count: int """Higher values signal higher documentation urgency.""" class StaleDocEntry(TypedDict): """Summary entry for a symbol whose docstring may be out of date.""" address: str name: str kind: SymbolKind file: str last_doc_commit: str | None last_impl_commit: str | None signature_changed: bool body_changed: bool class DocSummary(TypedDict): """Aggregate documentation health metrics for a :class:`DocReport`.""" total_symbols: int public_symbols: int documented: int undocumented: int stale_count: int avg_health: float doc_debt_score: float """``1.0 − caller-weighted average health``. 0.0 = pristine, 1.0 = catastrophic.""" class DocReport(TypedDict): """Complete documentation report for a set of symbols.""" commit_id: str generated_at: str symbols: list[SymbolDoc] missing: list[MissingDocEntry] stale: list[StaleDocEntry] summary: DocSummary # --------------------------------------------------------------------------- # Internal: docstring extraction # --------------------------------------------------------------------------- # Per-invocation cache mapping (file_path, content_hash) → {lineno: str | None} _DocCache = dict[tuple[str, str], dict[int, str | None]] def _build_lineno_docstring_map(source: bytes) -> dict[int, str | None]: """Return ``{lineno: docstring_or_None}`` for every def/class in *source*. Uses Python's ``ast.get_docstring`` — no code is executed. """ if len(source) > MAX_AST_BYTES: return {} try: module = ast.parse(source, type_comments=False) except SyntaxError: return {} result: dict[int, str | None] = {} for node in ast.walk(module): if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef): raw = ast.get_docstring(node, clean=True) result[node.lineno] = raw if raw else None return result def _get_docstring( root: pathlib.Path, file_path: str, lineno: int, content_hash: str, cache: _DocCache, ) -> str | None: """Return the docstring for the symbol at *lineno* in *file_path*. Uses *content_hash* as the cache key so different committed versions of the same path are cached separately. Args: root: Repository root directory. file_path: Workspace-relative file path. lineno: 1-based line number from :class:`SymbolRecord`. content_hash: SHA-256 of the raw file bytes. cache: Mutable per-invocation extraction cache. """ cache_key = (file_path, content_hash) if cache_key not in cache: raw: bytes | None = read_object(root, content_hash) if raw is None: full = root / file_path raw = full.read_bytes() if full.is_file() else b"" cache[cache_key] = _build_lineno_docstring_map(raw) raw_doc = cache[cache_key].get(lineno) if not raw_doc: return None stripped = raw_doc.strip() return stripped if stripped else None # --------------------------------------------------------------------------- # Internal: signature extraction # --------------------------------------------------------------------------- def _extract_signature(source: bytes, lineno: int, end_lineno: int) -> str: """Return the first ``def``/``class``/``async def`` line of a symbol. Reads lines ``[lineno, end_lineno]`` (1-indexed) and returns the first line that starts a definition, stripping leading whitespace. Falls back to an empty string on any error. """ try: text = source.decode("utf-8", errors="replace") lines = text.splitlines() for raw_line in lines[max(0, lineno - 1) : end_lineno]: stripped = raw_line.lstrip() if stripped.startswith(("def ", "async def ", "class ", "@")): return stripped.rstrip() if lineno <= len(lines): return lines[lineno - 1].strip() except Exception: pass return "" # --------------------------------------------------------------------------- # Internal: health scoring # --------------------------------------------------------------------------- def _compute_health( docstring: str | None, linked_tests: list[str], since_version: str | None, stale_info: StaleInfo, ) -> tuple[float, list[DocHealthReason]]: """Return ``(health_score, [reasons])`` for a symbol. Score breakdown: +0.30 has a docstring +0.20 docstring ≥ 40 chars +0.20 has at least one linked test +0.15 ``since_version`` is known +0.15 not stale """ score = 0.0 reasons: list[DocHealthReason] = [] if docstring: score += 0.30 if len(docstring) >= 40: score += 0.20 else: reasons.append("docstring_too_short") else: reasons.append("no_docstring") if linked_tests: score += 0.20 else: reasons.append("no_tests") if since_version is not None: score += 0.15 else: reasons.append("no_version_annotation") if not stale_info["is_stale"]: score += 0.15 else: reasons.append("stale_impl") return min(score, 1.0), reasons # --------------------------------------------------------------------------- # Internal: test linkage via BFS # --------------------------------------------------------------------------- _PY_TEST_PREFIXES: frozenset[str] = frozenset({"test_", "tests/"}) def _is_test_file(file_path: str) -> bool: """Return ``True`` when *file_path* is a test file by convention.""" import posixpath stem = posixpath.basename(file_path) return stem.startswith("test_") or "tests/" in file_path or "/test_" in file_path def _is_test_function(address: str, kind: SymbolKind) -> bool: """Return ``True`` when *address* / *kind* identifies a test function.""" if kind not in ("function", "method", "async_function", "async_method"): return False bare = address.rsplit("::", 1)[-1].rsplit(".", 1)[-1] return bare.startswith("test_") or bare == "test" def build_symbol_test_map( forward_graph: ForwardGraph, all_symbols: SymbolIndex, max_depth: int = 3, ) -> NameAddrMap: """Return a mapping from production symbol address → list[test_address]. Performs a BFS from every test function through the forward call graph (caller → callees by bare name), accumulating which production symbols each test reaches. The result is then inverted. Args: forward_graph: Caller address → frozenset[bare callee name]. all_symbols: All symbols in the snapshot. max_depth: Maximum BFS depth. Default 3. Returns: ``{production_address: [test_address, ...]}``, deduplicated and sorted. """ # Build name → addresses map for reverse lookup (bare name may be ambiguous). name_to_addrs: NameAddrMap = {} for addr, rec in all_symbols.items(): name_to_addrs.setdefault(rec["name"], []).append(addr) # For each test function, BFS through forward graph. # coverage: production_address → set of test addresses that reach it. coverage: CoverageMap = {} for test_addr, rec in all_symbols.items(): if not _is_test_function(test_addr, rec["kind"]): continue file_part = test_addr.split("::")[0] if not _is_test_file(file_part): continue # BFS from the test address using bare names as frontier nodes. visited_names: set[str] = {rec["name"]} q: deque[tuple[str, int]] = deque([(test_addr, 0)]) while q: current_addr, depth = q.popleft() if depth >= max_depth: continue for callee_name in forward_graph.get(current_addr, frozenset()): if callee_name in visited_names: continue visited_names.add(callee_name) for callee_addr in name_to_addrs.get(callee_name, []): if callee_addr == test_addr: continue coverage.setdefault(callee_addr, set()).add(test_addr) q.append((callee_addr, depth + 1)) return { addr: sorted(tests) for addr, tests in coverage.items() } # --------------------------------------------------------------------------- # Internal: callers resolution # --------------------------------------------------------------------------- def _callers_for( address: str, reverse_graph: ReverseGraph, all_symbols: SymbolIndex, ) -> list[str]: """Return addresses that directly call *address*, filtered to known symbols.""" bare = address.rsplit("::", 1)[-1].rsplit(".", 1)[-1] if "::" in address else address raw = reverse_graph.get(bare, []) return sorted(c for c in raw if c in all_symbols) # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def build_symbol_doc( root: pathlib.Path, repo_id: str, address: str, record: SymbolRecord, manifest: Manifest, forward_graph: ForwardGraph, reverse_graph: ReverseGraph, all_symbols: SymbolIndex, linked_tests: list[str], doc_cache: _DocCache, ) -> SymbolDoc: """Assemble a complete :class:`SymbolDoc` for one symbol. Reads the committed source bytes, extracts the docstring and signature, resolves version history and staleness from the index, and computes the health score. Args: root: Repository root directory. repo_id: Repository UUID (for tag lookups). address: Canonical symbol address. record: :class:`SymbolRecord` from the committed snapshot. manifest: Snapshot manifest: ``{file_path: sha256}``. forward_graph: Caller → frozenset[callee_bare_name]. reverse_graph: Callee bare name → [caller_address]. all_symbols: All symbols in the snapshot. linked_tests: Pre-computed test node IDs covering this symbol. doc_cache: Per-invocation docstring extraction cache. """ file_path = address.split("::")[0] if "::" in address else address content_hash = manifest.get(file_path, "") docstring = _get_docstring( root, file_path, record["lineno"], content_hash, doc_cache ) signature = "" if content_hash: raw_bytes: bytes | None = read_object(root, content_hash) if raw_bytes is None: full = root / file_path if full.is_file(): raw_bytes = full.read_bytes() if raw_bytes is not None: signature = _extract_signature( raw_bytes, record["lineno"], record["end_lineno"] ) events = get_symbol_version_events(root, repo_id, address) since_version = infer_since_version(events) last_changed_version = infer_last_changed_version(events) since_commit = events[0]["commit_id"] if events else None last_changed_commit = events[-1]["commit_id"] if events else None breaking_changes: list[str] = [] seen_bc: set[str] = set() for event in events: commit = read_commit(root, event["commit_id"]) if commit is not None: for bc in commit.breaking_changes: if bc not in seen_bc: seen_bc.add(bc) breaking_changes.append(bc) stale_info = detect_stale_docstring(root, address) callers = _callers_for(address, reverse_graph, all_symbols) callees = sorted(forward_graph.get(address, frozenset())) health, reasons = _compute_health(docstring, linked_tests, since_version, stale_info) return SymbolDoc( address=address, name=record["name"], qualified_name=record["qualified_name"], kind=record["kind"], file=file_path, lineno=record["lineno"], end_lineno=record["end_lineno"], signature=signature, docstring=docstring, callers=callers, callees=callees, since_commit=since_commit, since_version=since_version, last_changed_commit=last_changed_commit, last_changed_version=last_changed_version, breaking_changes=breaking_changes, linked_tests=linked_tests, doc_health=round(health, 4), doc_health_reasons=reasons, ) def _is_public(name: str) -> bool: """Return ``True`` when *name* does not begin with ``_``.""" return not name.startswith("_") def extract_docs( root: pathlib.Path, repo_id: str, targets: list[str] | None = None, commit_id: str | None = None, min_health: float | None = None, max_depth: int = 3, ) -> DocReport: """Build a :class:`DocReport` for the repository or a subset of symbols. This is the primary entry point. It loads the symbol cache, builds the call graph, resolves version history for every requested symbol, links tests, and computes health scores — all from committed data. Args: root: Repository root directory. repo_id: Repository UUID. targets: Optional list of symbol addresses or file paths to restrict the report to. ``None`` means document the full snapshot. commit_id: Specific commit to document. ``None`` uses HEAD. min_health: When set, only include symbols with ``doc_health < min_health`` in the output (useful for ``--missing`` / ``--stale`` filter modes). max_depth: Call-graph BFS depth for test-linkage resolution. """ generated_at = datetime.datetime.now(datetime.timezone.utc).isoformat() 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" if commit_id is None: commit_id = get_head_commit_id(root, branch) or "" empty_summary = DocSummary( total_symbols=0, public_symbols=0, documented=0, undocumented=0, stale_count=0, avg_health=0.0, doc_debt_score=1.0, ) if not commit_id: return DocReport( commit_id="", generated_at=generated_at, symbols=[], missing=[], stale=[], summary=empty_summary, ) raw_manifest = get_commit_snapshot_manifest(root, commit_id) if raw_manifest is None: return DocReport( commit_id=commit_id, generated_at=generated_at, symbols=[], missing=[], stale=[], summary=empty_summary, ) manifest: Manifest = raw_manifest cache = load_symbol_cache(root) all_symbols: SymbolIndex = {} for file_tree in symbols_for_snapshot(root, manifest, cache=cache).values(): all_symbols.update(file_tree) forward_graph = build_forward_graph(root, manifest, cache) reverse_graph = build_reverse_graph(root, manifest, cache) # Pre-compute test linkage for all symbols. test_map = build_symbol_test_map(forward_graph, all_symbols, max_depth) # Determine the set of addresses to document. if targets: target_set: set[str] = set() for t in targets: if "::" in t: if t in all_symbols: target_set.add(t) else: prefix = t if t.endswith("/") else t + "::" bare = t for addr in all_symbols: if addr.startswith(prefix) or addr.split("::")[0] == bare: target_set.add(addr) addresses = sorted(target_set) else: addresses = sorted(all_symbols) doc_cache: _DocCache = {} docs: list[SymbolDoc] = [] missing: list[MissingDocEntry] = [] stale_entries: list[StaleDocEntry] = [] for address in addresses: record = all_symbols.get(address) if record is None: continue linked = test_map.get(address, []) doc = build_symbol_doc( root=root, repo_id=repo_id, address=address, record=record, manifest=manifest, forward_graph=forward_graph, reverse_graph=reverse_graph, all_symbols=all_symbols, linked_tests=linked, doc_cache=doc_cache, ) if min_health is not None and doc["doc_health"] >= min_health: continue docs.append(doc) if _is_public(record["name"]) and doc["docstring"] is None: missing.append( MissingDocEntry( address=address, name=record["name"], kind=record["kind"], file=doc["file"], caller_count=len(doc["callers"]), ) ) stale_info = detect_stale_docstring(root, address) if stale_info["is_stale"]: stale_entries.append( StaleDocEntry( address=address, name=record["name"], kind=record["kind"], file=doc["file"], last_doc_commit=stale_info["last_doc_commit"], last_impl_commit=stale_info["last_impl_commit"], signature_changed=stale_info["signature_changed"], body_changed=stale_info["body_changed"], ) ) missing.sort(key=lambda e: e["caller_count"], reverse=True) total = len(docs) public_count = sum(1 for d in docs if _is_public(d["name"])) documented = sum(1 for d in docs if d["docstring"] is not None) undocumented = sum( 1 for d in docs if _is_public(d["name"]) and d["docstring"] is None ) stale_count = len(stale_entries) avg_health = sum(d["doc_health"] for d in docs) / total if total else 0.0 debt_total = 0.0 debt_weight = 0.0 for d in docs: if _is_public(d["name"]): w = float(len(d["callers"]) + 1) debt_total += (1.0 - d["doc_health"]) * w debt_weight += w doc_debt_score = (debt_total / debt_weight) if debt_weight > 0 else 0.0 return DocReport( commit_id=commit_id, generated_at=generated_at, symbols=docs, missing=missing, stale=stale_entries, summary=DocSummary( total_symbols=total, public_symbols=public_count, documented=documented, undocumented=undocumented, stale_count=stale_count, avg_health=round(avg_health, 4), doc_debt_score=round(doc_debt_score, 4), ), )