"""AST parsing and symbol extraction for the code domain plugin. This module provides the :class:`LanguageAdapter` protocol and concrete adapters for parsing source files into :type:`SymbolTree` structures. Language support matrix ----------------------- - **Python** (``*.py``, ``*.pyi``): Full AST-based extraction using the stdlib :mod:`ast` module. Content IDs are hashes of normalized (unparsed) AST text — insensitive to whitespace, comments, and formatting. - **JavaScript / TypeScript** (``*.js``, ``*.jsx``, ``*.mjs``, ``*.cjs``, ``*.ts``, ``*.tsx``): tree-sitter based. Async functions, arrow functions bound to ``const``/``let``, and module-level variables are all extracted. - **Go** (``*.go``): tree-sitter based. Method qualified names carry the receiver type (e.g. ``Dog.Bark``). Package-level ``const``/``var`` included. - **Rust** (``*.rs``): tree-sitter based. Functions inside ``impl`` blocks are qualified with the implementing type (e.g. ``Dog.bark``). ``static``, ``const``, type aliases, and ``mod`` declarations are extracted. - **Java** (``*.java``), **C#** (``*.cs``): tree-sitter based. - **C** (``*.c``, ``*.h``), **C++** (``*.cpp``, ``*.cc``, ``*.cxx``, ``*.hpp``, ``*.hxx``): tree-sitter based. Structs and enums extracted. - **Ruby** (``*.rb``), **Kotlin** (``*.kt``, ``*.kts``): tree-sitter based. - **Swift** (``*.swift``): tree-sitter based; requires ``py-tree-sitter-swift`` (degrades to file-level tracking if the package is unavailable). - **Markdown** (``*.md``, ``*.rst``, ``*.txt``): ATX headings extracted as ``section`` symbols; requires ``tree-sitter-markdown``. - **CSS** (``*.css``): rule-sets, keyframes, @media, @supports, and @layer extracted; requires ``tree-sitter-css``. - **SCSS** (``*.scss``): superset of CSS symbols plus ``$variables``, ``@mixin``, and ``@function``; requires ``tree-sitter-scss``. - **HTML** (``*.html``, ``*.htm``): semantic elements and id-bearing elements extracted; requires ``tree-sitter-html``. - **TOML** (``*.toml``): stdlib :mod:`tomllib` based (Python 3.11+, zero extra deps). Tables (``[section]``) and array-of-tables entries (``[[section]]``) become ``section`` symbols; scalar key-value pairs become ``variable`` symbols. Content IDs are computed from canonical JSON (sorted keys, ISO date strings) — insensitive to comments and key ordering. Symbol addresses ---------------- Every extracted symbol is stored in the :type:`SymbolTree` dict under a stable *address* key of the form:: "::" Nested symbols (class methods) use dotted qualified names:: "src/models.py::User.save" "src/models.py::User.__init__" Top-level symbols:: "src/utils.py::calculate_total" "src/utils.py::import::pathlib" Content IDs and rename / move detection ---------------------------------------- Each :class:`SymbolRecord` carries three hashes: ``content_id`` SHA-256 of the full normalized AST of the symbol (includes name, signature, and body). Two symbols are "the same thing" when their ``content_id`` matches — regardless of where in the repo they live. ``body_hash`` SHA-256 of the normalized body statements only (excludes the ``def`` line). Used to detect *renames*: same body, different name. ``signature_id`` SHA-256 of ``"name(args) -> return"``. Used to detect *implementation- only changes*: signature unchanged, body changed. Extending --------- Implement :class:`LanguageAdapter` and append an instance to :data:`ADAPTERS`. The adapter is selected by the file's suffix, with the first matching adapter taking priority. """ from __future__ import annotations from muse.core.validation import MAX_AST_BYTES import ast import datetime import hashlib import importlib import json import logging import pathlib import re import sys import tomllib import types as _types from typing import TYPE_CHECKING, Literal, Protocol, TypedDict, runtime_checkable if TYPE_CHECKING: from tree_sitter import Language, Node, Parser, Query, QueryCursor logger = logging.getLogger(__name__) type _IntMap = dict[str, int] type _KindMap = dict[str, SymbolKind] type _TomlMapping = dict[str, _TomlValue] # --------------------------------------------------------------------------- # Symbol record types # --------------------------------------------------------------------------- SymbolKind = Literal[ "function", "async_function", "class", "method", "async_method", "variable", "import", "section", # Markdown ATX/setext heading; HTML semantic element "rule", # CSS rule-set, @keyframes, @media, @supports, @layer (keyed by selector/query) "mixin", # SCSS @mixin — outputs CSS properties, can be @include'd "interface", # TypeScript interface, Java interface, C# interface, Go interface type "type_alias", # TypeScript `type Foo = …`, Go plain type alias (non-struct/interface) "enum", # TypeScript/Java/C#/Rust/C/C++ enum "struct", # Rust struct, C/C++ struct, C# struct, Go struct type "trait", # Rust trait, Swift protocol "namespace", # C++ namespace, TypeScript namespace/module "module", # Ruby module, Rust mod "object", # Kotlin singleton object declaration (`object Foo {}`) ] class SymbolRecord(TypedDict): """Content-addressed record for a single named symbol in source code. Hash dimensions (v2) -------------------- ``content_id`` SHA-256 of the full normalized symbol (name + signature + body). ``body_hash`` SHA-256 of body statements only. Same body, different name → rename. ``signature_id`` SHA-256 of ``"name(args) -> return"``. Stable across body changes. ``metadata_id`` SHA-256 of metadata that wraps the symbol but is not part of its body: decorators, ``async`` flag, class bases, visibility modifiers (where extractable by the language adapter). Empty string for legacy records or adapters that do not support metadata extraction. ``canonical_key`` Stable machine handle: ``{file}#{scope_path}#{kind}#{name}#{lineno}``. Disambiguates overloads and nested scopes. Unique within a snapshot. """ kind: SymbolKind name: str qualified_name: str # "ClassName.method" for nested; flat name for top-level content_id: str # SHA-256 of full normalized AST (name + signature + body) body_hash: str # SHA-256 of body stmts only — for rename detection signature_id: str # SHA-256 of "name(args)->return" — for impl-only changes metadata_id: str # SHA-256 of decorator/async/bases metadata (v2; "" = pre-v2) canonical_key: str # {file}#{scope}#{kind}#{name}#{lineno} — stable handle lineno: int end_lineno: int #: Flat map from symbol address to :class:`SymbolRecord`. #: Nested symbols (methods) appear at their qualified address alongside the #: parent class. SymbolTree = dict[str, SymbolRecord] # --------------------------------------------------------------------------- # Language adapter protocol # --------------------------------------------------------------------------- @runtime_checkable class LanguageAdapter(Protocol): """Protocol every language adapter must implement. Adapters are stateless. The same instance may be called concurrently for different files without synchronization. """ def supported_extensions(self) -> frozenset[str]: """Return the set of lowercase file suffixes this adapter handles.""" ... def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: """Extract the symbol tree from raw source bytes. Args: source: Raw bytes of the source file. file_path: Workspace-relative POSIX path — used to build the symbol address prefix. Returns: A :type:`SymbolTree` mapping symbol addresses to :class:`SymbolRecord` dicts. Returns an empty dict on parse errors so that the caller can fall through to file-level ops. """ ... def file_content_id(self, source: bytes) -> str: """Return a stable content identifier for the whole file. For AST-capable adapters: hash of the normalized (unparsed) module AST — insensitive to formatting and comments. For non-AST adapters: SHA-256 of raw bytes. Args: source: Raw bytes of the file. Returns: Hex-encoded SHA-256 digest. """ ... # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _sha256(text: str) -> str: return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() def _sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() # --------------------------------------------------------------------------- # Python adapter # --------------------------------------------------------------------------- class PythonAdapter: """Python language adapter — AST-based, zero external dependencies. Uses :func:`ast.parse` for parsing and :func:`ast.unparse` for normalization. The result is a deterministic, whitespace-insensitive representation that strips comments and normalizes indentation. ``ast.unparse`` is available since Python 3.9; Muse requires 3.12. """ def supported_extensions(self) -> frozenset[str]: return frozenset({".py", ".pyi"}) def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: if len(source) > MAX_AST_BYTES: logger.debug( "ast_parser: skipping %s (%d bytes exceeds MAX_AST_BYTES=%d)", file_path, len(source), MAX_AST_BYTES, ) return {} try: tree = ast.parse(source, filename=file_path) except SyntaxError: return {} symbols: SymbolTree = {} _extract_stmts(tree.body, file_path, "", symbols) return symbols def file_content_id(self, source: bytes) -> str: if len(source) > MAX_AST_BYTES: return _sha256_bytes(source) try: tree = ast.parse(source) return _sha256(ast.unparse(tree)) except SyntaxError: return _sha256_bytes(source) # --------------------------------------------------------------------------- # AST extraction helpers (module-level so they can be tested independently) # --------------------------------------------------------------------------- def _extract_stmts( stmts: list[ast.stmt], file_path: str, class_prefix: str, out: SymbolTree, ) -> None: """Recursively walk *stmts* and populate *out* with symbol records. Args: stmts: Statement list from an :class:`ast.Module` or :class:`ast.ClassDef` body. file_path: Workspace-relative POSIX path — used as address prefix. class_prefix: Dotted class path for methods (e.g. ``"MyClass."``). Empty string at top-level. out: Accumulator — modified in place. """ for node in stmts: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): is_async = isinstance(node, ast.AsyncFunctionDef) if class_prefix: kind: SymbolKind = "async_method" if is_async else "method" else: kind = "async_function" if is_async else "function" qualified = f"{class_prefix}{node.name}" addr = f"{file_path}::{qualified}" out[addr] = _make_func_record( node, node.name, qualified, kind, file_path=file_path, class_prefix=class_prefix, ) elif isinstance(node, ast.ClassDef): qualified = f"{class_prefix}{node.name}" addr = f"{file_path}::{qualified}" out[addr] = _make_class_record(node, qualified, file_path=file_path) _extract_stmts(node.body, file_path, f"{qualified}.", out) elif isinstance(node, (ast.Assign, ast.AnnAssign)) and not class_prefix: # Only top-level assignments — class-level attributes are captured # as part of the parent class's content_id. for name in _assignment_names(node): addr = f"{file_path}::{name}" out[addr] = _make_var_record(node, name, file_path=file_path) elif isinstance(node, ast.TypeAlias) and not class_prefix: # Python 3.12+ ``type X = ...`` soft-keyword statement. # Treated as a named variable so the symbol appears in the graph # and import-stale checks resolve correctly. if isinstance(node.name, ast.Name): name = node.name.id addr = f"{file_path}::{name}" out[addr] = _make_var_record(node, name, file_path=file_path) elif isinstance(node, (ast.Import, ast.ImportFrom)) and not class_prefix: for name, original in _import_names(node): addr = f"{file_path}::import::{name}" out[addr] = _make_import_record( node, name, file_path=file_path, original_name=original ) def _compute_metadata_id_func(node: ast.FunctionDef | ast.AsyncFunctionDef) -> str: """SHA-256 of Python function metadata: decorators + async flag.""" dec_src = " ".join(ast.unparse(d) for d in node.decorator_list) async_flag = "async" if isinstance(node, ast.AsyncFunctionDef) else "sync" return _sha256(f"{async_flag}:{dec_src}") def _compute_metadata_id_class(node: ast.ClassDef) -> str: """SHA-256 of Python class metadata: decorators + bases.""" dec_src = " ".join(ast.unparse(d) for d in node.decorator_list) base_src = ", ".join(ast.unparse(b) for b in node.bases) return _sha256(f"{dec_src}:{base_src}") def _canonical_key( file_path: str, scope: str, kind: str, name: str, lineno: int ) -> str: """Return the canonical machine handle for a symbol. Format: ``{file}#{scope}#{kind}#{name}#{lineno}`` ``scope`` is the dotted class prefix (e.g. ``User.``) or empty for top-level symbols. This key is unique within a snapshot and stable across renames (by lineno), though lineno drift after edits is expected. """ return f"{file_path}#{scope}#{kind}#{name}#{lineno}" def _make_func_record( node: ast.FunctionDef | ast.AsyncFunctionDef, name: str, qualified_name: str, kind: SymbolKind, file_path: str = "", class_prefix: str = "", ) -> SymbolRecord: full_src = ast.unparse(node) body_src = "\n".join(ast.unparse(s) for s in node.body) args_src = ast.unparse(node.args) ret_src = ast.unparse(node.returns) if node.returns else "" return SymbolRecord( kind=kind, name=name, qualified_name=qualified_name, content_id=_sha256(full_src), body_hash=_sha256(body_src), signature_id=_sha256(f"{name}({args_src})->{ret_src}"), metadata_id=_compute_metadata_id_func(node), canonical_key=_canonical_key(file_path, class_prefix, kind, name, node.lineno), lineno=node.lineno, end_lineno=node.end_lineno or node.lineno, ) def _make_class_record( node: ast.ClassDef, qualified_name: str, file_path: str = "", ) -> SymbolRecord: full_src = ast.unparse(node) base_src = ", ".join(ast.unparse(b) for b in node.bases) if node.bases else "" # Body hash captures class structure (bases + method names) but NOT method # bodies — those change independently and have their own records. method_names = sorted( n.name for n in node.body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) ) structure = f"class {node.name}({base_src}):{method_names}" header = f"class {node.name}({base_src})" if node.bases else f"class {node.name}" return SymbolRecord( kind="class", name=node.name, qualified_name=qualified_name, content_id=_sha256(full_src), body_hash=_sha256(structure), signature_id=_sha256(header), metadata_id=_compute_metadata_id_class(node), canonical_key=_canonical_key(file_path, "", "class", node.name, node.lineno), lineno=node.lineno, end_lineno=node.end_lineno or node.lineno, ) def _make_var_record( node: ast.Assign | ast.AnnAssign | ast.TypeAlias, name: str, file_path: str = "", ) -> SymbolRecord: normalized = ast.unparse(node) return SymbolRecord( kind="variable", name=name, qualified_name=name, content_id=_sha256(normalized), body_hash=_sha256(normalized), signature_id=_sha256(name), metadata_id="", canonical_key=_canonical_key(file_path, "", "variable", name, node.lineno), lineno=node.lineno, end_lineno=node.end_lineno or node.lineno, ) def _make_import_record( node: ast.Import | ast.ImportFrom, name: str, file_path: str = "", original_name: str = "", ) -> SymbolRecord: normalized = ast.unparse(node) # Encode the source module so downstream consumers (e.g. the breakage # checker) can distinguish muse-internal imports from stdlib / third-party # imports without re-parsing. # # Format: # "import::::" — from import [as ] # "import::" — import [as ] # # ``original_name`` is the name as written in the source (before any ``as`` # alias). ``name`` is the local alias (equal to original when there is no # ``as`` clause). Storing the original in qualified_name lets the breakage # checker resolve the real module path even when an alias is used. orig = original_name or name if isinstance(node, ast.ImportFrom) and node.module: qualified = f"import::{node.module}::{orig}" else: qualified = f"import::{orig}" return SymbolRecord( kind="import", name=name, qualified_name=qualified, content_id=_sha256(normalized), body_hash=_sha256(normalized), signature_id=_sha256(name), metadata_id="", canonical_key=_canonical_key(file_path, "", "import", name, node.lineno), lineno=node.lineno, end_lineno=node.lineno, ) def _assignment_names(node: ast.Assign | ast.AnnAssign) -> list[str]: if isinstance(node, ast.Assign): return [t.id for t in node.targets if isinstance(t, ast.Name)] if isinstance(node.target, ast.Name): return [node.target.id] return [] def _import_names(node: ast.Import | ast.ImportFrom) -> list[tuple[str, str]]: """Return ``(alias, original)`` pairs for each name in the import statement. ``alias`` is the local binding name (what the code uses after import). ``original`` is the name as written in the source before any ``as`` clause. When there is no alias, both are the same string. """ if isinstance(node, ast.Import): return [(a.asname or a.name, a.name) for a in node.names] # ImportFrom if node.names and node.names[0].name == "*": wildcard = f"*:{node.module or '?'}" return [(wildcard, wildcard)] return [(a.asname or a.name, a.name) for a in node.names] # --------------------------------------------------------------------------- # Fallback adapter — file-level identity only, no symbol extraction # --------------------------------------------------------------------------- class FallbackAdapter: """Fallback adapter for languages without a dedicated AST parser. Returns an empty :type:`SymbolTree` (file-level tracking only) and uses raw-bytes SHA-256 as the file content ID. """ def __init__(self, extensions: frozenset[str]) -> None: self._extensions = extensions def supported_extensions(self) -> frozenset[str]: return self._extensions def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: # noqa: ARG002 return {} def file_content_id(self, source: bytes) -> str: return _sha256_bytes(source) # --------------------------------------------------------------------------- # Markdown adapter — ATX heading extraction via tree-sitter-markdown # --------------------------------------------------------------------------- # Maps ATX heading marker node types to their integer level (1–6). _MD_HEADING_LEVEL: _IntMap = { "atx_h1_marker": 1, "atx_h2_marker": 2, "atx_h3_marker": 3, "atx_h4_marker": 4, "atx_h5_marker": 5, "atx_h6_marker": 6, } #: Maximum characters kept from heading text when building symbol addresses. #: Guards against pathological documents with enormously long headings. _MD_MAX_HEADING_LEN: int = 120 #: Compiled pattern that strips common inline Markdown markup from heading #: text so that minor formatting changes (adding/removing bold markers) do #: not alter the section's symbol address. #: #: Groups (in order): link text, reference-link text, bold/italic *-text, #: bold/italic _-text, inline-code text. Images are consumed with no group #: so they collapse to an empty string. Each text group is bounded to 200 #: characters to prevent catastrophic backtracking on adversarial input. _MD_INLINE_RE: re.Pattern[str] = re.compile( r"!\[[^\]]{0,200}\]\([^)]{0,500}\)" # ![alt](url) — drop r"|!\[[^\]]{0,200}\]\[[^\]]{0,200}\]" # ![alt][ref] — drop r"|\[([^\]]{0,200})\]\([^)]{0,500}\)" # [text](url) — keep text r"|\[([^\]]{0,200})\]\[[^\]]{0,200}\]" # [text][ref] — keep text r"|\*{1,3}([^*]{0,200})\*{1,3}" # *em* / **bold** — keep text r"|_{1,3}([^_]{0,200})_{1,3}" # _em_ / __bold__ — keep text r"|`+([^`]{0,200})`+" # `code` — keep text ) def _plain_heading(raw: str) -> str: """Reduce raw Markdown inline text to a plain, address-safe heading. Strips image tags, extracts link text from ``[text](url)`` syntax, and removes bold/italic/inline-code markers while keeping their inner text. The result is whitespace-collapsed and truncated to ``_MD_MAX_HEADING_LEN`` characters. The output is used as the symbol ``name`` and as a component of the ``qualified_name`` address, so it must be stable across minor formatting changes (e.g. wrapping a word in bold markers). Args: raw: Raw inline text from an ``atx_heading`` ``inline`` child node. Returns: Plain-text heading, collapsed and truncated. """ def _keep_text(m: re.Match[str]) -> str: # Return the first non-None captured group (the visible text). # Images have no group, so all groups are None → returns "". for g in m.groups(): if g is not None: return g return "" text = _MD_INLINE_RE.sub(_keep_text, raw) # Decode the five most common HTML entities that appear in headings. text = ( text.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace(""", '"') .replace("'", "'") ) return " ".join(text.split())[: _MD_MAX_HEADING_LEN] class MarkdownAdapter: """Semantic symbol extraction from Markdown (GFM) source files. Extracts three categories of independently addressable symbols from Markdown documents, enabling section-level and element-level diff precision far beyond file-level tracking: **Sections** (``section`` kind) Each ATX heading (``#``, ``##``, …) creates a ``section`` symbol. Symbols are nested hierarchically so that ``## Installation`` under ``# API Reference`` gets the qualified name ``API Reference.Installation``, preventing address collisions between identically-named headings in different contexts. ``content_id`` SHA-256 of the *full* section bytes — the heading line plus all content (paragraphs, code blocks, tables, and nested subsections) up to the next heading at the same or higher level. Two branches that each modify a *different* section will have non-overlapping ``content_id`` changes and auto-merge. Two branches that modify the *same* section conflict at section granularity rather than file granularity. ``body_hash`` SHA-256 of content *below* the heading line only (bytes from the end of the heading node to the end of the section node). A shared ``body_hash`` with a different ``signature_id`` signals a heading retitle — the section was renamed but its content is unchanged. ``signature_id`` SHA-256 of ``"h{level}:{plain_text}"``. Encodes both the heading level and the plain text, so promoting ``##`` to ``#`` is independently detectable from content changes. ``metadata_id`` SHA-256 of ``"level={N}"``. The level encoded separately so that a level-only change (``###`` → ``##`` with identical text) produces a ``metadata_id`` change but identical ``body_hash``. **Fenced code blocks** (``variable`` kind) Each ````` ```lang ... ``` ````` block is emitted as a ``variable`` symbol scoped to its containing section (or the document root when not inside any section). The symbol name encodes the language tag and start line: ``code[python]@L15``. ``content_id`` / ``body_hash`` SHA-256 of the code fence *content* only (whitespace-normalised), excluding delimiters and the language tag. Trivial reformatting (trailing spaces, final newline) produces no diff. ``signature_id`` SHA-256 of the language tag alone — stable across content changes, signals a language-tag change (e.g. ``python`` → ``python3``). **GFM pipe tables** (``section`` kind) Each ``| col | col |`` table is emitted as a ``section`` symbol scoped to its containing section. Name: ``table@L{start_line}``. ``content_id`` SHA-256 of the full table bytes (header + delimiter + data rows). ``body_hash`` SHA-256 of data rows only — adding a new column header is detectable independently from adding a new data row. ``signature_id`` SHA-256 of ``"|".join(column_names)`` — the table schema. Heading address stability ------------------------- Inline markup is stripped before building symbol addresses: bold, italic, inline-code, and link markup are removed while keeping their visible text. This makes the section address stable across formatting changes (``# **Setup**`` and ``# Setup`` produce the same address). Collision handling ------------------ When two headings at the same nesting level share the same plain text, the second heading gets ``@L{lineno}`` appended to its qualified name (e.g. ``Examples@L42``). This is stable within a commit because line numbers do not change unless the document is edited. Scope and limitations --------------------- - Only ATX-style headings create section symbols. Setext headings (underline-style) are not extracted. - ``.rst`` and ``.txt`` extensions are registered for backward compatibility; semantic extraction is only meaningful for ``.md`` files since RST uses different heading conventions. - Requires ``tree-sitter-markdown``; degrades to an empty symbol tree (file-level tracking only) when the grammar package is unavailable. - Maximum section nesting depth is ``_MAX_DEPTH`` (default 8). - Language tags in fenced code blocks are truncated to 40 characters to prevent unbounded symbol names. """ _EXTENSIONS: frozenset[str] = frozenset({".md", ".rst", ".txt"}) _MAX_DEPTH: int = 8 def __init__(self) -> None: self._parser: Parser | None = None try: from tree_sitter import Language, Parser import tree_sitter_markdown as _md lang = Language(_md.language()) self._parser = Parser(lang) except Exception as exc: # noqa: BLE001 logger.debug( "tree-sitter-markdown unavailable — Markdown file-level only: %s", exc ) def supported_extensions(self) -> frozenset[str]: return self._EXTENSIONS def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: """Extract section, code-block, and table symbols from *source*. When ``tree-sitter-markdown`` is available, a full CST walk extracts sections, fenced code blocks, and GFM tables. When the grammar package is absent the method falls back to a pure-Python ATX heading scan that extracts ``section`` symbols from lines matching ``^#{1,6} text``. The fallback produces the same address/hash scheme as the tree-sitter path so that diffs are accurate whenever the optional dependency is missing. Args: source: Raw bytes of the Markdown file. file_path: Workspace-relative POSIX path — used as address prefix. Returns: A :type:`SymbolTree` mapping qualified addresses to :class:`SymbolRecord` dicts. """ if self._parser is None: return self._parse_headings_regex(source, file_path) try: tree = self._parser.parse(source) except Exception as exc: # noqa: BLE001 logger.debug("Markdown parse error in %s: %s", file_path, exc) return self._parse_headings_regex(source, file_path) symbols: SymbolTree = {} self._walk_children(tree.root_node, source, file_path, "", symbols, 0) return symbols # ATX heading pattern: optional leading spaces (up to 3), 1–6 '#', space, text. _ATX_RE: re.Pattern[str] = re.compile(r"^ {0,3}(#{1,6})\s+(.*?)(?:\s+#+\s*)?$") def _parse_headings_regex(self, source: bytes, file_path: str) -> SymbolTree: """Pure-Python ATX heading extraction — fallback when tree-sitter is absent. Scans *source* line-by-line for ATX headings (``# Heading``) and emits a ``section`` symbol for each one, using the same address and hash scheme as the tree-sitter path. Fenced code blocks are skipped so that ``#`` inside a code fence is not misidentified as a heading. Args: source: Raw bytes of the Markdown file. file_path: Workspace-relative POSIX path — used as address prefix. Returns: A :type:`SymbolTree` with one ``section`` entry per ATX heading. """ text = source.decode("utf-8", errors="replace") lines = text.splitlines() symbols: SymbolTree = {} # Track heading stack to build hierarchical qualified names. # Each slot: (level, qualified_name). stack: list[tuple[int, str]] = [] in_fence = False fence_marker = "" # Collect (lineno, level, plain_text, section_start) tuples first so # we can compute end_lineno from the next heading at the same/higher level. heading_spans: list[tuple[int, int, str]] = [] for i, line in enumerate(lines, start=1): stripped = line.strip() # Detect fenced code block open/close. if not in_fence: if stripped.startswith("```") or stripped.startswith("~~~"): in_fence = True fence_marker = stripped[:3] continue else: if stripped.startswith(fence_marker): in_fence = False continue m = self._ATX_RE.match(line) if m: level = len(m.group(1)) plain_text = _plain_heading(m.group(2).strip()) if plain_text: heading_spans.append((i, level, plain_text)) # Build symbols from collected spans. for idx, (lineno, level, plain_text) in enumerate(heading_spans): # Compute end_lineno: line before next heading at same/higher level, # or end of file. end_lineno = len(lines) for j in range(idx + 1, len(heading_spans)): if heading_spans[j][1] <= level: end_lineno = heading_spans[j][0] - 1 break # Build qualified name from the heading stack. while stack and stack[-1][0] >= level: stack.pop() prefix = stack[-1][1] if stack else "" qualified = f"{prefix}.{plain_text}" if prefix else plain_text stack.append((level, qualified)) addr = f"{file_path}::{qualified}" if addr in symbols: qualified = f"{qualified}@L{lineno}" addr = f"{file_path}::{qualified}" stack[-1] = (level, qualified) # Compute hashes over the section's raw bytes. section_lines = lines[lineno - 1 : end_lineno] section_bytes = "\n".join(section_lines).encode("utf-8", errors="replace") body_lines = lines[lineno : end_lineno] body_bytes = "\n".join(body_lines).encode("utf-8", errors="replace") symbols[addr] = SymbolRecord( kind="section", name=plain_text, qualified_name=qualified, content_id=_sha256_bytes(_norm_ws(section_bytes)), body_hash=( _sha256_bytes(_norm_ws(body_bytes)) if body_bytes.strip() else _sha256("") ), signature_id=_sha256(f"h{level}:{plain_text}"), metadata_id=_sha256(f"level={level}"), canonical_key=_canonical_key( file_path, prefix, "section", plain_text, lineno ), lineno=lineno, end_lineno=end_lineno, ) return symbols # ------------------------------------------------------------------ # Tree walkers # ------------------------------------------------------------------ def _walk_children( self, node: Node, src: bytes, file_path: str, prefix: str, out: SymbolTree, depth: int, ) -> None: """Walk direct children of *node*, dispatching to per-type emitters. Processes ``section``, ``fenced_code_block``, and ``pipe_table`` nodes. All other node types (paragraphs, lists, block quotes, thematic breaks) are left as implicit content inside their ancestor section's ``content_id`` and are not emitted as independent symbols. Args: node: CST node whose children to walk. src: Raw source bytes. file_path: Workspace-relative POSIX path. prefix: Dotted qualified-name accumulated so far (empty at root). out: Symbol accumulator — modified in place. depth: Current nesting depth; stops at ``_MAX_DEPTH``. """ if depth > self._MAX_DEPTH: return for child in node.children: ctype = child.type if ctype == "section": self._emit_section(child, src, file_path, prefix, out, depth) elif ctype == "fenced_code_block": self._emit_code_block(child, src, file_path, prefix, out) elif ctype == "pipe_table": self._emit_table(child, src, file_path, prefix, out) def _emit_section( self, node: Node, src: bytes, file_path: str, prefix: str, out: SymbolTree, depth: int, ) -> None: """Emit a ``section`` symbol for *node* and recurse into its children. Builds a hierarchical qualified name and de-duplicates address collisions by appending ``@L{lineno}``. The ``content_id`` hashes the full section node bytes so that any change to content beneath the heading — paragraphs, code, nested subsections — is detected. Args: node: ``section`` CST node (heading + all content beneath it). src: Raw source bytes. file_path: Workspace-relative POSIX path. prefix: Parent qualified-name path (empty at document root). out: Symbol accumulator — modified in place. depth: Current nesting depth. """ heading_node = next( (c for c in node.children if c.type == "atx_heading"), None ) if heading_node is None: # Setext or unrecognised heading style — skip symbol, still recurse. self._walk_children(node, src, file_path, prefix, out, depth + 1) return level, plain_text = self._extract_heading(heading_node, src) if not plain_text: self._walk_children(node, src, file_path, prefix, out, depth + 1) return qualified = f"{prefix}.{plain_text}" if prefix else plain_text lineno = heading_node.start_point[0] + 1 addr = f"{file_path}::{qualified}" # De-duplicate: two headings with identical plain text at the same # level within the same parent get the second one's line number # appended so addresses remain unique within the snapshot. if addr in out: qualified = f"{qualified}@L{lineno}" addr = f"{file_path}::{qualified}" section_bytes = _node_text(src, node) # body_hash covers everything below the heading line. body_bytes = src[heading_node.end_byte : node.end_byte] out[addr] = SymbolRecord( kind="section", name=plain_text, qualified_name=qualified, content_id=_sha256_bytes(_norm_ws(section_bytes)), body_hash=( _sha256_bytes(_norm_ws(body_bytes)) if body_bytes.strip() else _sha256("") ), signature_id=_sha256(f"h{level}:{plain_text}"), metadata_id=_sha256(f"level={level}"), canonical_key=_canonical_key( file_path, prefix, "section", plain_text, lineno ), lineno=lineno, end_lineno=node.end_point[0] + 1, ) self._walk_children(node, src, file_path, qualified, out, depth + 1) def _emit_code_block( self, node: Node, src: bytes, file_path: str, prefix: str, out: SymbolTree, ) -> None: """Emit a ``variable`` symbol for a fenced code block. The symbol name encodes the language tag and start line so that multiple code blocks in the same section are independently addressable. ``content_id`` hashes only the fence content (not the delimiter lines or language tag) so changing the language tag appears in ``signature_id`` but not ``body_hash``. Args: node: ``fenced_code_block`` CST node. src: Raw source bytes. file_path: Workspace-relative POSIX path. prefix: Containing section's qualified name (empty at root). out: Symbol accumulator — modified in place. """ lineno = node.start_point[0] + 1 lang = self._extract_code_lang(node, src) lang_suffix = f"[{lang}]" if lang else "" name = f"code{lang_suffix}@L{lineno}" qualified = f"{prefix}.{name}" if prefix else name addr = f"{file_path}::{qualified}" code_bytes = self._extract_code_content(node, src) normalised = _norm_ws(code_bytes) out[addr] = SymbolRecord( kind="variable", name=name, qualified_name=qualified, content_id=_sha256_bytes(normalised), body_hash=_sha256_bytes(normalised), signature_id=_sha256(lang), metadata_id=_sha256(lang), canonical_key=_canonical_key( file_path, prefix, "variable", name, lineno ), lineno=lineno, end_lineno=node.end_point[0] + 1, ) def _emit_table( self, node: Node, src: bytes, file_path: str, prefix: str, out: SymbolTree, ) -> None: """Emit a ``section`` symbol for a GFM pipe table. ``signature_id`` encodes the column headers so that schema changes are detectable independently of row data changes. ``body_hash`` covers data rows only, enabling a column header rename to be distinguished from adding a new row. Args: node: ``pipe_table`` CST node. src: Raw source bytes. file_path: Workspace-relative POSIX path. prefix: Containing section's qualified name (empty at root). out: Symbol accumulator — modified in place. """ lineno = node.start_point[0] + 1 name = f"table@L{lineno}" qualified = f"{prefix}.{name}" if prefix else name addr = f"{file_path}::{qualified}" headers = self._extract_table_headers(node, src) header_sig = "|".join(headers) table_bytes = _node_text(src, node) data_bytes = self._extract_table_data_bytes(node, src) out[addr] = SymbolRecord( kind="section", name=name, qualified_name=qualified, content_id=_sha256_bytes(_norm_ws(table_bytes)), body_hash=( _sha256_bytes(_norm_ws(data_bytes)) if data_bytes else _sha256("") ), signature_id=_sha256(header_sig), metadata_id=_sha256(header_sig), canonical_key=_canonical_key( file_path, prefix, "section", name, lineno ), lineno=lineno, end_lineno=node.end_point[0] + 1, ) # ------------------------------------------------------------------ # Heading extraction # ------------------------------------------------------------------ def _extract_heading(self, node: Node, src: bytes) -> tuple[int, str]: """Return ``(level, plain_text)`` for an ``atx_heading`` node. ``level`` is 1–6. ``plain_text`` has inline markup stripped, whitespace collapsed, and is truncated to ``_MD_MAX_HEADING_LEN`` characters. Returns ``(0, "")`` when the heading cannot be parsed. Args: node: ``atx_heading`` CST node. src: Raw source bytes. Returns: Tuple of (heading level, sanitised plain text). """ level = 0 raw_text = "" for child in node.children: if child.type in _MD_HEADING_LEVEL: level = _MD_HEADING_LEVEL[child.type] elif child.type == "inline": raw_text = ( _node_text(src, child) .decode("utf-8", errors="replace") .strip() ) return level, _plain_heading(raw_text) # ------------------------------------------------------------------ # Code block extraction # ------------------------------------------------------------------ def _extract_code_lang(self, node: Node, src: bytes) -> str: """Return the language tag from a fenced code block, or ``""``.""" for child in node.children: if child.type == "info_string": for lang_node in child.children: if lang_node.type == "language": return ( _node_text(src, lang_node) .decode("utf-8", errors="replace") .strip() .lower()[:40] # guard against unbounded lang tags ) return "" def _extract_code_content(self, node: Node, src: bytes) -> bytes: """Return the raw bytes of the code fence content (excluding delimiters).""" for child in node.children: if child.type == "code_fence_content": return _node_text(src, child) return b"" # ------------------------------------------------------------------ # Table extraction # ------------------------------------------------------------------ def _extract_table_headers(self, node: Node, src: bytes) -> list[str]: """Return the column header strings from a pipe table's header row.""" for child in node.children: if child.type == "pipe_table_header": return [ _node_text(src, cell).decode("utf-8", errors="replace").strip() for cell in child.children if cell.type == "pipe_table_cell" ] return [] def _extract_table_data_bytes(self, node: Node, src: bytes) -> bytes: """Return concatenated bytes of all data rows (excludes header/delimiter).""" parts: list[bytes] = [] for child in node.children: if child.type == "pipe_table_row": parts.append(_node_text(src, child)) return b"".join(parts) def file_content_id(self, source: bytes) -> str: """Return raw-bytes SHA-256 of *source*. Markdown formatting is semantically significant (whitespace, line breaks, indentation all affect rendered output), so content IDs are intentionally byte-exact, unlike the TOML adapter which normalises key ordering and strips comments. Args: source: Raw bytes of the Markdown file. Returns: Hex-encoded SHA-256 digest. """ return _sha256_bytes(source) # --------------------------------------------------------------------------- # HTML adapter — semantic element and id-bearing element extraction # --------------------------------------------------------------------------- _HTML_SEMANTIC_TAGS: frozenset[str] = frozenset({ "main", "header", "footer", "nav", "article", "section", "aside", "h1", "h2", "h3", "h4", "h5", "h6", "form", "fieldset", "dialog", "figure", "details", "summary", "template", "slot", }) _HTML_HEADINGS: frozenset[str] = frozenset({"h1", "h2", "h3", "h4", "h5", "h6"}) # Elements whose visible text content IS their label — extract like headings. # summary → disclosure button text; figcaption → figure caption; legend → fieldset legend. _HTML_LABEL_TAGS: frozenset[str] = frozenset({"summary", "figcaption", "legend", "caption"}) class HtmlAdapter: """Extract named HTML elements as symbols. Emits a symbol for any element that can be meaningfully named: - **Headings** (h1–h6): ``section`` kind, name = ``"h1: heading text"`` - **id-bearing elements**: ``section`` kind, name = ``"tag#id"`` - **aria-label elements**: ``section`` kind, name = ``"tag[label]"`` - **Named form/fieldset/slot** (``name`` attr): ``section`` kind, name = ``"tag[name-value]"`` - **Semantic elements** with a child heading: ``section`` kind, name = ``"tag: child-heading-text"`` - **Custom elements** (hyphenated tags — Web Components) with id, aria-label, or name: ``section`` kind, name derived by same priority - **Semantic elements without any name signal**: ``section`` kind, name = ``"tag@lineno"`` (last resort; line-number is fragile but beats nothing for structural awareness) Requires ``tree-sitter-html``; degrades to empty symbol tree without it. Name-resolution priority (highest to lowest): 1. ``id`` attribute → ``tag#id`` 2. ``aria-label`` attribute → ``tag[label]`` 3. ``name`` attribute (form / fieldset / slot / input) → ``tag[name]`` 4. Heading text content (h1–h6 only) 5. First direct child heading text → ``tag: heading text`` 6. ``@lineno`` fallback """ _EXTENSIONS: frozenset[str] = frozenset({".html", ".htm"}) def __init__(self) -> None: self._parser: Parser | None = None try: from tree_sitter import Language, Parser import tree_sitter_html as _html lang = Language(_html.language()) self._parser = Parser(lang) except Exception as exc: # noqa: BLE001 logger.debug("tree-sitter-html unavailable — HTML file-level only: %s", exc) def supported_extensions(self) -> frozenset[str]: return self._EXTENSIONS def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: if self._parser is None: return {} try: tree = self._parser.parse(source) except Exception as exc: # noqa: BLE001 logger.debug("HTML parse error in %s: %s", file_path, exc) return {} symbols: SymbolTree = {} self._walk(tree.root_node, source, file_path, symbols) return symbols def _walk(self, node: Node, src: bytes, file_path: str, out: SymbolTree) -> None: if node.type == "element": self._try_emit(node, src, file_path, out) for child in node.children: self._walk(child, src, file_path, out) def _try_emit( self, node: Node, src: bytes, file_path: str, out: SymbolTree ) -> None: start_tag = next((c for c in node.children if c.type == "start_tag"), None) if start_tag is None: return tag_node = start_tag.child_by_field_name("name") if tag_node is None: tag_node = next( (c for c in start_tag.named_children if c.type == "tag_name"), None ) if tag_node is None: return tag = _node_text(src, tag_node).decode("utf-8", errors="replace").lower() is_heading = tag in _HTML_HEADINGS is_label = tag in _HTML_LABEL_TAGS is_semantic = tag in _HTML_SEMANTIC_TAGS is_custom = "-" in tag # Web Component custom element # Attribute lookups — ordered by resolution priority. element_id = self._find_attr(start_tag, src, "id") aria_label = self._find_attr(start_tag, src, "aria-label") # ``name`` attr is meaningful on form, fieldset, slot, and input. name_attr = ( self._find_attr(start_tag, src, "name") if tag in {"form", "fieldset", "slot", "input"} else "" ) has_name_signal = bool(element_id or aria_label or name_attr) if not (is_heading or is_label or is_semantic or is_custom or has_name_signal): return lineno = node.start_point[0] + 1 if element_id: name = f"{tag}#{element_id}" elif aria_label: name = f"{tag}[{aria_label}]" elif name_attr: name = f"{tag}[{name_attr}]" elif is_heading or is_label: # Headings and label elements (summary, figcaption, legend, caption): # the visible text content IS the symbol name. text_parts = [ _node_text(src, c).decode("utf-8", errors="replace").strip() for c in node.named_children if c.type == "text" ] text = " ".join(text_parts).strip() or tag name = f"{tag}: {text}" else: # Semantic / custom element without an explicit name attribute. # Try to derive a name from the first direct child heading — this # gives `section: About Us` instead of the fragile `section@42`. child_heading = self._find_child_heading(node, src) if child_heading: name = f"{tag}: {child_heading}" else: name = f"{tag}@{lineno}" addr = f"{file_path}::{name}" node_bytes = _node_text(src, node) out[addr] = SymbolRecord( kind="section", name=name, qualified_name=name, content_id=_sha256_bytes(_norm_ws(node_bytes)), body_hash=_sha256_bytes(_norm_ws(node_bytes)), signature_id=_sha256(name), metadata_id="", canonical_key=_canonical_key(file_path, "", "section", name, lineno), lineno=lineno, end_lineno=node.end_point[0] + 1, ) def _find_attr(self, start_tag: Node, src: bytes, attr_name: str) -> str: """Return the value of *attr_name* on *start_tag*, or ``""``.""" for attr in start_tag.named_children: if attr.type != "attribute": continue children = attr.named_children if not children: continue name = _node_text(src, children[0]).decode("utf-8", errors="replace").strip() if name == attr_name and len(children) >= 2: val = _node_text(src, children[-1]).decode("utf-8", errors="replace") return val.strip('"\'').strip() return "" def _find_child_heading(self, element_node: Node, src: bytes) -> str: """Return the text content of the first direct child heading element, or ``""``.""" for child in element_node.children: if child.type != "element": continue start = next((c for c in child.children if c.type == "start_tag"), None) if start is None: continue tag_node = next( (c for c in start.named_children if c.type == "tag_name"), None ) if tag_node is None: continue child_tag = _node_text(src, tag_node).decode("utf-8", errors="replace").lower() if child_tag in _HTML_HEADINGS: text_parts = [ _node_text(src, c).decode("utf-8", errors="replace").strip() for c in child.named_children if c.type == "text" ] return " ".join(text_parts).strip() return "" def file_content_id(self, source: bytes) -> str: return _sha256_bytes(source) # --------------------------------------------------------------------------- # TOML adapter — stdlib tomllib, zero external dependencies # --------------------------------------------------------------------------- # Recursive type alias for the complete set of values tomllib can produce. # PEP 695 syntax (Python 3.12+); recursive aliases are fully supported. type _TomlScalar = ( str | int | float | bool | datetime.datetime | datetime.date | datetime.time ) type _TomlValue = _TomlScalar | list[_TomlValue] | dict[str, _TomlValue] class TomlAdapter: """TOML configuration file adapter — Python 3.11+ stdlib :mod:`tomllib`. Extracts addressable symbols from ``*.toml`` files so that Muse can detect *which* tables and keys changed rather than just *that the file changed*. Symbol taxonomy --------------- ``section`` A TOML table (``[project]``, ``[tool.mypy]``) or an array-of-tables entry (``[[tool.mypy.overrides]]``). The symbol name is the full dotted path; array entries carry an integer index suffix, e.g. ``tool.mypy.overrides[0]``. ``variable`` A scalar key-value pair (``name = "muse"``, ``strict = true``) or a list whose elements are not all tables. The symbol name is the dotted key path; the value is encoded in ``body_hash`` so that two keys holding the same value share a ``body_hash`` regardless of their name (rename detection). Semantic content IDs -------------------- Both ``file_content_id`` and the per-symbol ``content_id`` / ``body_hash`` are computed from ``json.dumps(value, sort_keys=True, default=str)``. This makes content IDs: - **Comment-insensitive** — TOML comments are stripped by the parser. - **Key-order-insensitive** — ``sort_keys=True`` normalises key ordering. - **Whitespace-insensitive** — ``tomllib`` discards insignificant whitespace. - **Date-stable** — ``datetime`` objects serialise to ISO 8601 strings via ``default=str``, which is deterministic. Line-number limitation ---------------------- :mod:`tomllib` does not expose positional information. All symbols are recorded with ``lineno=0`` and ``end_lineno=0``. Because TOML forbids duplicate keys within a table, the combination ``(file, prefix, kind, key)`` is already globally unique, so ``canonical_key`` remains collision-free despite the constant line number. Depth limit ----------- Recursion stops at ``_MAX_DEPTH`` (default 6) to prevent enormous configs from emitting thousands of symbols. Tables beyond that depth are still hashed into their ancestor's ``content_id``; they simply do not appear as independent symbols. """ _EXTENSIONS: frozenset[str] = frozenset({".toml"}) _MAX_DEPTH: int = 6 def supported_extensions(self) -> frozenset[str]: return self._EXTENSIONS def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: """Extract table and key-value symbols from *source*. Args: source: Raw bytes of the TOML file. file_path: Workspace-relative POSIX path — used as address prefix. Returns: A :type:`SymbolTree` mapping dotted addresses to :class:`SymbolRecord` dicts. Returns an empty dict on parse errors so that the caller falls back to file-level diffing. """ try: data = tomllib.loads(source.decode("utf-8", errors="replace")) except tomllib.TOMLDecodeError: return {} symbols: SymbolTree = {} self._walk(data, file_path, "", symbols, 0) return symbols def _walk( self, mapping: _TomlMapping, file_path: str, prefix: str, out: SymbolTree, depth: int, ) -> None: """Recursively emit :class:`SymbolRecord` entries for *mapping*. Args: mapping: The TOML dict to walk (top-level or a nested table). file_path: Workspace-relative path — used as address prefix. prefix: Dotted key path accumulated so far (empty at top level). out: Accumulator — modified in place. depth: Current nesting depth; stops recursing at ``_MAX_DEPTH``. """ if depth > self._MAX_DEPTH: return for key, value in mapping.items(): dotted = f"{prefix}.{key}" if prefix else key addr = f"{file_path}::{dotted}" if isinstance(value, dict): # Inline table or [section] — emit as section, then recurse. content = json.dumps(value, sort_keys=True, default=str) out[addr] = SymbolRecord( kind="section", name=key, qualified_name=dotted, content_id=_sha256(content), body_hash=_sha256(content), signature_id=_sha256(dotted), metadata_id="", canonical_key=_canonical_key(file_path, prefix, "section", key, 0), lineno=0, end_lineno=0, ) self._walk(value, file_path, dotted, out, depth + 1) elif ( isinstance(value, list) and len(value) > 0 and all(isinstance(item, dict) for item in value) ): # [[array.of.tables]] — each entry becomes an indexed section. for idx, item in enumerate(value): entry_name = f"{dotted}[{idx}]" entry_addr = f"{file_path}::{entry_name}" content = json.dumps(item, sort_keys=True, default=str) out[entry_addr] = SymbolRecord( kind="section", name=entry_name, qualified_name=entry_name, content_id=_sha256(content), body_hash=_sha256(content), signature_id=_sha256(entry_name), metadata_id="", canonical_key=_canonical_key( file_path, prefix, "section", key, idx ), lineno=idx, end_lineno=idx, ) # isinstance guard needed for mypy narrowing from _TomlValue. if isinstance(item, dict): self._walk(item, file_path, entry_name, out, depth + 1) else: # Scalar or mixed list — emit as variable. # content_id binds the key so two different keys with the same # value produce different content_ids. body_hash binds only # the value so that a rename (same value, new key) is detected. val_str = json.dumps(value, sort_keys=True, default=str) out[addr] = SymbolRecord( kind="variable", name=key, qualified_name=dotted, content_id=_sha256(f"{dotted}={val_str}"), body_hash=_sha256(val_str), signature_id=_sha256(dotted), metadata_id="", canonical_key=_canonical_key( file_path, prefix, "variable", key, 0 ), lineno=0, end_lineno=0, ) def file_content_id(self, source: bytes) -> str: """Return a semantic content ID for the whole file. Computed from ``json.dumps(parsed, sort_keys=True, default=str)`` so the ID is insensitive to comments, key ordering, and whitespace. Falls back to raw-bytes SHA-256 when parsing fails. Args: source: Raw bytes of the TOML file. Returns: Hex-encoded SHA-256 digest. """ try: data = tomllib.loads(source.decode("utf-8", errors="replace")) return _sha256(json.dumps(data, sort_keys=True, default=str)) except Exception: # noqa: BLE001 return _sha256_bytes(source) # --------------------------------------------------------------------------- # tree-sitter adapter — shared infrastructure for all non-Python languages # --------------------------------------------------------------------------- _WS_RE: re.Pattern[bytes] = re.compile(rb"\s+") def _norm_ws(src: bytes) -> bytes: """Collapse all whitespace runs to a single space and strip the result.""" return _WS_RE.sub(b" ", src).strip() def _node_text(src: bytes, node: Node) -> bytes: """Extract the raw source bytes covered by a tree-sitter node.""" return src[node.start_byte : node.end_byte] def _resolve_scss_selector(src: bytes, rule_set_node: "Node") -> str: """Return the fully expanded CSS selector for a (possibly nested) SCSS rule_set. tree-sitter-scss handles two forms of ``&`` nesting differently: * ``&:hover`` / ``&.mod`` — the ``&`` is kept as a ``nesting_selector`` node *inside* the ``selectors`` child, so the raw text contains ``&``. * ``&__name`` (BEM suffix) — tree-sitter splits this into a preceding ``ERROR`` node containing a ``nesting_selector`` (the bare ``&``) and a ``rule_set`` whose ``selectors`` child is just ``__name``. The ``&`` is not present in the captured text at all. Both forms are detected and resolved recursively so that multi-level nesting like ``.a { &__b { &__c {} } }`` produces ``.a__b__c``. """ sel_node: "Node | None" = None for child in rule_set_node.children: if child.type == "selectors": sel_node = child break if sel_node is None: return "" raw_txt = _node_text(src, sel_node).decode("utf-8", errors="replace").strip() # Case 1 — explicit & (e.g. `&:hover`, `&.active` kept by tree-sitter-css) has_explicit_amp = "&" in raw_txt # Case 2 — implicit & — tree-sitter split `&__name` into a preceding ERROR # sibling that contains a nesting_selector, then this rule_set with `__name`. has_implicit_amp = False parent_block = rule_set_node.parent if parent_block is not None and parent_block.type == "block": prev: "Node | None" = None for child in parent_block.children: # tree-sitter wraps each access in a new Python object — use # byte-position identity instead of `is`. if child.start_byte == rule_set_node.start_byte: break prev = child if prev is not None and prev.type == "ERROR": has_implicit_amp = any(c.type == "nesting_selector" for c in prev.children) if not has_explicit_amp and not has_implicit_amp: return raw_txt # Find nearest ancestor rule_set for the parent selector. parent_rule_set: "Node | None" = None node = rule_set_node.parent while node is not None and node.type != "stylesheet": if node.type == "rule_set": parent_rule_set = node break node = node.parent if parent_rule_set is None: # & at top level — strip it (produces the bare suffix as a symbol name). return raw_txt.replace("&", "").strip() if has_explicit_amp else raw_txt parent_resolved = _resolve_scss_selector(src, parent_rule_set) if has_explicit_amp: return raw_txt.replace("&", parent_resolved) # has_implicit_amp: selector is `__name`, parent is `.block` → `.block__name` return parent_resolved + raw_txt def _class_name_from(src: bytes, node: Node, field: str) -> str | None: """Extract a class/struct name from a parent CST node. Tries ``child_by_field_name(field)`` first (covers Java, C#, C++, Rust). Falls back to the first ``identifier``-typed named child to handle languages like Kotlin where the class name is not a named field. """ child = node.child_by_field_name(field) if child is None: for c in node.named_children: if c.type == "identifier": child = c break if child is None: return None return _node_text(src, child).decode("utf-8", errors="replace") def _qualified_name_ts( src: bytes, sym_node: Node, name: str, class_node_types: frozenset[str], class_name_field: str, ) -> str: """Walk the CST parent chain to build a dotted qualified name. For a method ``bark`` inside ``class Dog``, returns ``"Dog.bark"``. For a top-level function, returns just ``"standalone"``. """ parts = [name] parent = sym_node.parent while parent is not None: if parent.type in class_node_types: class_name = _class_name_from(src, parent, class_name_field) if class_name: parts.insert(0, class_name) parent = parent.parent return ".".join(parts) class LangSpec(TypedDict): """Per-language tree-sitter configuration consumed by :class:`TreeSitterAdapter`. Each entry in :data:`_TS_LANG_SPECS` is one of these dicts. The ``name`` field is the canonical lowercase identifier used in ``.muse/code_config.toml`` ``[code] languages`` lists — e.g. ``"javascript"``, ``"typescript"``, ``"bash"``. Users who want symbol-level analysis only for a subset of languages write:: [code] languages = ["python", "typescript", "css"] Any spec whose ``name`` is absent from that list is replaced by a :class:`FallbackAdapter` for the current process, so the grammar package need not be uninstalled — just un-listed. """ name: str # Canonical lowercase language identifier used in code_config.toml extensions: frozenset[str] module_name: str # Python import name, e.g. ``"tree_sitter_javascript"`` lang_func: str # Attribute on the module returning the raw capsule query_str: str # tree-sitter S-expr query — must capture ``@sym`` and ``@name`` kind_map: _KindMap # CST node type → SymbolKind child_kind_overrides: _KindMap # direct-child node type → SymbolKind override; # checked after kind_map to refine parent-node kinds # (e.g. Go type_spec with struct_type child → "struct") class_node_types: frozenset[str] # Ancestor types that scope methods class_name_field: str # Field name for the class name (e.g. ``"name"`` or ``"type"``) receiver_capture: str # Capture name for Go-style method receivers; ``""`` to skip async_node_child: str # Direct child type marking async (``"async"`` for JS/TS; ``""`` to skip) resolve_selector: bool # True for CSS/SCSS: expand ``&`` nesting in rule_set selectors class TreeSitterAdapter: """Implements :class:`LanguageAdapter` using tree-sitter for real CST parsing. tree-sitter is the same parsing technology used by GitHub Copilot, VS Code, Neovim, and Zed. It produces a concrete syntax tree from every source file, even if the file has syntax errors — making it suitable for real-world repos that may contain partially-written code. Parsing is error-tolerant: individual file failures are logged at DEBUG level and return an empty :type:`SymbolTree` so the caller falls back to file-level diffing rather than crashing. """ def __init__( self, spec: LangSpec, parser: Parser, query: Query, ) -> None: self._spec = spec self._parser = parser self._query = query def supported_extensions(self) -> frozenset[str]: return self._spec["extensions"] def parse_symbols(self, source: bytes, file_path: str) -> SymbolTree: from tree_sitter import QueryCursor try: tree = self._parser.parse(source) cursor = QueryCursor(self._query) symbols: SymbolTree = {} recv_cap = self._spec["receiver_capture"] for _pat, caps in cursor.matches(tree.root_node): sym_list = caps.get("sym", []) name_list = caps.get("name", []) if not sym_list or not name_list: continue sym_node = sym_list[0] name_node = name_list[0] name_txt = _node_text(source, name_node).decode( "utf-8", errors="replace" ) # SCSS nesting: expand `&` refs into the compiled class name. # tree-sitter-css splits `&__name` into a preceding ERROR node # and a rule_set with selector `__name` — _resolve_scss_selector # handles both BEM suffixes and pseudo-class forms (&:hover). if sym_node.type == "rule_set" and self._spec["resolve_selector"]: name_txt = _resolve_scss_selector(source, sym_node) kind = self._spec["kind_map"].get(sym_node.type, "function") # Refine kind using direct child node types. Used for languages # like Go where the same parent node type (``type_spec``) covers # structs, interfaces, and plain aliases — the child distinguishes. child_overrides = self._spec["child_kind_overrides"] if child_overrides: for child in sym_node.children: if child.type in child_overrides: kind = child_overrides[child.type] break # Promote function/method to async variant when the node has # an "async" keyword as a direct child (JS, TS, Swift, etc.). async_child = self._spec["async_node_child"] if async_child and kind in ("function", "method"): if any(c.type == async_child for c in sym_node.children[:3]): kind = "async_function" if kind == "function" else "async_method" # Build qualified name — walking ancestor chain for methods. qualified = _qualified_name_ts( source, sym_node, name_txt, self._spec["class_node_types"], self._spec["class_name_field"], ) # Go-style receiver prefix: (d *Dog) → "Dog.Bark" if recv_cap: recv_list = caps.get(recv_cap, []) if recv_list: recv_txt = ( _node_text(source, recv_list[0]) .decode("utf-8", errors="replace") .lstrip("*") .strip() ) if recv_txt: qualified = f"{recv_txt}.{qualified}" addr = f"{file_path}::{qualified}" node_bytes = _node_text(source, sym_node) name_bytes = _node_text(source, name_node) # Substitute the name with a placeholder to isolate the body # from the identifier — two symbols with the same body but # different names share the same body_hash, signalling a rename. body_bytes = node_bytes.replace(name_bytes, b"\xfe", 1) params_node = ( sym_node.child_by_field_name("parameters") or sym_node.child_by_field_name("formal_parameters") or sym_node.child_by_field_name("function_value_parameters") ) params_bytes = ( _node_text(source, params_node) if params_node is not None else b"" ) sym_lineno = sym_node.start_point[0] + 1 # Determine class prefix for canonical_key (dotted scope path). scope_prefix = ".".join(qualified.split(".")[:-1]) + "." if "." in qualified else "" symbols[addr] = SymbolRecord( kind=kind, name=name_txt, qualified_name=qualified, content_id=_sha256_bytes(_norm_ws(node_bytes)), body_hash=_sha256_bytes(_norm_ws(body_bytes)), signature_id=_sha256_bytes(_norm_ws(name_bytes + params_bytes)), # metadata_id: tree-sitter adapters extract annotations/visibility # where available. Currently stubbed as "" — future adapters can # enrich this by reading modifier nodes. metadata_id="", canonical_key=_canonical_key(file_path, scope_prefix, kind, name_txt, sym_lineno), lineno=sym_lineno, end_lineno=sym_node.end_point[0] + 1, ) return symbols except Exception as exc: # noqa: BLE001 logger.debug("tree-sitter parse error in %s: %s", file_path, exc) return {} def file_content_id(self, source: bytes) -> str: """Whitespace-normalised SHA-256 of the source — insensitive to reformatting.""" return _sha256_bytes(_norm_ws(source)) def validate_source(self, source: bytes) -> str | None: """Return an error description if *source* has syntax errors, else None. tree-sitter always produces a parse tree even for broken code. Errors appear as nodes with ``type == "ERROR"`` or ``is_missing == True``. ``root_node.has_error`` is the fast top-level check. """ try: tree = self._parser.parse(source) except Exception as exc: # noqa: BLE001 return f"parser error: {exc}" if not tree.root_node.has_error: return None # Walk the tree to find the first concrete error site. error_node = _first_error_node(tree.root_node) if error_node is not None: line = error_node.start_point[0] + 1 fragment = source[ error_node.start_byte : min(error_node.end_byte, error_node.start_byte + 60) ].decode("utf-8", errors="replace").strip() msg = f"syntax error on line {line}" if fragment: msg += f": {fragment!r}" return msg return "syntax error (unknown location)" def _make_ts_adapter(spec: LangSpec) -> LanguageAdapter: """Build a :class:`TreeSitterAdapter`; fall back to :class:`FallbackAdapter` on error. Importing the grammar capsule is deferred to this factory so that a missing or incompatible grammar package degrades gracefully rather than preventing the entire plugin from loading. tree_sitter itself is also imported here — not at module level — so that importing ast_parser does not pay the C-extension load cost unless semantic analysis is actually requested. """ try: import warnings from tree_sitter import Language, Parser, Query mod = importlib.import_module(spec["module_name"]) raw_lang = getattr(mod, spec["lang_func"])() # tree_sitter_scss 1.0.0 returns an int (old C-pointer API) while all # other grammars return a PyCapsule. Suppress the "int argument # support is deprecated" DeprecationWarning that tree-sitter emits # when it receives an int so that the warning does not surface in tests # and user output. The int path still works — it is only deprecated. with warnings.catch_warnings(): warnings.filterwarnings( "ignore", message="int argument support is deprecated", category=DeprecationWarning, ) lang = Language(raw_lang) parser = Parser(lang) query = Query(lang, spec["query_str"]) return TreeSitterAdapter(spec, parser, query) except Exception as exc: # noqa: BLE001 logger.debug( "tree-sitter grammar %s.%s unavailable — using file-level fallback: %s", spec["module_name"], spec["lang_func"], exc, ) return FallbackAdapter(spec["extensions"]) # --------------------------------------------------------------------------- # Per-language tree-sitter specs # --------------------------------------------------------------------------- _JS_SPEC: LangSpec = { "name": "javascript", "extensions": frozenset({".js", ".jsx", ".mjs", ".cjs"}), "module_name": "tree_sitter_javascript", "lang_func": "language", # tree-sitter-javascript uses "class" for named class expressions. # Arrow functions and function expressions assigned to variables are # captured via variable_declarator so that `const greet = () => {}` is # a first-class symbol. "query_str": ( "(function_declaration name: (identifier) @name) @sym\n" "(function_expression name: (identifier) @name) @sym\n" "(generator_function_declaration name: (identifier) @name) @sym\n" "(class_declaration name: (identifier) @name) @sym\n" "(class name: (identifier) @name) @sym\n" "(method_definition name: (property_identifier) @name) @sym\n" "(variable_declarator name: (identifier) @name" " value: (arrow_function)) @sym\n" "(variable_declarator name: (identifier) @name" " value: (function_expression)) @sym\n" "(variable_declarator name: (identifier) @name" " value: (generator_function)) @sym" ), "kind_map": { "function_declaration": "function", "function_expression": "function", "generator_function_declaration": "function", "class_declaration": "class", "class": "class", "method_definition": "method", "variable_declarator": "function", }, "child_kind_overrides": {}, "class_node_types": frozenset({"class_declaration", "class"}), "class_name_field": "name", "receiver_capture": "", # async keyword appears as a direct child token on function/method nodes. "async_node_child": "async", "resolve_selector": False, } _TS_QUERY = ( # TypeScript uses type_identifier (not identifier) for class names. "(function_declaration name: (identifier) @name) @sym\n" "(function_expression name: (identifier) @name) @sym\n" "(generator_function_declaration name: (identifier) @name) @sym\n" "(class_declaration name: (type_identifier) @name) @sym\n" "(class name: (type_identifier) @name) @sym\n" "(abstract_class_declaration name: (type_identifier) @name) @sym\n" "(method_definition name: (property_identifier) @name) @sym\n" "(interface_declaration name: (type_identifier) @name) @sym\n" "(type_alias_declaration name: (type_identifier) @name) @sym\n" "(enum_declaration name: (identifier) @name) @sym\n" "(internal_module name: (identifier) @name) @sym\n" "(variable_declarator name: (identifier) @name" " value: (arrow_function)) @sym\n" "(variable_declarator name: (identifier) @name" " value: (function_expression)) @sym\n" "(variable_declarator name: (identifier) @name" " value: (generator_function)) @sym" ) _TS_KIND_MAP: _KindMap = { "function_declaration": "function", "function_expression": "function", "generator_function_declaration": "function", "class_declaration": "class", "class": "class", "abstract_class_declaration": "class", "method_definition": "method", "interface_declaration": "interface", "type_alias_declaration": "type_alias", "enum_declaration": "enum", "internal_module": "namespace", "variable_declarator": "function", } _TS_CLASS_NODES: frozenset[str] = frozenset( {"class_declaration", "class", "abstract_class_declaration"} ) _TS_SPEC: LangSpec = { "name": "typescript", "extensions": frozenset({".ts"}), "module_name": "tree_sitter_typescript", "lang_func": "language_typescript", "query_str": _TS_QUERY, "kind_map": _TS_KIND_MAP, "child_kind_overrides": {}, "class_node_types": _TS_CLASS_NODES, "class_name_field": "name", "receiver_capture": "", "async_node_child": "async", "resolve_selector": False, } _TSX_SPEC: LangSpec = { "name": "tsx", "extensions": frozenset({".tsx"}), "module_name": "tree_sitter_typescript", "lang_func": "language_tsx", "query_str": _TS_QUERY, "kind_map": _TS_KIND_MAP, "child_kind_overrides": {}, "class_node_types": _TS_CLASS_NODES, "class_name_field": "name", "receiver_capture": "", "async_node_child": "async", "resolve_selector": False, } _GO_SPEC: LangSpec = { "name": "go", "extensions": frozenset({".go"}), "module_name": "tree_sitter_go", "lang_func": "language", "query_str": ( "(function_declaration name: (identifier) @name) @sym\n" "(method_declaration\n" " receiver: (parameter_list\n" " (parameter_declaration type: _ @recv))\n" " name: (field_identifier) @name) @sym\n" "(type_spec name: (type_identifier) @name) @sym\n" "(const_spec name: (identifier) @name) @sym\n" "(var_spec name: (identifier) @name) @sym" ), "kind_map": { "function_declaration": "function", "method_declaration": "method", "type_spec": "type_alias", # refined by child_kind_overrides below "const_spec": "variable", "var_spec": "variable", }, # Go type_spec covers struct, interface, and plain type aliases — the direct # child node type is the discriminant. child_kind_overrides refines the # parent-level kind after the kind_map lookup. "child_kind_overrides": { "struct_type": "struct", "interface_type": "interface", }, "class_node_types": frozenset(), "class_name_field": "name", "receiver_capture": "recv", "async_node_child": "", "resolve_selector": False, } _RUST_SPEC: LangSpec = { "name": "rust", "extensions": frozenset({".rs"}), "module_name": "tree_sitter_rust", "lang_func": "language", "query_str": ( "(function_item name: (identifier) @name) @sym\n" "(struct_item name: (type_identifier) @name) @sym\n" "(enum_item name: (type_identifier) @name) @sym\n" "(trait_item name: (type_identifier) @name) @sym\n" "(type_item name: (type_identifier) @name) @sym\n" "(mod_item name: (identifier) @name) @sym\n" "(static_item name: (identifier) @name) @sym\n" "(const_item name: (identifier) @name) @sym" ), "kind_map": { "function_item": "function", "struct_item": "struct", "enum_item": "enum", "trait_item": "trait", "type_item": "type_alias", "mod_item": "module", "static_item": "variable", "const_item": "variable", }, "child_kind_overrides": {}, # impl_item scopes methods; its implementing type is in the "type" field. "class_node_types": frozenset({"impl_item"}), "class_name_field": "type", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } _JAVA_SPEC: LangSpec = { "name": "java", "extensions": frozenset({".java"}), "module_name": "tree_sitter_java", "lang_func": "language", "query_str": ( "(method_declaration name: (identifier) @name) @sym\n" "(constructor_declaration name: (identifier) @name) @sym\n" "(class_declaration name: (identifier) @name) @sym\n" "(interface_declaration name: (identifier) @name) @sym\n" "(enum_declaration name: (identifier) @name) @sym\n" "(annotation_type_declaration name: (identifier) @name) @sym\n" "(record_declaration name: (identifier) @name) @sym" ), "kind_map": { "method_declaration": "method", "constructor_declaration": "function", "class_declaration": "class", "interface_declaration": "interface", "enum_declaration": "enum", "annotation_type_declaration": "class", "record_declaration": "struct", }, "child_kind_overrides": {}, "class_node_types": frozenset( {"class_declaration", "interface_declaration", "enum_declaration", "record_declaration"} ), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } _C_SPEC: LangSpec = { "name": "c", "extensions": frozenset({".c", ".h"}), "module_name": "tree_sitter_c", "lang_func": "language", "query_str": ( "(function_definition\n" " declarator: (function_declarator\n" " declarator: (identifier) @name)) @sym\n" # Structs and enums defined via typedef or direct declaration. "(struct_specifier name: (type_identifier) @name) @sym\n" "(enum_specifier name: (type_identifier) @name) @sym" ), "kind_map": { "function_definition": "function", "struct_specifier": "struct", "enum_specifier": "enum", }, "child_kind_overrides": {}, "class_node_types": frozenset(), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } _CPP_SPEC: LangSpec = { "name": "cpp", "extensions": frozenset({".cpp", ".cc", ".cxx", ".hpp", ".hxx"}), "module_name": "tree_sitter_cpp", "lang_func": "language", "query_str": ( # Plain function definitions (top-level or namespaced). "(function_definition\n" " declarator: (function_declarator\n" " declarator: (identifier) @name)) @sym\n" # Out-of-class method definitions: void Dog::bark() {} "(function_definition\n" " declarator: (function_declarator\n" " declarator: (qualified_identifier\n" " name: (identifier) @name))) @sym\n" "(class_specifier name: (type_identifier) @name) @sym\n" "(struct_specifier name: (type_identifier) @name) @sym\n" "(enum_specifier name: (type_identifier) @name) @sym\n" "(namespace_definition (namespace_identifier) @name) @sym" ), "kind_map": { "function_definition": "function", "class_specifier": "class", "struct_specifier": "struct", "enum_specifier": "enum", "namespace_definition": "namespace", }, "child_kind_overrides": {}, "class_node_types": frozenset({"class_specifier", "struct_specifier"}), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } _CS_SPEC: LangSpec = { "name": "csharp", "extensions": frozenset({".cs"}), "module_name": "tree_sitter_c_sharp", "lang_func": "language", "query_str": ( "(method_declaration name: (identifier) @name) @sym\n" "(constructor_declaration name: (identifier) @name) @sym\n" "(class_declaration name: (identifier) @name) @sym\n" "(interface_declaration name: (identifier) @name) @sym\n" "(enum_declaration name: (identifier) @name) @sym\n" "(struct_declaration name: (identifier) @name) @sym\n" "(record_declaration name: (identifier) @name) @sym\n" "(property_declaration name: (identifier) @name) @sym" ), "kind_map": { "method_declaration": "method", "constructor_declaration": "function", "class_declaration": "class", "interface_declaration": "interface", "enum_declaration": "enum", "struct_declaration": "struct", "record_declaration": "struct", "property_declaration": "variable", }, "child_kind_overrides": {}, "class_node_types": frozenset( {"class_declaration", "interface_declaration", "struct_declaration", "record_declaration"} ), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } _RUBY_SPEC: LangSpec = { "name": "ruby", "extensions": frozenset({".rb"}), "module_name": "tree_sitter_ruby", "lang_func": "language", "query_str": ( "(method name: (identifier) @name) @sym\n" "(singleton_method name: (identifier) @name) @sym\n" "(class name: (constant) @name) @sym\n" "(module name: (constant) @name) @sym\n" "(singleton_class value: (self) @name) @sym" ), "kind_map": { "method": "method", "singleton_method": "method", "class": "class", "module": "module", "singleton_class": "class", }, "child_kind_overrides": {}, "class_node_types": frozenset({"class", "module"}), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } _KT_SPEC: LangSpec = { "name": "kotlin", "extensions": frozenset({".kt", ".kts"}), "module_name": "tree_sitter_kotlin", "lang_func": "language", # Kotlin uses plain `identifier` for all names (no type_identifier or # simple_identifier variants at this grammar version). "query_str": ( "(function_declaration (identifier) @name) @sym\n" "(class_declaration (identifier) @name) @sym\n" "(object_declaration (identifier) @name) @sym\n" "(property_declaration (variable_declaration" " (identifier) @name)) @sym" ), "kind_map": { "function_declaration": "function", "class_declaration": "class", "object_declaration": "object", # singleton — semantically distinct from class "property_declaration": "variable", }, "child_kind_overrides": {}, # Kotlin methods are function_declaration nodes inside class_body. # child_by_field_name("name") is None for Kotlin classes; _class_name_from # falls back to the first identifier-typed named child automatically. # object_declaration excluded: methods on a singleton scope as Singleton.method # but object_declaration itself is not a "class" container for instantiation. "class_node_types": frozenset({"class_declaration", "object_declaration"}), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } # Swift: requires py-tree-sitter-swift (builds from source). _make_ts_adapter # degrades to FallbackAdapter if the package is not available. _SWIFT_SPEC: LangSpec = { "name": "swift", "extensions": frozenset({".swift"}), "module_name": "py_tree_sitter_swift", "lang_func": "language", "query_str": ( "(function_declaration name: (simple_identifier) @name) @sym\n" "(class_declaration name: (type_identifier) @name) @sym\n" "(struct_declaration name: (type_identifier) @name) @sym\n" "(enum_declaration name: (type_identifier) @name) @sym\n" "(protocol_declaration name: (type_identifier) @name) @sym\n" "(typealias_declaration name: (type_identifier) @name) @sym\n" "(computed_property (simple_identifier) @name) @sym\n" "(init_declaration) @sym" ), "kind_map": { "function_declaration": "function", "class_declaration": "class", "struct_declaration": "struct", "enum_declaration": "enum", "protocol_declaration": "trait", # Swift protocol ≈ Rust trait "typealias_declaration": "type_alias", "computed_property": "variable", "init_declaration": "function", }, "child_kind_overrides": {}, "class_node_types": frozenset( {"class_declaration", "struct_declaration", "enum_declaration"} ), "class_name_field": "name", "receiver_capture": "", "async_node_child": "async", "resolve_selector": False, } # CSS: rule-sets, @keyframes, @media, @supports, and @layer become addressable # symbols so diffs report "selector .button changed" instead of "file changed". # tree-sitter-css handles all modern CSS syntax including cascade layers. # # @media query naming: three child types cover all real-world forms: # keyword_query → @media print # feature_query → @media (max-width: 768px) # binary_query → @media screen and (min-width: 1024px) _CSS_SPEC: LangSpec = { "name": "css", "extensions": frozenset({".css"}), "module_name": "tree_sitter_css", "lang_func": "language", "query_str": ( # Selector rules — the primary symbol in CSS. "(rule_set (selectors) @name) @sym\n" # Named animation blocks. "(keyframes_statement (keyframes_name) @name) @sym\n" # @media — three child types for the query condition: "(media_statement (keyword_query) @name) @sym\n" "(media_statement (feature_query) @name) @sym\n" "(media_statement (binary_query) @name) @sym\n" # @supports (display: grid) { … } "(supports_statement (feature_query) @name) @sym\n" # @layer base { … } / @layer utilities { … } "(at_rule (keyword_query) @name) @sym" ), "kind_map": { "rule_set": "rule", "keyframes_statement": "rule", "media_statement": "rule", "supports_statement": "rule", "at_rule": "rule", }, "child_kind_overrides": {}, "class_node_types": frozenset(), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, # plain CSS has no `&` nesting } # SCSS: superset of CSS with variables, mixins, and functions as first-class # named symbols — the most important constructs in a design-system stylesheet. # Uses tree-sitter-scss (separate grammar from tree-sitter-css). # # Symbol taxonomy: # variable — $primary-color: #333; (top-level $var declarations) # mixin — @mixin flex-center($dir) { … } (outputs CSS, @include'd) # function — @function em($px) { @return … } (returns a value) # rule — .button { … }, @keyframes, @media, @supports # # SCSS nesting: `&:hover` / `&__element` (BEM) is expanded by # _resolve_scss_selector so nested rules carry their compiled selector as the # symbol address instead of a bare `&` or suffix fragment. _SCSS_SPEC: LangSpec = { "name": "scss", "extensions": frozenset({".scss"}), "module_name": "tree_sitter_scss", "lang_func": "language", "query_str": ( # $variable declarations at stylesheet level only (not property values # inside rule blocks, which are also `declaration` nodes). "(stylesheet (declaration (property_name) @name) @sym)\n" # @mixin name(…) { … } — parameterized CSS output blocks. "(mixin_statement (identifier) @name) @sym\n" # @function name(…) { … } — returns a computed value. "(function_statement (identifier) @name) @sym\n" # Selector rule-sets (same as CSS). "(rule_set (selectors) @name) @sym\n" # Named animation blocks. "(keyframes_statement (keyframes_name) @name) @sym\n" # @media — three child types for the query condition: "(media_statement (keyword_query) @name) @sym\n" "(media_statement (feature_query) @name) @sym\n" "(media_statement (binary_query) @name) @sym" ), "kind_map": { "declaration": "variable", # $var: value at stylesheet level "mixin_statement": "mixin", "function_statement": "function", "rule_set": "rule", "keyframes_statement": "rule", "media_statement": "rule", }, "child_kind_overrides": {}, "class_node_types": frozenset(), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": True, # expand SCSS `&` nesting in rule_set selectors } # Shell: bash/sh/zsh function definitions and variable assignments. # tree-sitter-bash handles both POSIX sh and bash/zsh syntax (function_definition # nodes cover ``function foo() {}`` and ``foo() {}`` forms; zsh superset is parsed # correctly because it is backward-compatible with bash at the AST level). _BASH_SPEC: LangSpec = { "name": "bash", "extensions": frozenset({".sh", ".bash", ".zsh", ".plugin.zsh"}), "module_name": "tree_sitter_bash", "lang_func": "language", "query_str": ( # Named function definitions: function foo() {} and foo() {} "(function_definition name: (word) @name) @sym\n" # Top-level variable assignments: FOO=bar ARRAY=(...) "(variable_assignment name: (variable_name) @name) @sym" ), "kind_map": { "function_definition": "function", "variable_assignment": "variable", }, "child_kind_overrides": {}, "class_node_types": frozenset(), "class_name_field": "name", "receiver_capture": "", "async_node_child": "", "resolve_selector": False, } #: All tree-sitter language specs, loaded in registration order. _TS_LANG_SPECS: list[LangSpec] = [ _JS_SPEC, _TS_SPEC, _TSX_SPEC, _GO_SPEC, _RUST_SPEC, _JAVA_SPEC, _C_SPEC, _CPP_SPEC, _CS_SPEC, _RUBY_SPEC, _KT_SPEC, _SWIFT_SPEC, _CSS_SPEC, _SCSS_SPEC, _BASH_SPEC, ] # --------------------------------------------------------------------------- # Per-repo language configuration (.muse/code_config.toml [code] section) # --------------------------------------------------------------------------- from dataclasses import dataclass as _dataclass @_dataclass class CodeConfig: """Parsed ``[code]`` section from ``.muse/code_config.toml``. Controls which tree-sitter language adapters are active for the current repository. The defaults reproduce the previous "all installed grammars" behaviour so that repos without a config file are unaffected. Example ``.muse/code_config.toml``:: # Python + TypeScript web project — skip Go, Rust, Java, … [code] languages = ["python", "typescript", "tsx", "css", "html", "bash"] [framework_detection] # framework config unchanged Language names correspond to the ``name`` field of :class:`LangSpec` plus the built-in adapters: ``"python"``, ``"markdown"``, ``"html"``, ``"toml"``. When ``languages`` is absent or ``None``, every installed grammar is active (the zero-config default). """ active_languages: frozenset[str] | None = None # None → all installed grammars enabled def load_code_config(root: pathlib.Path) -> CodeConfig: """Read ``[code]`` from ``.muse/code_config.toml`` at *root*. Returns a :class:`CodeConfig` with defaults (all languages active) when: * the file does not exist, * the file is malformed TOML, * the ``[code]`` section is absent, or * ``languages`` is not a list of strings. Args: root: Absolute path to the repository root (the directory containing ``.muse/``). Returns: A :class:`CodeConfig` describing which languages should have symbol-level analysis for this process. """ config_path = root / ".muse" / "code_config.toml" if not config_path.exists(): return CodeConfig() try: with open(config_path, "rb") as fh: raw = tomllib.load(fh) except Exception as exc: # noqa: BLE001 logger.warning("⚠️ Could not parse .muse/code_config.toml: %s", exc) return CodeConfig() section = raw.get("code", {}) if not isinstance(section, dict): return CodeConfig() langs_raw = section.get("languages") if langs_raw is None: return CodeConfig() if not isinstance(langs_raw, list) or not all(isinstance(x, str) for x in langs_raw): logger.warning( "⚠️ .muse/code_config.toml [code] languages must be a list of strings; ignoring" ) return CodeConfig() return CodeConfig(active_languages=frozenset(s.lower().strip() for s in langs_raw)) # --------------------------------------------------------------------------- # Adapter registry and public helpers # --------------------------------------------------------------------------- #: Fallback adapter for file types without a registered adapter — always cheap. _FALLBACK: FallbackAdapter = FallbackAdapter(frozenset()) #: Internal caches — populated on first call to :func:`_adapters`. _ADAPTERS_CACHE: list[LanguageAdapter] | None = None _SEM_EXT_CACHE: frozenset[str] | None = None #: Active language filter — ``None`` means all installed grammars are enabled. #: Set once per process via :func:`configure_language_filter` before the first #: call to :func:`_adapters`. Changing it after the cache is built has no #: effect; call :func:`reset_adapter_cache` first if you need to re-configure #: (test-only — production code never resets the cache mid-process). _LANGUAGE_FILTER: frozenset[str] | None = None #: Public lazy attributes — real values injected by ``__getattr__`` on first #: access. The annotations here exist solely so the symbol graph can index #: these names; the runtime values are set via ``setattr`` in :func:`_adapters`. ADAPTERS: list[LanguageAdapter] SEMANTIC_EXTENSIONS: frozenset[str] def configure_language_filter(names: frozenset[str] | None) -> None: """Restrict which tree-sitter grammars are active for this process. Must be called **before** the first use of :func:`adapter_for_path`, :func:`parse_symbols`, or any command that touches the symbol graph. Calling it after the cache is already built has no effect — use :func:`reset_adapter_cache` first (tests only). Args: names: Lowercase language names to keep active (e.g. ``frozenset({"python", "typescript", "bash"})``). ``None`` re-enables all installed grammars (the default). Example — call from a command's ``run()`` after ``require_repo()``:: root = require_repo() cfg = load_code_config(root) configure_language_filter(cfg.active_languages) """ global _LANGUAGE_FILTER _LANGUAGE_FILTER = names def reset_adapter_cache() -> None: """Clear the adapter cache so the next call to :func:`_adapters` rebuilds it. **Test use only.** Production code — where ``require_repo()`` is called once per process — never needs this. Tests that call :func:`configure_language_filter` with different values in the same process must call this between them so the new filter takes effect. """ global _ADAPTERS_CACHE, _SEM_EXT_CACHE _ADAPTERS_CACHE = None _SEM_EXT_CACHE = None # Remove promoted module attributes so __getattr__ re-fires on next access. _mod: _types.ModuleType = sys.modules[__name__] for attr in ("ADAPTERS", "SEMANTIC_EXTENSIONS"): try: delattr(_mod, attr) except AttributeError: pass #: Built-in adapter names that are always available regardless of installed #: grammar packages. When a language filter is active, these names are #: recognised so users can include ``"python"`` or ``"markdown"`` in their #: ``languages`` list and have it work as expected. _BUILTIN_ADAPTER_NAMES: frozenset[str] = frozenset({"python", "markdown", "html", "toml"}) def _adapters() -> list[LanguageAdapter]: """Return the global adapter list, building it on first call. Tree-sitter grammar packages are imported here — not at module level — so that importing :mod:`ast_parser` costs nothing for commands that do not perform semantic analysis (e.g. ``muse init``, ``muse log``). When :data:`_LANGUAGE_FILTER` is set (via :func:`configure_language_filter`), only specs whose ``name`` appears in the filter are built as real adapters; all others become :class:`FallbackAdapter` instances. Built-in adapters (Python, Markdown, HTML, TOML) are included if their name is in the filter or if the filter is ``None``. """ global _ADAPTERS_CACHE, _SEM_EXT_CACHE if _ADAPTERS_CACHE is None: lang_filter = _LANGUAGE_FILTER # snapshot — avoids TOCTOU if threads ever appear builtin_adapters: list[LanguageAdapter] = [] for adapter_name, adapter_cls in ( ("python", PythonAdapter), ("markdown", MarkdownAdapter), ("html", HtmlAdapter), ("toml", TomlAdapter), ): if lang_filter is None or adapter_name in lang_filter: builtin_adapters.append(adapter_cls()) result: list[LanguageAdapter] = builtin_adapters for spec in _TS_LANG_SPECS: if lang_filter is None or spec["name"] in lang_filter: result.append(_make_ts_adapter(spec)) # Specs excluded by the filter produce no adapter — their extensions # fall through to FallbackAdapter in adapter_for_path(). _ADAPTERS_CACHE = result _SEM_EXT_CACHE = frozenset().union( *(a.supported_extensions() for a in result if not isinstance(a, FallbackAdapter)) ) # Promote computed values to real module attributes so subsequent # attribute lookups bypass __getattr__ and become O(1) dict access. _mod: _types.ModuleType = sys.modules[__name__] setattr(_mod, "ADAPTERS", result) setattr(_mod, "SEMANTIC_EXTENSIONS", _SEM_EXT_CACHE) return _ADAPTERS_CACHE def _semantic_extensions() -> frozenset[str]: """Return the set of extensions with AST-level support, building it on first call.""" _adapters() if _SEM_EXT_CACHE is None: return frozenset() return _SEM_EXT_CACHE def __getattr__(name: str) -> list[LanguageAdapter] | frozenset[str]: """Lazy module attributes — ADAPTERS and SEMANTIC_EXTENSIONS. Both are computed on first access (which triggers adapter building) and then cached as real module attributes so subsequent lookups are O(1). """ if name == "ADAPTERS": return _adapters() if name == "SEMANTIC_EXTENSIONS": return _semantic_extensions() raise AttributeError(f"module {__name__!r} has no attribute {name!r}") def adapter_for_path(file_path: str) -> LanguageAdapter: """Return the best :class:`LanguageAdapter` for *file_path*. Checks registered adapters in order; falls back to :class:`FallbackAdapter` when no adapter claims the suffix. Args: file_path: Workspace-relative POSIX path (e.g. ``"src/utils.py"``). Returns: The first adapter whose :meth:`~LanguageAdapter.supported_extensions` set contains the file's lowercase suffix. """ suffix = pathlib.PurePosixPath(file_path).suffix.lower() for adapter in _adapters(): if suffix in adapter.supported_extensions(): return adapter return _FALLBACK def parse_symbols(source: bytes, file_path: str) -> SymbolTree: """Parse *source* with the best available adapter for *file_path*. Args: source: Raw bytes of the source file. file_path: Workspace-relative POSIX path. Returns: A :type:`SymbolTree` (may be empty for unsupported file types). """ return adapter_for_path(file_path).parse_symbols(source, file_path) def file_content_id(source: bytes, file_path: str) -> str: """Return the semantic content ID for *file_path* given its raw *source*. Args: source: Raw bytes of the file. file_path: Workspace-relative POSIX path. Returns: Hex-encoded SHA-256 digest — AST-based for Python, raw-bytes for others. """ return adapter_for_path(file_path).file_content_id(source) def _first_error_node(node: Node) -> Node | None: """Return the first ERROR or MISSING node in *node*'s subtree, depth-first.""" if node.type == "ERROR" or node.is_missing: return node for child in node.children: found = _first_error_node(child) if found is not None: return found return None def validate_syntax(source: bytes, file_path: str) -> str | None: """Return a human-readable error description if *source* has syntax errors. Covers Python (via :mod:`ast`) and all tree-sitter languages. Returns ``None`` for valid files and for file types without a parser. This is used by ``muse patch`` to verify that a surgical replacement does not introduce a syntax error before writing the result to disk. Args: source: UTF-8 encoded source bytes to validate. file_path: Workspace-relative path — used to select the parser. Returns: A human-readable error string, or ``None`` if the file is valid. """ suffix = pathlib.PurePosixPath(file_path).suffix.lower() if suffix in {".py", ".pyi"}: if len(source) > MAX_AST_BYTES: return None # too large to validate — treat as structurally valid try: ast.parse(source) return None except SyntaxError as exc: return f"syntax error on line {exc.lineno}: {exc.msg}" adapter = adapter_for_path(file_path) if isinstance(adapter, TreeSitterAdapter): return adapter.validate_source(source) # MarkdownAdapter and HtmlAdapter use tree-sitter internally when available # but do not expose validation — no syntax errors to report for prose files. return None