From ceb1a20d83bc6c31a7674c86eaa549d09f47f4d5 Mon Sep 17 00:00:00 2001 From: Fabricio Ceolin Date: Mon, 2 Nov 2026 17:15:24 -0300 Subject: [PATCH 1/5] deps: move tree-sitter-language-pack to 1.x, and read Dart type arguments from the call The pin goes from >=0.8.0,<0.9.0 to >=1.23.1,<2.0.0. Grammars that 0.8 does not ship (Lean, among others) are only in 1.x. The 1.x Dart grammar parses `Foo(x)` as an identifier with a call selector whose argument_part holds the type_arguments, not as a relational expression. `_handle_generic_invocation` never saw it, so the type-argument edges (`Garage.open -> Item`, `carProvider -> Car`) were lost. `_process_chain` now takes them from the call. The relational expression path stays for grammars that still misparse it. 1.x downloads each grammar on first use and caches it. The Docker image fetches the grammars the analyzers load at build time, so it runs offline. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 ++++++++ codewiki/src/be/dependency_analyzer/analyzers/dart.py | 11 +++++++++++ docker/Dockerfile | 4 ++++ guides/development.md | 2 ++ pyproject.toml | 2 +- requirements.txt | 2 +- 6 files changed, 27 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 47d174ca..b8bfa0b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -93,6 +93,14 @@ uses [Semantic Versioning](https://semver.org/). `--update` keeps it, so existing flat docs stay flat. The GitHub Pages viewer, the web app and the MCP tools (`doc_path` in `processing_order.json`) follow the layout. +- **tree-sitter-language-pack 1.x.** The pin moves from `>=0.8.0,<0.9.0` to + `>=1.23.1,<2.0.0`, which brings grammars 0.8 does not ship (Lean, among + others). 1.x downloads a grammar on first use and caches it under + `tree_sitter_language_pack.cache_dir()`, so the first run of the Dart, Go, + JavaScript/TypeScript-in-Vue, Vue and Zig analyzers needs network access; + the Docker image fetches them at build time. The 1.x Dart grammar parses + `Foo(x)` as a call instead of a relational expression, and the Dart + analyzer now takes the type-argument edges from that call too. ## [2.0.0] - 2026-09-18 diff --git a/codewiki/src/be/dependency_analyzer/analyzers/dart.py b/codewiki/src/be/dependency_analyzer/analyzers/dart.py index c773265d..1b38491c 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/dart.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/dart.py @@ -762,6 +762,17 @@ def _process_chain(self, target, head, selectors, local_types, lift_to): return caller, owner = target.caller, target.owner line = head.start_point[0] + 1 + # `Foo(x)` / `foo(x)`: the type arguments sit in the call's + # argument_part (grammars that misparse it as a relational expression + # are handled by _handle_generic_invocation). + for token_kind, _, token_node in tokens: + if token_kind != "call": + continue + type_args = _first_child(token_node, "type_arguments") + if type_args is not None: + for ref in _descendants(type_args): + if ref.type == "type_identifier": + self._add_type_edge(caller, _text(ref), line) name = _text(head) if head.type == "identifier" and name in self.import_prefixes and tokens[0][0] == "member": name, tokens = tokens[0][1] or "", tokens[1:] diff --git a/docker/Dockerfile b/docker/Dockerfile index d0a70d16..c6b9b086 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -18,6 +18,10 @@ COPY requirements.txt . # Install Python dependencies RUN pip install --no-cache-dir -r requirements.txt +# tree-sitter-language-pack 1.x downloads each grammar on first use. Fetch the +# ones the analyzers load at build time, so the image runs offline. +RUN python -c "import tree_sitter_language_pack as t; t.download(['dart', 'go', 'javascript', 'typescript', 'vue', 'zig'])" + # Copy application code COPY codewiki ./codewiki COPY img ./img diff --git a/guides/development.md b/guides/development.md index 4c88fb32..8626e9b5 100644 --- a/guides/development.md +++ b/guides/development.md @@ -123,6 +123,8 @@ Analyzers are standalone classes, one file per language. Follow and `requirements.txt`, or take the grammar from `tree-sitter-language-pack` (already a dependency) with `get_parser("")`, as `analyzers/dart.py` and `analyzers/go.py` do. + The pack (1.x) downloads a grammar on first use, so also add the name to + the `t.download([...])` list in `docker/Dockerfile`. 2. **Analyzer.** Create `analyzers/.py` with a `TreeSitterAnalyzer` class that extracts components and relationships, plus a module-level `analyze__file(...)` function. diff --git a/pyproject.toml b/pyproject.toml index a407109a..a96d9216 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ dependencies = [ "GitPython>=3.1.40", "Jinja2>=3.1.6", "tree-sitter>=0.25", - "tree-sitter-language-pack>=0.8.0,<0.9.0", + "tree-sitter-language-pack>=1.23.1,<2.0.0", "tree-sitter-python>=0.23.6", "tree-sitter-java>=0.23.5", "tree-sitter-javascript>=0.21.4", diff --git a/requirements.txt b/requirements.txt index 7551bf8a..1d764c05 100644 --- a/requirements.txt +++ b/requirements.txt @@ -148,7 +148,7 @@ tree-sitter-embedded-template==0.23.2 tree-sitter-java==0.23.5 tree-sitter-javascript==0.21.4 tree-sitter-kotlin==1.1.0 -tree-sitter-language-pack==0.8.0 +tree-sitter-language-pack==1.23.1 tree-sitter-python==0.23.6 tree-sitter-ruby==0.23.1 tree-sitter-rust==0.23.2 From 474a1d85efbb24488d6ff5cb1500b1d91df19790 Mon Sep 17 00:00:00 2001 From: Fabricio Ceolin Date: Fri, 6 Nov 2026 08:31:06 -0300 Subject: [PATCH 2/5] deps: stop the analysis once, with a clear error, when a pack grammar cannot be downloaded tree-sitter-language-pack 1.x downloads each grammar on first use. Without network access the first load raises a DownloadError that names a URL but not the cause or the fix, and the per-file except blocks of the analyzers swallow it, so the language drops out of the graph with a traceback per file. `utils/grammars.py` loads pack grammars through `get_pack_parser`, which turns the DownloadError into a GrammarUnavailableError that names the grammar, the cache folder and the `t.download([...])` command. `analyze_code_files` calls `ensure_pack_grammars` before the first file, so the run stops once. `PACK_GRAMMARS_BY_LANGUAGE` lists the grammars each language loads, and a test checks that docker/Dockerfile downloads the same set. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 +- .../analysis/call_graph_analyzer.py | 4 + .../be/dependency_analyzer/analyzers/dart.py | 4 +- .../be/dependency_analyzer/analyzers/go.py | 4 +- .../be/dependency_analyzer/analyzers/vue.py | 4 +- .../be/dependency_analyzer/analyzers/zig.py | 4 +- .../be/dependency_analyzer/utils/grammars.py | 63 +++++++++++++++ guides/development.md | 9 ++- tests/test_pack_grammars.py | 80 +++++++++++++++++++ 9 files changed, 165 insertions(+), 12 deletions(-) create mode 100644 codewiki/src/be/dependency_analyzer/utils/grammars.py create mode 100644 tests/test_pack_grammars.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b8bfa0b8..fc96c028 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -98,7 +98,10 @@ uses [Semantic Versioning](https://semver.org/). others). 1.x downloads a grammar on first use and caches it under `tree_sitter_language_pack.cache_dir()`, so the first run of the Dart, Go, JavaScript/TypeScript-in-Vue, Vue and Zig analyzers needs network access; - the Docker image fetches them at build time. The 1.x Dart grammar parses + the Docker image fetches them at build time. Without network access and + without a cached grammar, the analysis stops before the first file with one + error that names the grammar, the cache folder and the `t.download([...])` + command, instead of a traceback per file. The 1.x Dart grammar parses `Foo(x)` as a call instead of a relational expression, and the Dart analyzer now takes the type-argument edges from that call too. diff --git a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py index c52d090f..58b21b22 100644 --- a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py +++ b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py @@ -22,6 +22,7 @@ is_external_symbol, is_macro_name, ) +from codewiki.src.be.dependency_analyzer.utils.grammars import ensure_pack_grammars from codewiki.src.be.dependency_analyzer.utils.patterns import CODE_EXTENSIONS from codewiki.src.be.dependency_analyzer.utils.security import safe_open_text @@ -96,6 +97,9 @@ def analyze_code_files(self, code_files: list[dict], base_dir: str) -> dict: self.functions = {} self.call_relationships = [] code_files = self._route_contextual_headers(code_files, base_dir) + # Fail once, with a clear message, when a pack grammar cannot be + # downloaded, instead of once per file inside the per-file handlers. + ensure_pack_grammars(f["language"] for f in code_files) self._python_project_modules = self._collect_python_modules(code_files) self._python_external_import_roots = set() self._dart_directives = {} diff --git a/codewiki/src/be/dependency_analyzer/analyzers/dart.py b/codewiki/src/be/dependency_analyzer/analyzers/dart.py index 1b38491c..6af1b0c9 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/dart.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/dart.py @@ -36,7 +36,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.analyzers.dart_flutter import ( COMPOSITION_METHOD_RE, @@ -212,7 +212,7 @@ def _local_declared_types(*roots) -> dict[str, str]: @lru_cache(maxsize=1) def _dart_parser(): - return get_parser("dart") + return get_pack_parser("dart") def _text(node) -> str: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/go.py b/codewiki/src/be/dependency_analyzer/analyzers/go.py index 7a241812..8c0042fa 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/go.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/go.py @@ -48,7 +48,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_go_file @@ -326,7 +326,7 @@ def _default_import_names(import_path: str) -> set[str]: @lru_cache(maxsize=1) def _go_parser(): - return get_parser("go") + return get_pack_parser("go") class TreeSitterGoAnalyzer: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/vue.py b/codewiki/src/be/dependency_analyzer/analyzers/vue.py index 1a17f1ab..4934f72c 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/vue.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/vue.py @@ -37,7 +37,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node @@ -107,7 +107,7 @@ @lru_cache(maxsize=None) def _parser(language: str): - return get_parser(language) + return get_pack_parser(language) def _is_sfc(target: str) -> bool: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/zig.py b/codewiki/src/be/dependency_analyzer/analyzers/zig.py index cd5f1b7c..feb60e2e 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/zig.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/zig.py @@ -47,7 +47,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_zig_file @@ -107,7 +107,7 @@ def _build_zig_modules(repo_path: str) -> dict[str, str]: def _zig_parser(): - return get_parser("zig") + return get_pack_parser("zig") def _alias_chain(var_decl) -> tuple[tuple[str, str], tuple[str, ...]] | None: diff --git a/codewiki/src/be/dependency_analyzer/utils/grammars.py b/codewiki/src/be/dependency_analyzer/utils/grammars.py new file mode 100644 index 00000000..76de670b --- /dev/null +++ b/codewiki/src/be/dependency_analyzer/utils/grammars.py @@ -0,0 +1,63 @@ +"""Parsers from ``tree-sitter-language-pack``. + +Pack 1.x does not ship its grammars in the wheel. It downloads each grammar +the first time it is loaded and caches it under ``cache_dir()``. Without +network access that first load fails with a ``DownloadError`` that names a +URL but not the cause or the fix. The per-file ``except`` blocks of the +analyzers would also swallow it, and the language would silently drop out of +the graph. So the analysis checks the grammars it needs once, before it +parses any file, and stops with one clear message. +""" + +from functools import lru_cache + +import tree_sitter_language_pack as tslp + +# Pack grammars each analyzed language loads. ``docker/Dockerfile`` downloads +# the same set at build time (``tests/test_pack_grammars.py`` checks this). +PACK_GRAMMARS_BY_LANGUAGE: dict[str, tuple[str, ...]] = { + "dart": ("dart",), + "go": ("go",), + "vue": ("vue", "javascript", "typescript"), + "zig": ("zig",), +} + +PACK_GRAMMARS: tuple[str, ...] = tuple( + sorted({g for grammars in PACK_GRAMMARS_BY_LANGUAGE.values() for g in grammars}) +) + + +class GrammarUnavailableError(RuntimeError): + """A pack grammar is not cached and could not be downloaded.""" + + def __init__(self, grammar: str, cause: Exception): + try: + cache = tslp.cache_dir() + except Exception: # noqa: BLE001 — the path only improves the message + cache = "the tree-sitter-language-pack cache" + names = ", ".join(repr(g) for g in PACK_GRAMMARS) + super().__init__( + f"The tree-sitter grammar '{grammar}' is not in {cache} and could not be " + f"downloaded ({cause}). tree-sitter-language-pack downloads each grammar on " + "first use, so the first analysis of these files needs network access. " + "To prepare a machine without network access, run this once on a machine " + "with access and copy the cache folder: " + f'python -c "import tree_sitter_language_pack as t; t.download([{names}])"' + ) + self.grammar = grammar + + +@lru_cache(maxsize=None) +def get_pack_parser(grammar: str): + """Return the pack parser for ``grammar``, with a clear error when it is unavailable.""" + try: + return tslp.get_parser(grammar) + except tslp.DownloadError as e: + raise GrammarUnavailableError(grammar, e) from e + + +def ensure_pack_grammars(languages) -> None: + """Load every pack grammar that ``languages`` need, or raise ``GrammarUnavailableError``.""" + for language in sorted(set(languages)): + for grammar in PACK_GRAMMARS_BY_LANGUAGE.get(language, ()): + get_pack_parser(grammar) diff --git a/guides/development.md b/guides/development.md index 8626e9b5..2c208019 100644 --- a/guides/development.md +++ b/guides/development.md @@ -122,9 +122,12 @@ Analyzers are standalone classes, one file per language. Follow 1. **Grammar.** Add the `tree-sitter-` package to `pyproject.toml` and `requirements.txt`, or take the grammar from `tree-sitter-language-pack` (already a dependency) with - `get_parser("")`, as `analyzers/dart.py` and `analyzers/go.py` do. - The pack (1.x) downloads a grammar on first use, so also add the name to - the `t.download([...])` list in `docker/Dockerfile`. + `get_pack_parser("")` from `utils/grammars.py`, as + `analyzers/dart.py` and `analyzers/go.py` do. Add the language and its + grammars to `PACK_GRAMMARS_BY_LANGUAGE` in the same file. The pack (1.x) downloads a grammar on first + use, so also add the name to the `t.download([...])` list in + `docker/Dockerfile`. `tests/test_pack_grammars.py` checks that the two + lists match. 2. **Analyzer.** Create `analyzers/.py` with a `TreeSitterAnalyzer` class that extracts components and relationships, plus a module-level `analyze__file(...)` function. diff --git a/tests/test_pack_grammars.py b/tests/test_pack_grammars.py new file mode 100644 index 00000000..88803fd9 --- /dev/null +++ b/tests/test_pack_grammars.py @@ -0,0 +1,80 @@ +"""Grammars from tree-sitter-language-pack: a failed download stops the +analysis once, with a clear message, and the Docker image fetches every +grammar the analyzers load.""" + +import re +from pathlib import Path + +import pytest +import tree_sitter_language_pack as tslp + +from codewiki.src.be.dependency_analyzer.analysis.call_graph_analyzer import ( + CallGraphAnalyzer, +) +from codewiki.src.be.dependency_analyzer.utils import grammars +from codewiki.src.be.dependency_analyzer.utils.grammars import ( + PACK_GRAMMARS, + GrammarUnavailableError, + ensure_pack_grammars, + get_pack_parser, +) + +DOCKERFILE = Path(__file__).resolve().parents[1] / "docker" / "Dockerfile" + + +@pytest.fixture +def offline(monkeypatch): + """Every pack grammar load fails as it does without network access.""" + calls = [] + + def fail(name): + calls.append(name) + raise tslp.DownloadError(f"Failed to fetch manifest for {name}: Connection refused") + + get_pack_parser.cache_clear() + monkeypatch.setattr(grammars.tslp, "get_parser", fail) + yield calls + get_pack_parser.cache_clear() + + +def test_download_failure_gives_a_clear_error(offline): + with pytest.raises(GrammarUnavailableError) as info: + get_pack_parser("go") + message = str(info.value) + assert info.value.grammar == "go" + assert "'go'" in message + assert "network access" in message + assert "Connection refused" in message + assert "t.download([" in message + assert isinstance(info.value.__cause__, tslp.DownloadError) + + +def test_only_the_grammars_of_present_languages_load(offline, monkeypatch): + monkeypatch.setattr(grammars.tslp, "get_parser", lambda name: offline.append(name)) + ensure_pack_grammars(["python", "vue", "java", "vue"]) + assert sorted(offline) == ["javascript", "typescript", "vue"] + + +def test_languages_without_pack_grammars_need_no_network(offline): + ensure_pack_grammars(["python", "java", "rust"]) + assert offline == [] + + +def test_analysis_stops_before_the_first_file(offline, tmp_path, monkeypatch): + (tmp_path / "main.go").write_text("package main\n\nfunc main() {}\n", encoding="utf-8") + analyzed = [] + monkeypatch.setattr( + CallGraphAnalyzer, "_analyze_code_file", lambda self, base, info: analyzed.append(info) + ) + with pytest.raises(GrammarUnavailableError, match="'go'"): + CallGraphAnalyzer().analyze_code_files( + [{"path": "main.go", "language": "go"}], str(tmp_path) + ) + assert analyzed == [] + + +def test_dockerfile_downloads_every_pack_grammar(): + match = re.search(r"t\.download\(\[([^\]]*)\]\)", DOCKERFILE.read_text(encoding="utf-8")) + assert match, "docker/Dockerfile must download the pack grammars at build time" + in_dockerfile = sorted(re.findall(r"'([^']+)'", match.group(1))) + assert in_dockerfile == list(PACK_GRAMMARS) From 5e7e863926b7f36283d7dc7a7fcf965e99b4ba74 Mon Sep 17 00:00:00 2001 From: Fabricio Ceolin Date: Fri, 6 Nov 2026 09:15:25 -0300 Subject: [PATCH 3/5] feat: Lean 4 analyzer, with header-scanner structure and exact-name resolution The tree-sitter Lean grammar fails on many commands and notations and gives `end` as a flat node, so the analyzer uses two layers. A header scanner reads declarations, the namespace/section/mutual stack, `open` (also `open X in`), imports, spans and docstrings from the source with comments and strings masked. References come from the tree-sitter identifiers in each declaration, also inside ERROR nodes, without binders, let/have/fun/intro names and match patterns. CallGraphAnalyzer resolves a Lean reference only to an exact fully qualified name, in Lean's order: the declaration's namespaces, the root, then the open namespaces, in the caller's file and the files it imports transitively. Private names stay in their file. No match, or more than one, drops the reference, and a `lean` branch keeps this authoritative like Go and Dart. Constructor and field names link to their structure or inductive, and a dotted name falls back to its longest matching prefix (`origin.x`). Lean is registered with the file discovery, the CLI validators and the MCP tools. `.lake/` and `lake-packages/` are skipped, lakefile.lean/.toml are build artifacts, and lake-manifest.json and lean-toolchain are manifests. The Docker image downloads the `lean` grammar. Refs #154 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 19 + README.md | 4 +- codewiki/cli/utils/repo_validator.py | 1 + codewiki/cli/utils/validation.py | 1 + codewiki/mcp/tools/analysis.py | 1 + .../analysis/analysis_service.py | 2 + .../analysis/call_graph_analyzer.py | 155 +++- .../dependency_analyzer/analyzers/artifact.py | 5 + .../be/dependency_analyzer/analyzers/lean.py | 703 ++++++++++++++++++ .../src/be/dependency_analyzer/ast_parser.py | 1 + .../utils/file_selection.py | 8 + .../be/dependency_analyzer/utils/grammars.py | 1 + .../be/dependency_analyzer/utils/patterns.py | 19 + codewiki/src/be/prompt_template.py | 1 + docker/Dockerfile | 2 +- guides/artifact-aware-generation.md | 4 +- guides/development.md | 4 +- tests/test_artifact_analyzer.py | 5 + tests/test_lean_analyzer.py | 274 +++++++ 19 files changed, 1201 insertions(+), 9 deletions(-) create mode 100644 codewiki/src/be/dependency_analyzer/analyzers/lean.py create mode 100644 tests/test_lean_analyzer.py diff --git a/CHANGELOG.md b/CHANGELOG.md index fc96c028..77adf74a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,25 @@ uses [Semantic Versioning](https://semver.org/). `addModule`; `std`, `builtin`, and third-party packages are dropped. `test` blocks and `zig-cache/`, `.zig-cache/`, `zig-out/` are skipped, `build.zig` is also a build artifact, and `build.zig.zon` is a manifest. +- **Lean 4 analyzer** ([#154](https://github.com/FSoft-AI4Code/CodeWiki/issues/154)). + `def`, `theorem`/`lemma`, `abbrev`, `instance`, `structure`, `class`, + `inductive`, `axiom` and `opaque` declarations become components, keyed by + their fully qualified name (`Geo.Point.add`). A header scanner reads the + structure from the source with comments and strings masked: declarations, + the `namespace`/`section`/`mutual`/`end` stack, `open` (also `open X in`), + `import`, spans and `/-- -/` docstrings. So a region that the tree-sitter + grammar cannot parse does not hide a declaration. References come from the + tree-sitter identifiers, without binders, `let`/`have`/`fun`/`intro` names + and match patterns. A reference resolves only to an exact name in the + declaration's namespaces, the root or the open namespaces, in a file that + the caller imports (transitively); `private` names stay in their file, and + a reference with no match or more than one is dropped. Constructor and + field names (`Point.mk`, `Point.x`, `Shape.dot`) link to their structure or + inductive, and `origin.x` links to `origin`. `structure` is a `struct`, + `class` an `interface`, `inductive` a `class`, and the rest are functions. + `.lake/` and `lake-packages/` are skipped, `lakefile.lean` and + `lakefile.toml` are build artifacts, and `lake-manifest.json` and + `lean-toolchain` are manifests. - **Vue single-file components.** Each `.vue` file becomes a component (`ArticleList.vue::ArticleList`), and its options-API members (`methods`, `computed`, `watch`, lifecycle hooks such as `mounted` or `setup`) become diff --git a/README.md b/README.md index 594eeb31..bab8597e 100644 --- a/README.md +++ b/README.md @@ -120,8 +120,8 @@ Full list in the [CHANGELOG](./CHANGELOG.md). ## Features **Languages.** Python, Java, JavaScript, TypeScript, C, C++, C#, Kotlin, -PHP, Ruby, Scala, Rust, Dart (Flutter-aware), Go, Zig, Vue (single-file -components). +PHP, Ruby, Scala, Rust, Dart (Flutter-aware), Go, Zig, Lean 4, Vue +(single-file components). **Hierarchical decomposition.** The dependency graph is clustered into a module tree, recursively, so a 1.4M-line repository gets the same treatment diff --git a/codewiki/cli/utils/repo_validator.py b/codewiki/cli/utils/repo_validator.py index 3a56587a..ebd93f3f 100644 --- a/codewiki/cli/utils/repo_validator.py +++ b/codewiki/cli/utils/repo_validator.py @@ -37,6 +37,7 @@ ".dart", # Dart ".go", # Go ".zig", # Zig + ".lean", # Lean 4 ".vue", # Vue single-file components } diff --git a/codewiki/cli/utils/validation.py b/codewiki/cli/utils/validation.py index d5bddfd0..82f77b6b 100644 --- a/codewiki/cli/utils/validation.py +++ b/codewiki/cli/utils/validation.py @@ -171,6 +171,7 @@ def detect_supported_languages(directory: Path) -> list[tuple[str, int]]: "Dart": [".dart"], "Go": [".go"], "Zig": [".zig"], + "Lean": [".lean"], "Vue": [".vue"], } diff --git a/codewiki/mcp/tools/analysis.py b/codewiki/mcp/tools/analysis.py index 0d05040d..1a1e7859 100644 --- a/codewiki/mcp/tools/analysis.py +++ b/codewiki/mcp/tools/analysis.py @@ -232,6 +232,7 @@ def _detect_via_mtime( ".dart", ".go", ".zig", + ".lean", ".vue", } diff --git a/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py b/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py index 0be19e62..7b41b7fb 100644 --- a/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py +++ b/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py @@ -365,6 +365,7 @@ def _filter_supported_languages(self, code_files: list[dict]) -> list[dict]: "scala", "dart", "zig", + "lean", "vue", } @@ -392,6 +393,7 @@ def _get_supported_languages(self) -> list[str]: "dart", "go", "zig", + "lean", "vue", ] diff --git a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py index 58b21b22..213720da 100644 --- a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py +++ b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py @@ -28,6 +28,9 @@ logger = logging.getLogger(__name__) +# `_resolve_lean_name` result: no declaration has this name (try a shorter one). +_LEAN_NO_MATCH = object() + class TimeoutError(BaseException): """Raised when file parsing exceeds timeout. @@ -76,6 +79,7 @@ def __init__(self): self._dart_unscoped = False self._go_visible: dict[str, set[str]] = {} self._go_index = None + self._reset_lean_state() self.call_relationships: list[CallRelationship] = [] self._python_project_modules: set = set() self._python_external_import_roots: set = set() @@ -108,6 +112,7 @@ def analyze_code_files(self, code_files: list[dict], base_dir: str) -> dict: self._dart_unscoped = False self._go_visible = {} self._go_index = None + self._reset_lean_state() files_analyzed = 0 files_failed = 0 @@ -293,6 +298,8 @@ def _analyze_code_file(self, repo_dir: str, file_info: dict): self._analyze_go_file(file_path, content, repo_dir) elif language == "zig": self._analyze_zig_file(file_path, content, repo_dir) + elif language == "lean": + self._analyze_lean_file(file_path, content, repo_dir) elif language == "vue": self._analyze_vue_file(file_path, content, repo_dir) # else: @@ -668,6 +675,145 @@ def _analyze_vue_file(self, file_path: str, content: str, repo_dir: str): except Exception: logger.exception(f"Failed to analyze Vue file {file_path}") + def _reset_lean_state(self) -> None: + # Relative path -> imported module names, per Lean file. + self._lean_imports: dict[str, list[str]] = {} + # Node id -> (namespace prefixes innermost first, open namespaces). + self._lean_scopes: dict[str, tuple[list[str], list[str]]] = {} + # Constructor/field FQN -> ids of the structures or inductives it names. + self._lean_aliases: dict[str, list[str]] = defaultdict(list) + self._lean_private: set[str] = set() + self._lean_index: dict[str, list[str]] | None = None + self._lean_modules: dict[str, set[str]] = {} + self._lean_visible: dict[str, frozenset[str]] = {} + + def _analyze_lean_file(self, file_path: str, content: str, repo_dir: str): + """ + Analyze a Lean 4 file. + + Besides components and references, records the file's imports, each + declaration's namespaces and open namespaces, and the constructor and + field names of its structures, for Lean resolution by exact name. + + Args: + file_path: Path to the Lean file + content: File content string + repo_dir: Repository base directory + """ + from codewiki.src.be.dependency_analyzer.analyzers.lean import TreeSitterLeanAnalyzer + + try: + analyzer = TreeSitterLeanAnalyzer(str(file_path), content, repo_path=repo_dir) + + for func in analyzer.nodes: + self.functions[func.id] = func + + self.call_relationships.extend(analyzer.call_relationships) + self._lean_imports[analyzer.relative_path] = analyzer.imports + self._lean_scopes.update(analyzer.scopes) + for alias, node_id in analyzer.aliases.items(): + self._lean_aliases[alias].append(node_id) + self._lean_private.update(analyzer.private_ids) + except Exception: + logger.exception(f"Failed to analyze Lean file {file_path}") + + def _build_lean_index(self) -> dict[str, list[str]]: + """FQN -> Lean node ids, and module name -> files. A module name maps + to every file whose path ends with it (`Geo.Point` -> `src/Geo/Point.lean`), + because the Lake source folder is not known.""" + if self._lean_index is None: + index: dict[str, list[str]] = defaultdict(list) + for func_id, func in self.functions.items(): + if func.language == "lean" and func.qualified_name: + index[func.qualified_name].append(func_id) + for alias, node_ids in self._lean_aliases.items(): + for node_id in node_ids: + if node_id not in index[alias]: + index[alias].append(node_id) + modules: dict[str, set[str]] = defaultdict(set) + for rel in self._lean_imports: + parts = rel[: -len(".lean")].split("/") + for k in range(len(parts)): + modules[".".join(parts[k:])].add(rel) + self._lean_index = index + self._lean_modules = modules + return self._lean_index + + def _lean_visible_files(self, rel: str) -> frozenset[str]: + """The file itself and every repository file it imports, transitively.""" + if rel not in self._lean_visible: + self._build_lean_index() + seen = {rel} + queue = [rel] + while queue: + current = queue.pop() + for module in self._lean_imports.get(current, []): + for target in self._lean_modules.get(module, ()): + if target not in seen: + seen.add(target) + queue.append(target) + self._lean_visible[rel] = frozenset(seen) + return self._lean_visible[rel] + + def _resolve_lean_callee(self, callee: str, caller: Node) -> str | None: + """Resolve a Lean reference to one declaration by exact FQN, in Lean's + order: the caller's namespaces (innermost first), the root, then the + open namespaces. No match, or more than one, drops the reference. + + When the whole dotted name matches nothing, its longest prefix that + matches is the constant and the rest are projections, as in Lean + (`origin.x` is `origin` then the field `x`).""" + parts = callee.split(".") + for end in range(len(parts), 0, -1): + found = self._resolve_lean_name(".".join(parts[:end]), caller) + if found is not _LEAN_NO_MATCH: + return found + return None + + def _resolve_lean_name(self, callee: str, caller: Node): + """One exact name: a node id, None (ambiguous or recursion: stop), or + _LEAN_NO_MATCH (try a shorter prefix).""" + index = self._build_lean_index() + visible = self._lean_visible_files(caller.relative_path) + namespaces, opens = self._lean_scopes.get(caller.id, ([], [])) + + def qualify(prefix: str, name: str) -> str: + return f"{prefix}.{name}" if prefix else name + + def matches(fqn: str) -> list[str]: + return [ + node_id + for node_id in index.get(fqn, []) + if self.functions[node_id].relative_path in visible + and ( + node_id not in self._lean_private + or self.functions[node_id].relative_path == caller.relative_path + ) + ] + + def unique(found) -> str | None: + # A reference to the caller itself is recursion, not an edge. + found = list(found) + return found[0] if len(found) == 1 and found[0] != caller.id else None + + if callee.startswith("_root_."): + found = matches(callee[len("_root_.") :]) + return unique(found) if found else _LEAN_NO_MATCH + + prefixes = [*namespaces, ""] + for prefix in prefixes: + found = matches(qualify(prefix, callee)) + if found: + return unique(found) # the innermost match shadows the outer ones + + opened = { + node_id + for name in opens + for prefix in prefixes + for node_id in matches(qualify(qualify(prefix, name), callee)) + } + return unique(opened) if opened else _LEAN_NO_MATCH + def _analyze_go_file(self, file_path: str, content: str, repo_dir: str): """ Analyze Go file using tree-sitter based analyzer. @@ -860,11 +1006,11 @@ def _resolve_call_relationships(self): self.call_relationships.extend(link_go_interfaces(self.functions)) def _has_authoritative_resolution(self, relationship: CallRelationship) -> bool: - """A Dart or Go caller's callee that the language's scoped resolution + """A Dart, Go or Lean caller's callee that the language's scoped resolution rejected is external or invisible; drop it so ast_parser's name-only fallback cannot bind it to an unrelated component.""" caller = self.functions.get(relationship.caller) - return caller is not None and caller.language in ("dart", "go") + return caller is not None and caller.language in ("dart", "go", "lean") def _dotted_project_packages(self) -> dict[str, set]: """Project packages/namespaces, partitioned by language. Java and C# @@ -1007,6 +1153,9 @@ def _resolve_callee( if caller_language == "go" and caller is not None: # So is Go's: a name resolves only in the package it names. return self._resolve_go_callee(relationship.callee, caller) + if caller_language == "lean" and caller is not None: + # And Lean's: only an exact FQN in a visible file. + return self._resolve_lean_callee(relationship.callee, caller) if lang_indexes: match = self._resolve_callee_in( relationship, lang_indexes["exact"], lang_indexes["simple"] @@ -1152,6 +1301,8 @@ def _generate_visualization_data(self) -> dict: node_classes.append("lang-go") elif file_ext == ".zig": node_classes.append("lang-zig") + elif file_ext == ".lean": + node_classes.append("lang-lean") elif file_ext == ".vue": node_classes.append("lang-vue") diff --git a/codewiki/src/be/dependency_analyzer/analyzers/artifact.py b/codewiki/src/be/dependency_analyzer/analyzers/artifact.py index d7be67a9..e87d4d8b 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/artifact.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/artifact.py @@ -244,6 +244,8 @@ def is_artifact_file_node(node: Any) -> bool: "build.sbt", "Package.swift", "build.zig.zon", + "lake-manifest.json", + "lean-toolchain", "pubspec.yaml", "Pipfile", "environment.yml", @@ -296,6 +298,8 @@ def is_artifact_file_node(node: Any) -> bool: "Gruntfile.js", "Herebyfile.mjs", "build.zig", + "lakefile.lean", + "lakefile.toml", # Autotools inputs; the generated configure, Makefile.in and aclocal.m4 # are not build sources. "configure.ac", @@ -451,6 +455,7 @@ def classify_artifact( ".scala", ".sc", ".zig", + ".lean", ".vue", }: if name in _TEST_INFRA_NAMES: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/lean.py b/codewiki/src/be/dependency_analyzer/analyzers/lean.py new file mode 100644 index 00000000..0bbec7f1 --- /dev/null +++ b/codewiki/src/be/dependency_analyzer/analyzers/lean.py @@ -0,0 +1,703 @@ +"""Lean 4 analyzer: a header scanner for structure, tree-sitter for references. + +The tree-sitter Lean grammar (``lean`` in tree-sitter-language-pack) parses +most declarations, but it fails on many commands and notations (``#eval``, +``¬``, ``←``, ``do`` blocks, user notation), and it gives ``end`` as a flat +node, so the tree does not show namespace nesting. This analyzer therefore +uses two layers: + +1. **Structure, from a header scanner.** Comments and strings are masked, + and the Lean commands are read line by line: declarations (``def``, + ``theorem``/``lemma``, ``abbrev``, ``instance``, ``structure``, ``class``, + ``inductive``, ``axiom``, ``opaque``), the ``namespace``/``section``/ + ``mutual``/``end`` stack, ``open`` (also ``open X in``), ``import``, + ``/-- -/`` docstrings and spans. The scanner does not depend on the + parse tree, so an ``ERROR`` region does not hide a declaration. +2. **References, from tree-sitter.** Every ``identifier`` node in a + declaration's lines is a reference candidate, also inside ``ERROR`` + nodes. Text that an ``ERROR`` node does not cover with child nodes is + tokenized from the masked source. Local names (binders, ``let``, + ``have``, ``fun``, ``intro``, match patterns) are removed, and so is a + dotted name whose first part is local (``p.x``). + +Each declaration is keyed by its fully qualified name (FQN). Structures and +classes also answer to their constructor and field names (``Point.mk``, +``Point.x``), and inductives to their constructors (``Tree.node``). The +analyzer emits each reference as written. ``CallGraphAnalyzer`` resolves it +to an exact FQN only, in Lean's order: the namespaces of the declaration +(innermost first), the root namespace, then the open namespaces. Only the +file itself and the files it imports (transitively) are visible, and a +``private`` name is visible only in its own file. A reference that matches +no declaration, or more than one, is dropped. A bare name is never guessed. + +Known limitations: +- Generalized field notation (``v.norm`` for ``Vec.norm``) needs type + information, so it does not resolve. +- References made through user ``notation``/``macro`` are lost. +- The scanner assumes that each command starts a line, which is the usual + Lean style. A declaration ends at the next command at the same or a + smaller indentation. +- An anonymous ``instance`` gets the synthetic name ``instance_L``, + not the name that Lean generates. ``example`` makes no component. +- Local names are removed by name for the whole declaration, so a global + with the same simple name as a local of that declaration is not linked. +""" + +import bisect +import logging +import os +import re +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import Path + +from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node +from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_lean_file +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser + +logger = logging.getLogger(__name__) + +# `component_type` per declaration keyword; `node_type` keeps the keyword. +COMPONENT_TYPES = { + "structure": "struct", + "class": "interface", + "inductive": "class", + "class inductive": "class", +} + +_IDENT_PART = r"(?:«[^»\n]*»|[^\W\d][\w'!?₀-₉ₐ-ₜ]*)" +IDENT = rf"{_IDENT_PART}(?:\.(?:{_IDENT_PART}|\d+))*" +_IDENT_RE = re.compile(IDENT) + +_MODIFIERS = r"(?:private|protected|noncomputable|partial|unsafe|nonrec|scoped|local|public|meta)" +_ATTRS = r"(?:@\[[^\]\n]*\]\s*)*" +_DECL_RE = re.compile( + rf"^(?P\s*){_ATTRS}(?P(?:{_MODIFIERS}\s+)*)" + r"(?Pclass\s+inductive|def|theorem|lemma|abbrev|instance|structure|class|inductive" + r"|axiom|opaque|example)\b(?P.*)$" +) +_ATTR_ONLY_RE = re.compile(r"^\s*@\[[^\]\n]*\]\s*$") +_IMPORT_RE = re.compile(r"^\s*(?:(?:public|meta|private)\s+)*import\s+(?P.+)$") +_NAMESPACE_RE = re.compile(rf"^\s*namespace\s+(?P{IDENT})") +_SECTION_RE = re.compile(rf"^\s*(?:noncomputable\s+)?section\b\s*(?P{IDENT})?") +_MUTUAL_RE = re.compile(r"^\s*mutual\b") +_END_RE = re.compile(rf"^\s*end\b\s*(?P{IDENT})?\s*$") +_OPEN_RE = re.compile(r"^\s*(?:(?:public|scoped|local)\s+)*(?:open|export)\s+(?P.*)$") +# Commands that end the span of the declaration before them. +_COMMAND_RE = re.compile( + r"^\s*(?:@\[|#|(?:" + r"import|namespace|section|end|open|export|mutual|attribute|variable|universe|set_option" + r"|notation|infix|infixl|infixr|prefix|postfix|macro|macro_rules|syntax|elab|elab_rules" + r"|deriving|initialize|builtin_initialize|declare_syntax_cat|add_decl_doc|omit|include" + r"|run_cmd" + r")\b)" +) +_NAME_AFTER_KIND_RE = re.compile(rf"^\s*(?P{IDENT})") +_INSTANCE_NAME_RE = re.compile(rf"^\s*(?:\(\s*priority\s*:=[^)]*\)\s*)?(?P{IDENT})?\s*") +_EXTENDS_RE = re.compile(r"\bextends\s+(?P[^:=]*?)(?:\bwhere\b|:=|:|$)") +_STRUCT_CTOR_RE = re.compile(rf"^\s*(?P{IDENT})\s*::\s*$") +_FIELD_RE = re.compile(rf"^\s*\(?\s*(?P{IDENT}(?:\s+{IDENT})*)\s*(?::(?!=)|:=)") + +# Local names introduced in a declaration's text. Tree rules cover the parsed +# parts; these also cover `ERROR` regions and tactic blocks. +_LOCAL_RES = [ + re.compile(r"\bfun\s+(?P[^=]*?)=>"), + re.compile(r"(?:∀|∃|∃!|Σ|Π|λ)\s*(?P[^,]*?),"), + re.compile(rf"\blet\s+(?:mut\s+|rec\s+)?(?P{IDENT})"), + re.compile(rf"\bhave\s+(?P{IDENT})\s*:"), + re.compile(rf"\bfor\s+(?P{IDENT})\s+in\b"), + re.compile(rf"[({{⦃\[]\s*(?P{IDENT}(?:\s+{IDENT})*)\s*:(?!=)"), + re.compile(rf"\(\s*(?P{IDENT})\s*:="), + re.compile( + r"\b(?:intro|intros|rintro|rename_i|next|case|obtain|rcases|cases'|induction')\s+(?P[^\n:=]*)" + ), + re.compile(rf"(?P{IDENT})\s*←"), +] + +_KEYWORDS = { + "fun", "let", "have", "show", "from", "by", "do", "then", "else", "if", "match", "with", + "at", "in", "for", "return", "where", "deriving", "mut", "rec", "Type", "Sort", "Prop", + "true", "false", "True", "False", "this", "_", "sorry", "calc", "suffices", "using", +} # fmt: skip + +_LOCAL_BINDER_PARENTS = {"binders"} +_BINDER_TYPES = { + "explicit_binder", + "implicit_binder", + "inst_binder", + "strict_implicit_binder", +} +_INTRO_TACTICS = {"intro", "intros", "rintro", "obtain", "rcases", "rename_i", "next", "case"} + + +@lru_cache(maxsize=1) +def _lean_parser(): + return get_pack_parser("lean") + + +# ---------------------------------------------------------------------- +# Masking +# ---------------------------------------------------------------------- + + +@dataclass +class _Masked: + text: str # same lines and the same UTF-8 byte offsets as the source + docstrings: list = field(default_factory=list) # (end line index, text) + + +_RAW_STRING_RE = re.compile(r'r(#*)"') +_MASK_START_RE = re.compile(r"--|/-|r#*\"|\"|'") +_BLOCK_TOKEN_RE = re.compile(r"/-|-/") +_NOT_NEWLINE_RE = re.compile(rb"[^\n]") +_CHAR_RE = re.compile(r"'(?:\\[^']*|[^\\'\n])'") + + +def _blank(text: str) -> str: + return _NOT_NEWLINE_RE.sub(b" ", text.encode("utf8")).decode("ascii") + + +def mask_source(source: str) -> _Masked: + """Blank comments, strings and char literals with spaces of the same + UTF-8 length, so lines and byte offsets do not move. Docstrings + (``/-- -/``) are kept aside with the line on which they end.""" + out: list[str] = [] + docstrings = [] + i, n = 0, len(source) + line, counted = 0, 0 # newlines before `counted`, so line numbers stay linear + + while i < n: + m = _MASK_START_RE.search(source, i) + if m is None: + out.append(source[i:]) + break + start, token = m.start(), m.group(0) + out.append(source[i:start]) + if token == "--": + j = source.find("\n", start) + j = n if j < 0 else j + elif token == "/-": + depth, j = 1, start + 2 + while depth: + b = _BLOCK_TOKEN_RE.search(source, j) + if b is None: + j = n + break + depth += 1 if b.group(0) == "/-" else -1 + j = b.end() + if source.startswith("/--", start): + line += source.count("\n", counted, j) + counted = j + docstrings.append((line, _clean_docstring(source[start:j]))) + elif token != '"' and _ident_before(source, start): + # `add_comm'` or `r"` after a name: not a literal. + out.append(source[start]) + i = start + 1 + continue + elif token == "'": + c = _CHAR_RE.match(source, start) + if c is None: + out.append(token) + i = start + 1 + continue + j = c.end() + else: + j = _string_end(source, start) + out.append(_blank(source[start:j])) + i = j + return _Masked("".join(out), docstrings) + + +def _ident_before(source: str, i: int) -> bool: + return i > 0 and (source[i - 1].isalnum() or source[i - 1] in "_'!?»") + + +def _string_end(source: str, i: int) -> int: + n = len(source) + if source[i] == "r": + hashes = len(_RAW_STRING_RE.match(source, i).group(1)) + close = '"' + "#" * hashes + j = source.find(close, i + 2 + hashes) + return n if j < 0 else j + len(close) + j = i + 1 + while j < n: + if source[j] == "\\": + j += 2 + elif source[j] == '"': + return j + 1 + else: + j += 1 + return n + + +def _clean_docstring(block: str) -> str: + body = block[3:-2] if block.endswith("-/") else block[3:] + return "\n".join(line.strip() for line in body.strip().splitlines()).strip() + + +# ---------------------------------------------------------------------- +# Header scanner +# ---------------------------------------------------------------------- + + +@dataclass +class LeanDecl: + fqn: str + kind: str + start: int # 0-based line of the first attribute or keyword line + keyword_line: int + end: int # 0-based, inclusive + indent: int + private: bool + namespaces: list[str] # the declaration's namespace prefixes, innermost first + opens: list[str] + docstring: str = "" + aliases: list[str] = field(default_factory=list) # constructor and field FQNs + bases: list[str] = field(default_factory=list) + + +@dataclass +class _Scope: + kind: str # "namespace", "section" or "mutual" + name: str + opens: list[str] = field(default_factory=list) + + +def _qualify(prefix: str, name: str) -> str: + return f"{prefix}.{name}" if prefix else name + + +def _strip_guillemets(name: str) -> str: + return name.replace("«", "").replace("»", "") + + +def _parse_open_names(body: str) -> list[str]: + """`open A B (x y) hiding z renaming p → q` -> ["A", "B"].""" + body = re.split(r"\b(?:hiding|renaming)\b", body)[0] + body = re.sub(r"\([^)]*\)", " ", body) + return [_strip_guillemets(m.group(0)) for m in _IDENT_RE.finditer(body)] + + +def scan_headers(masked: _Masked) -> tuple[list[LeanDecl], list[str]]: + """Return the declarations and the imported module names of a file.""" + lines = masked.text.split("\n") + docs = {end: text for end, text in masked.docstrings} + decls: list[LeanDecl] = [] + imports: list[str] = [] + stack: list[_Scope] = [] + file_opens: list[str] = [] + pending_open_in: list[str] = [] + attr_start: int | None = None + + def namespace() -> str: + return ".".join(s.name for s in stack if s.kind == "namespace") + + def active_opens() -> list[str]: + return file_opens + [o for s in stack for o in s.opens] + + def close_previous(line_idx: int, indent: int) -> None: + if decls and decls[-1].end < 0 and indent <= decls[-1].indent: + end = line_idx - 1 + while end > decls[-1].keyword_line and not lines[end].strip(): + end -= 1 + decls[-1].end = end + + for idx, line in enumerate(lines): + if not line.strip(): + continue + indent = len(line) - len(line.lstrip()) + if decls and decls[-1].end < 0 and indent > decls[-1].indent: + continue # a body line of the open declaration, even if it looks like a command + decl_m = _DECL_RE.match(line) + if decl_m is None and not _COMMAND_RE.match(line): + continue # a continuation line at the declaration's indentation + close_previous(idx, indent) + + if _ATTR_ONLY_RE.match(line): + if attr_start is None: + attr_start = idx + continue + start = attr_start if attr_start is not None else idx + attr_start = None + + if decl_m is not None: + decl = _decl_from_match(decl_m, idx, start, namespace(), pending_open_in) + pending_open_in = [] + if decl is None: + continue + decl.opens = active_opens() + decl.opens + decl.docstring = _docstring_before(docs, lines, start) + decls.append(decl) + continue + + # `open X in` covers only the next command, whatever it is. + pending_open_in = [] + if m := _IMPORT_RE.match(line): + imports.extend( + _strip_guillemets(x.group(0)) for x in _IDENT_RE.finditer(m.group("mods")) + ) + elif m := _NAMESPACE_RE.match(line): + stack.append(_Scope("namespace", _strip_guillemets(m.group("name")))) + elif _MUTUAL_RE.match(line): + stack.append(_Scope("mutual", "")) + elif m := _END_RE.match(line): + if stack: + stack.pop() + elif m := _SECTION_RE.match(line): + stack.append(_Scope("section", m.group("name") or "")) + elif m := _OPEN_RE.match(line): + body = m.group("body") + parts = re.split(r"\bin\b", body, maxsplit=1) + names = _parse_open_names(parts[0]) + if len(parts) == 2: + pending_open_in = names + rest = parts[1] + if rest.strip() and (inner := _DECL_RE.match(" " * indent + rest.strip())): + decl = _decl_from_match(inner, idx, start, namespace(), pending_open_in) + pending_open_in = [] + if decl is not None: + decl.opens = active_opens() + decl.opens + decl.docstring = _docstring_before(docs, lines, start) + decls.append(decl) + elif stack: + stack[-1].opens.extend(names) + else: + file_opens.extend(names) + + if decls and decls[-1].end < 0: + end = len(lines) - 1 + while end > decls[-1].keyword_line and not lines[end].strip(): + end -= 1 + decls[-1].end = end + for decl in decls: + if decl.kind in ("structure", "class", "inductive", "class inductive"): + decl.aliases = _member_aliases(decl, lines) + return decls, imports + + +def _decl_from_match(m, idx, start, ns, open_in) -> LeanDecl | None: + kind = re.sub(r"\s+", " ", m.group("kind")) + rest = m.group("rest") + if kind == "example": + name = None + elif kind == "instance": + nm = _INSTANCE_NAME_RE.match(rest) + name = nm.group("name") if nm else None + if name in _KEYWORDS: + name = None + if name is None: + name = f"instance_L{idx + 1}" + else: + nm = _NAME_AFTER_KIND_RE.match(rest) + name = nm.group("name") if nm else None + if not name: + # `example` and unnamed forms still end the previous declaration. + return None + name = _strip_guillemets(name) + if name.startswith("_root_."): + fqn = name[len("_root_.") :] + else: + fqn = _qualify(ns, name) + parts = fqn.split(".") + namespaces = [".".join(parts[:k]) for k in range(len(parts) - 1, 0, -1)] + bases = [] + if kind in ("structure", "class") and (em := _EXTENDS_RE.search(rest)): + bases = [_strip_guillemets(b.group(0)) for b in _IDENT_RE.finditer(em.group("bases"))] + return LeanDecl( + fqn=fqn, + kind=kind, + start=start, + keyword_line=idx, + end=-1, + indent=len(m.group("indent")), + private="private" in m.group("mods").split(), + namespaces=namespaces, + opens=list(open_in), + bases=bases, + ) + + +def _docstring_before(docs: dict, lines: list[str], start: int) -> str: + line = start - 1 + while line >= 0 and not lines[line].strip(): + if line in docs: + return docs[line] + line -= 1 + return docs.get(line, "") if line >= 0 else "" + + +def _member_aliases(decl: LeanDecl, lines: list[str]) -> list[str]: + aliases: list[str] = [] + body = lines[decl.keyword_line + 1 : decl.end + 1] + header = lines[decl.keyword_line] + if decl.kind in ("inductive", "class inductive"): + for line in [header.split(" where", 1)[-1], *body]: + for chunk in line.split("|")[1:]: + if m := _IDENT_RE.match(chunk.strip()): + aliases.append(_qualify(decl.fqn, _strip_guillemets(m.group(0)))) + return aliases + ctor = "mk" + if " where" in header or header.rstrip().endswith("where"): + for line in body: + if line.strip().startswith("deriving"): + break + if m := _STRUCT_CTOR_RE.match(line): + ctor = _strip_guillemets(m.group("name")) + elif m := _FIELD_RE.match(line): + for name in m.group("names").split(): + aliases.append(_qualify(decl.fqn, _strip_guillemets(name))) + aliases.append(_qualify(decl.fqn, ctor)) + return aliases + + +# ---------------------------------------------------------------------- +# References +# ---------------------------------------------------------------------- + + +def _node_text(node) -> str: + return node.text.decode("utf8", errors="replace") + + +def _collect_tree_names(root, masked_bytes: bytes): + """Return (references, locals): references as (line, name); locals as + (line, simple name).""" + refs: list[tuple[int, str]] = [] + locals_: list[tuple[int, str]] = [] + line_starts = [0] + [m.end() for m in re.finditer(rb"\n", masked_bytes)] + skip: set[int] = set() # ids of identifier nodes that are not references + + def mark_local(node) -> None: + for ident in _identifiers(node): + skip.add(ident.id) + text = _node_text(ident) + if "." not in text: + locals_.append((ident.start_point[0], text)) + + stack = [root] + while stack: + node = stack.pop() + t = node.type + children = node.children + if t in _LOCAL_BINDER_PARENTS: + for child in children: + if child.type == "identifier": + mark_local(child) + elif t in _BINDER_TYPES: + for child in children: + if child.type == ":": + break + if child.type == "identifier": + mark_local(child) + elif t in ("let", "have"): + for child in children[1:]: + if child.type == "identifier": + mark_local(child) + break + if child.type in (":=", ":", "←"): + break + elif t == "match_alt": + for child in children: + if child.type == "=>": + break + for ident in _identifiers(child): + if "." not in _node_text(ident): + mark_local(ident) + elif t in ("struct_field", "field"): + for child in children: + if child.type in (":=", ":"): + break + if child.type == "identifier": + skip.add(child.id) + elif t == "app" and children and children[0].type == "identifier": + if _node_text(children[0]) in _INTRO_TACTICS: + for child in children[1:]: + mark_local(child) + elif t == "ERROR": + covered = 0 + for child in children: + if child.start_byte > covered: + _tokens_in( + masked_bytes, + line_starts, + max(covered, node.start_byte), + child.start_byte, + refs, + ) + covered = max(covered, child.end_byte) + if covered < node.end_byte: + _tokens_in( + masked_bytes, line_starts, max(covered, node.start_byte), node.end_byte, refs + ) + if t == "identifier" and node.id not in skip: + refs.append((node.start_point[0], _node_text(node))) + stack.extend(reversed(children)) + return refs, locals_ + + +def _identifiers(node): + if node.type == "identifier": + yield node + return + for child in node.children: + yield from _identifiers(child) + + +def _tokens_in( + masked_bytes: bytes, line_starts: list[int], start: int, end: int, out: list +) -> None: + if end <= start: + return + chunk = masked_bytes[start:end].decode("utf8", errors="replace") + base_line = bisect.bisect_right(line_starts, start) - 1 + for m in _IDENT_RE.finditer(chunk): + out.append((base_line + chunk.count("\n", 0, m.start()), m.group(0))) + + +def _text_locals(text: str) -> set[str]: + names: set[str] = set() + for regex in _LOCAL_RES: + for m in regex.finditer(text): + for ident in _IDENT_RE.finditer(m.group("names")): + if "." not in ident.group(0): + names.add(ident.group(0)) + return names - _KEYWORDS + + +def normalize_reference(name: str) -> str | None: + """Strip `@`, guillemets and numeric projections; drop `.foo` and keywords.""" + name = name.lstrip("@") + if not name or name.startswith("."): + return None + parts = [p for p in _strip_guillemets(name).split(".") if p] + while parts and parts[-1].isdigit(): + parts.pop() + if not parts or parts[0] in _KEYWORDS: + return None + return ".".join(parts) + + +# ---------------------------------------------------------------------- +# Analyzer +# ---------------------------------------------------------------------- + + +class TreeSitterLeanAnalyzer: + def __init__(self, file_path: str, content: str, repo_path: str | None = None): + self.file_path = Path(file_path) + self.content = content + self.repo_path = repo_path or "" + self.nodes: list[Node] = [] + self.call_relationships: list[CallRelationship] = [] + self.imports: list[str] = [] + # node id -> (namespace prefixes innermost first, open namespaces) + self.scopes: dict[str, tuple[list[str], list[str]]] = {} + # constructor/field FQN -> node id of the structure or inductive + self.aliases: dict[str, str] = {} + self.private_ids: set[str] = set() + self.relative_path = Path(self._relative_path()).as_posix() + if is_skipped_lean_file(self.relative_path): + return + try: + self._analyze() + except Exception as e: # noqa: BLE001 — a broken file must not abort the sweep + logger.error(f"Error parsing Lean file {self.file_path}: {e}") + + def _relative_path(self) -> str: + if self.repo_path: + try: + return os.path.relpath(str(self.file_path), self.repo_path) + except ValueError: + return str(self.file_path) + return str(self.file_path) + + def _component_id(self, fqn: str) -> str: + return f"{self.relative_path}::{fqn}" + + def _analyze(self) -> None: + masked = mask_source(self.content) + decls, self.imports = scan_headers(masked) + source_lines = self.content.split("\n") + by_fqn: dict[str, Node] = {} + owners: list[tuple[LeanDecl, Node]] = [] + for decl in decls: + if decl.fqn in by_fqn: + continue + node = self._add_node(decl, source_lines) + by_fqn[decl.fqn] = node + owners.append((decl, node)) + for alias in decl.aliases: + self.aliases.setdefault(alias, node.id) + if not owners: + return + + tree = _lean_parser().parse(self.content.encode("utf8")) + masked_bytes = masked.text.encode("utf8") + refs, tree_locals = _collect_tree_names(tree.root_node, masked_bytes) + line_owner: dict[int, tuple[LeanDecl, Node]] = {} + for decl, node in owners: + for line in range(decl.keyword_line, decl.end + 1): + line_owner.setdefault(line, (decl, node)) + + locals_by_decl: dict[str, set[str]] = {} + masked_lines = masked.text.split("\n") + for decl, node in owners: + text = "\n".join(masked_lines[decl.keyword_line : decl.end + 1]) + locals_by_decl[node.id] = _text_locals(text) + for line, name in tree_locals: + if line in line_owner: + locals_by_decl[line_owner[line][1].id].add(name) + + seen: set[tuple[str, str]] = set() + for line, raw in refs: + owner = line_owner.get(line) + if owner is None: + continue + decl, node = owner + name = normalize_reference(raw) + if name is None or name.split(".")[0] in locals_by_decl[node.id]: + continue + if line == decl.keyword_line and name == decl.fqn.split(".")[-1]: + continue # the declaration's own name + key = (node.id, name) + if key in seen: + continue + seen.add(key) + self.call_relationships.append( + CallRelationship(caller=node.id, callee=name, call_line=line + 1) + ) + + def _add_node(self, decl: LeanDecl, source_lines: list[str]) -> Node: + component_id = self._component_id(decl.fqn) + component_type = COMPONENT_TYPES.get(decl.kind, "function") + node_type = "theorem" if decl.kind == "lemma" else decl.kind + node = Node( + id=component_id, + name=decl.fqn, + component_type=component_type, + file_path=str(self.file_path), + relative_path=self.relative_path, + source_code="\n".join(source_lines[decl.start : decl.end + 1]), + start_line=decl.start + 1, + end_line=decl.end + 1, + has_docstring=bool(decl.docstring), + docstring=decl.docstring, + node_type=node_type, + base_classes=decl.bases or None, + display_name=f"{node_type} {decl.fqn}", + component_id=component_id, + language="lean", + qualified_name=decl.fqn, + ) + self.nodes.append(node) + self.scopes[component_id] = (decl.namespaces, decl.opens) + if decl.private: + self.private_ids.add(component_id) + return node + + +def analyze_lean_file( + file_path: str, content: str, repo_path: str | None = None +) -> tuple[list[Node], list[CallRelationship]]: + analyzer = TreeSitterLeanAnalyzer(file_path, content, repo_path) + return analyzer.nodes, analyzer.call_relationships diff --git a/codewiki/src/be/dependency_analyzer/ast_parser.py b/codewiki/src/be/dependency_analyzer/ast_parser.py index e5f67c40..8e626112 100644 --- a/codewiki/src/be/dependency_analyzer/ast_parser.py +++ b/codewiki/src/be/dependency_analyzer/ast_parser.py @@ -230,6 +230,7 @@ def _file_to_module_path(self, file_path: str) -> str: ".dart", ".go", ".zig", + ".lean", ".vue", ] for ext in extensions: diff --git a/codewiki/src/be/dependency_analyzer/utils/file_selection.py b/codewiki/src/be/dependency_analyzer/utils/file_selection.py index 02fa5942..6793a970 100644 --- a/codewiki/src/be/dependency_analyzer/utils/file_selection.py +++ b/codewiki/src/be/dependency_analyzer/utils/file_selection.py @@ -5,6 +5,7 @@ PHP_TEMPLATE_PATTERNS = {".blade.php", ".phtml", ".twig.php"} PHP_TEMPLATE_DIRECTORIES = {"views", "templates", "resources/views"} ZIG_SKIPPED_DIRS = {"zig-cache", ".zig-cache", "zig-out"} +LEAN_SKIPPED_DIRS = {".lake", "lake-packages"} def is_skipped_go_file(relative_path: str) -> bool: @@ -17,6 +18,11 @@ def is_skipped_zig_file(relative_path: str) -> bool: return any(part in ZIG_SKIPPED_DIRS for part in Path(relative_path).parts[:-1]) +def is_skipped_lean_file(relative_path: str) -> bool: + """Lake's build folder and its package checkouts are not project code.""" + return any(part in LEAN_SKIPPED_DIRS for part in Path(relative_path).parts[:-1]) + + def is_php_template_file(file_path: str) -> bool: file_str = str(file_path) return any(file_str.endswith(pattern) for pattern in PHP_TEMPLATE_PATTERNS) or any( @@ -32,6 +38,8 @@ def is_skipped_source_file(file_info: dict, base_dir: str) -> bool: return is_skipped_go_file(file_info["path"]) if language == "zig": return is_skipped_zig_file(file_info["path"]) + if language == "lean": + return is_skipped_lean_file(file_info["path"]) if language == "php": return is_php_template_file(str(Path(base_dir) / file_info["path"])) return False diff --git a/codewiki/src/be/dependency_analyzer/utils/grammars.py b/codewiki/src/be/dependency_analyzer/utils/grammars.py index 76de670b..bdbae313 100644 --- a/codewiki/src/be/dependency_analyzer/utils/grammars.py +++ b/codewiki/src/be/dependency_analyzer/utils/grammars.py @@ -18,6 +18,7 @@ PACK_GRAMMARS_BY_LANGUAGE: dict[str, tuple[str, ...]] = { "dart": ("dart",), "go": ("go",), + "lean": ("lean",), "vue": ("vue", "javascript", "typescript"), "zig": ("zig",), } diff --git a/codewiki/src/be/dependency_analyzer/utils/patterns.py b/codewiki/src/be/dependency_analyzer/utils/patterns.py index 6692df36..84c4ffe2 100644 --- a/codewiki/src/be/dependency_analyzer/utils/patterns.py +++ b/codewiki/src/be/dependency_analyzer/utils/patterns.py @@ -70,6 +70,11 @@ "zig-cache/", "zig-out", "zig-out/", + # Lean: Lake build folder and package checkouts + ".lake", + ".lake/", + "lake-packages", + "lake-packages/", # IDEs and editors / Java ".project", # C/C++ @@ -184,6 +189,7 @@ "*.sc", "*.dart", "*.zig", + "*.lean", "*.vue", "*.clj", "*.hs", @@ -250,6 +256,9 @@ "Procfile", "go.mod", "build.zig.zon", + "lakefile.toml", + "lake-manifest.json", + "lean-toolchain", ".editorconfig", ".pre-commit-config.yaml", ".eslintrc*", @@ -321,6 +330,7 @@ ".dart": "dart", ".cs": "csharp", ".zig": "zig", + ".lean": "lean", ".vue": "vue", } @@ -556,6 +566,15 @@ "const {name} = enum", "const {name} = union", ], + "lean": [ + "def {name}", + "theorem {name}", + "lemma {name}", + "abbrev {name}", + "structure {name}", + "class {name}", + "inductive {name}", + ], "c": ["void {name}", "int {name}", "{name}("], "cpp": ["void {name}", "int {name}", "{name}("], "php": [ diff --git a/codewiki/src/be/prompt_template.py b/codewiki/src/be/prompt_template.py index db16a55d..f649cead 100644 --- a/codewiki/src/be/prompt_template.py +++ b/codewiki/src/be/prompt_template.py @@ -400,6 +400,7 @@ ".dart": "dart", ".go": "go", ".zig": "zig", + ".lean": "lean", ".vue": "vue", ".php": "php", ".phtml": "php", diff --git a/docker/Dockerfile b/docker/Dockerfile index c6b9b086..38fca3c2 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -20,7 +20,7 @@ RUN pip install --no-cache-dir -r requirements.txt # tree-sitter-language-pack 1.x downloads each grammar on first use. Fetch the # ones the analyzers load at build time, so the image runs offline. -RUN python -c "import tree_sitter_language_pack as t; t.download(['dart', 'go', 'javascript', 'typescript', 'vue', 'zig'])" +RUN python -c "import tree_sitter_language_pack as t; t.download(['dart', 'go', 'javascript', 'lean', 'typescript', 'vue', 'zig'])" # Copy application code COPY codewiki ./codewiki diff --git a/guides/artifact-aware-generation.md b/guides/artifact-aware-generation.md index 90db8486..17d7212b 100644 --- a/guides/artifact-aware-generation.md +++ b/guides/artifact-aware-generation.md @@ -20,8 +20,8 @@ Files are classified by name, extension, and directory. Nine classes: | Class | Examples | | --- | --- | -| `manifest` | `package.json`, `pyproject.toml`, `setup.py`, `setup.cfg`, `Cargo.toml`, `go.mod`, `build.zig.zon`, `Gemfile`, `pnpm-workspace.yaml` | -| `build` | `Makefile`, `CMakeLists.txt`, `configure.ac`, `Makefile.am`, `m4/*.m4`, `build.zig`, `*.gradle`, `BUILD.gn`, `*.gni`, `DEPS`, `Rakefile`, `rakelib/*.rake`, `build/` | +| `manifest` | `package.json`, `pyproject.toml`, `setup.py`, `setup.cfg`, `Cargo.toml`, `go.mod`, `build.zig.zon`, `lake-manifest.json`, `lean-toolchain`, `Gemfile`, `pnpm-workspace.yaml` | +| `build` | `Makefile`, `CMakeLists.txt`, `configure.ac`, `Makefile.am`, `m4/*.m4`, `build.zig`, `lakefile.lean`, `lakefile.toml`, `*.gradle`, `BUILD.gn`, `*.gni`, `DEPS`, `Rakefile`, `rakelib/*.rake`, `build/` | | `container` | `Dockerfile*`, `Containerfile`, `docker-compose*.yml`, files under `docker/` | | `ci` | `.github/workflows/*.yml`, `.github/actions/*/action.yml`, `.gitlab-ci.yml`, `Jenkinsfile`, `ci/`, `.circleci/` | | `packaging` | `debian/`, `rpm/`, `pkg/`, `packages/`, `*.spec`, `*.service`, install and lifecycle scripts | diff --git a/guides/development.md b/guides/development.md index 2c208019..73a1bcd5 100644 --- a/guides/development.md +++ b/guides/development.md @@ -143,8 +143,8 @@ Analyzers are standalone classes, one file per language. Follow snippets that check component and edge extraction. See `tests/test_scala_analyzer.py`, `tests/test_ruby_analyzer.py`, `tests/test_rust_analyzer.py`, `tests/test_dart_analyzer.py`, - `tests/test_go_analyzer.py`, `tests/test_zig_analyzer.py`, and - `tests/test_vue_analyzer.py`. + `tests/test_go_analyzer.py`, `tests/test_zig_analyzer.py`, + `tests/test_lean_analyzer.py`, and `tests/test_vue_analyzer.py`. 6. **Docs.** Add the language to the README list. ## Tests and lint diff --git a/tests/test_artifact_analyzer.py b/tests/test_artifact_analyzer.py index 6ba521a2..8c73e8bf 100644 --- a/tests/test_artifact_analyzer.py +++ b/tests/test_artifact_analyzer.py @@ -130,6 +130,11 @@ def _artifact_nodes(components: dict[str, Node]) -> dict[str, Node]: ("build.zig", "build"), ("build.zig.zon", "manifest"), ("src/main.zig", None), + ("lakefile.lean", "build"), + ("lakefile.toml", "build"), + ("lake-manifest.json", "manifest"), + ("lean-toolchain", "manifest"), + ("Geo/Point.lean", None), ("src/App.vue", None), ("pytest.ini", "test_infra"), ("conftest.py", "test_infra"), diff --git a/tests/test_lean_analyzer.py b/tests/test_lean_analyzer.py new file mode 100644 index 00000000..096437f5 --- /dev/null +++ b/tests/test_lean_analyzer.py @@ -0,0 +1,274 @@ +"""Tests for the Lean 4 analyzer: header-scanner structure, tree-sitter +references, and exact-name resolution in CallGraphAnalyzer.""" + +from pathlib import Path + +import pytest + +pytest.importorskip("tree_sitter_language_pack") + +from codewiki.src.be.dependency_analyzer.analysis.call_graph_analyzer import ( + CallGraphAnalyzer, +) +from codewiki.src.be.dependency_analyzer.analyzers.lean import ( + analyze_lean_file, + mask_source, + normalize_reference, + scan_headers, +) +from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_lean_file + +POINT = """\ +/-! Geometry primitives. -/ +namespace Geo + +/-- A point in the plane. +Coordinates are naturals. -/ +structure Point where + x : Nat + y : Nat + deriving Repr + +def Point.add (p q : Point) : Point := + let s := p.x + q.x + { x := s, y := p.y + q.y } + +@[simp] +theorem Point.add_zero (p : Point) : Point.add p ⟨0, 0⟩ = p := by + cases p + simp [Point.add, Nat.add_zero] + +inductive Shape where + | dot (p : Point) + | segment (a b : Point) + +def Shape.start : Shape → Point + | .dot p => p + | .segment a _ => a + +private def secret : Nat := 1 + +instance : Add Point := ⟨Point.add⟩ + +section Ops +open Point in +def double (p : Point) : Point := add p p +end Ops + +end Geo +""" + +USE = """\ +import Geo.Point + +open Geo + +def origin : Point := Point.mk 0 0 + +def first (s : Shape) : Point := Shape.start s + +def shapes : List Shape := [Shape.dot origin, .segment origin origin] + +def leak : Nat := secret + +lemma origin_x : origin.x = 0 := rfl + +def report : IO Unit := do + let p ← pure origin + IO.println s!"{p.x}" + #eval first (Shape.dot p) + +def after : Point := double origin + +def external := Nat.add_comm 1 2 +""" + +ISLAND = """\ +def lonely : Nat := Geo.secret + +def notImported := Geo.Point.add +""" + + +def _write(root: Path, rel: str, text: str) -> str: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return rel + + +@pytest.fixture +def lean_repo(tmp_path: Path): + files = [ + _write(tmp_path, "src/Geo/Point.lean", POINT), + _write(tmp_path, "src/Geo/Use.lean", USE), + _write(tmp_path, "src/Island.lean", ISLAND), + _write(tmp_path, ".lake/packages/dep/Dep.lean", "def dep := 1\n"), + ] + analyzer = CallGraphAnalyzer() + analyzer.analyze_code_files([{"path": f, "language": "lean"} for f in files], str(tmp_path)) + edges = {(r.caller, r.callee) for r in analyzer.call_relationships} + return analyzer.functions, edges + + +P = "src/Geo/Point.lean::" +U = "src/Geo/Use.lean::" +I = "src/Island.lean::" # noqa: E741 + + +def test_declarations_namespaces_and_types(lean_repo) -> None: + functions, _ = lean_repo + kinds = {f.qualified_name: (f.component_type, f.node_type) for f in functions.values()} + assert kinds["Geo.Point"] == ("struct", "structure") + assert kinds["Geo.Point.add"] == ("function", "def") + assert kinds["Geo.Point.add_zero"] == ("function", "theorem") + assert kinds["Geo.Shape"] == ("class", "inductive") + assert kinds["Geo.double"] == ("function", "def") + assert kinds["origin_x"] == ("function", "theorem") # `lemma` is a theorem + assert any(name.startswith("Geo.instance_L") for name in kinds) + assert all(f.language == "lean" for f in functions.values()) + + +def test_spans_docstrings_and_attributes(lean_repo) -> None: + functions, _ = lean_repo + point = functions[P + "Geo.Point"] + assert point.docstring == "A point in the plane.\nCoordinates are naturals." + assert (point.start_line, point.end_line) == (6, 9) + theorem = functions[P + "Geo.Point.add_zero"] + assert theorem.start_line == 15 # the `@[simp]` line + assert "simp [Point.add" in (theorem.source_code or "") + assert not functions[P + "Geo.Point.add"].docstring + + +def test_lake_folders_are_skipped(lean_repo) -> None: + functions, _ = lean_repo + assert not any(".lake" in f.relative_path for f in functions.values()) + assert is_skipped_lean_file(".lake/packages/dep/Dep.lean") + assert is_skipped_lean_file("lake-packages/std/Std.lean") + assert not is_skipped_lean_file("src/Geo/Point.lean") + + +def test_same_file_references_resolve_in_the_namespace(lean_repo) -> None: + _, edges = lean_repo + assert (P + "Geo.Point.add", P + "Geo.Point") in edges + assert (P + "Geo.Point.add_zero", P + "Geo.Point.add") in edges + assert (P + "Geo.Shape", P + "Geo.Point") in edges # constructor argument types + assert (P + "Geo.Shape.start", P + "Geo.Shape") in edges + + +def test_open_in_applies_to_the_next_declaration(lean_repo) -> None: + functions, edges = lean_repo + assert (P + "Geo.double", P + "Geo.Point.add") in edges # `add` through `open Point in` + instance = next(i for i in functions if i.startswith(P + "Geo.instance_L")) + assert (instance, P + "Geo.Point.add") in edges + + +def test_imports_open_and_member_aliases(lean_repo) -> None: + _, edges = lean_repo + assert (U + "origin", P + "Geo.Point") in edges # `Point.mk` is the structure + assert (U + "first", P + "Geo.Shape.start") in edges + assert (U + "shapes", P + "Geo.Shape") in edges # `Shape.dot` is the inductive + assert (U + "shapes", U + "origin") in edges + assert (U + "origin_x", U + "origin") in edges + assert (U + "after", P + "Geo.double") in edges + + +def test_error_regions_still_give_references(lean_repo) -> None: + _, edges = lean_repo + assert (U + "report", U + "origin") in edges + assert (U + "report", U + "first") in edges + assert (U + "report", P + "Geo.Shape") in edges + + +def test_locals_externals_and_recursion_give_no_edges(lean_repo) -> None: + functions, edges = lean_repo + callees = {callee for caller, callee in edges if caller == P + "Geo.Point.add"} + assert callees == {P + "Geo.Point"} # p, q, s, p.x are locals + assert not any(caller == U + "external" for caller, _ in edges) # Nat.add_comm + assert all(caller != callee for caller, callee in edges) + assert all(callee in functions for _, callee in edges) + + +def test_private_and_unimported_names_are_invisible(lean_repo) -> None: + _, edges = lean_repo + assert not any(caller == U + "leak" for caller, _ in edges) # private in another file + assert not any(caller.startswith(I) for caller, _ in edges) # Island imports nothing + + +def test_ambiguous_name_is_dropped(tmp_path: Path) -> None: + files = [ + _write(tmp_path, "A.lean", "namespace X\ndef f := 1\nend X\n"), + _write(tmp_path, "B.lean", "namespace Y\ndef f := 2\nend Y\n"), + _write(tmp_path, "C.lean", "import A\nimport B\nopen X Y\ndef g := f\ndef h := X.f\n"), + ] + analyzer = CallGraphAnalyzer() + analyzer.analyze_code_files([{"path": f, "language": "lean"} for f in files], str(tmp_path)) + edges = {(r.caller, r.callee) for r in analyzer.call_relationships} + assert not any(caller == "C.lean::g" for caller, _ in edges) + assert ("C.lean::h", "A.lean::X.f") in edges + + +def test_root_prefix_mutual_and_nested_namespaces(tmp_path: Path) -> None: + source = """\ +namespace A.B +def _root_.top := 1 +mutual + def even : Nat → Bool + | 0 => true + | n + 1 => odd n + def odd : Nat → Bool + | 0 => false + | n + 1 => even n +end +end A.B +def last := top +""" + nodes, _ = analyze_lean_file(str(tmp_path / "M.lean"), source, repo_path=str(tmp_path)) + names = {n.qualified_name for n in nodes} + assert names == {"top", "A.B.even", "A.B.odd", "last"} + + +def test_masking_keeps_lines_and_byte_offsets() -> None: + source = 'def a := "-- not a comment" -- é comment\n/- block /- nested -/ -/ def b := \'x\'\n' + masked = mask_source(source) + assert masked.text.count("\n") == source.count("\n") + assert len(masked.text.encode("utf8")) == len(source.encode("utf8")) + assert "not a comment" not in masked.text and "nested" not in masked.text + assert "def b" in masked.text + + +def test_interpolated_strings_and_primes_are_masked_correctly() -> None: + masked = mask_source('def c := s!"x {y} -- z"\ndef d\' := r#"q"# ++ a\'\n') + assert "-- z" not in masked.text and '"q"' not in masked.text + assert "def d' :=" in masked.text and "a'" in masked.text + + +def test_header_scanner_without_the_parse_tree() -> None: + decls, imports = scan_headers( + mask_source("import Mathlib.Data.Nat\nimport Foo\n#eval 1\ntheorem t : True := trivial\n") + ) + assert imports == ["Mathlib.Data.Nat", "Foo"] + assert [d.fqn for d in decls] == ["t"] + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("Foo.bar", "Foo.bar"), + ("@Foo.bar", "Foo.bar"), + ("«weird name».x", "weird name.x"), + ("p.1", "p"), + (".dot", None), + ("fun", None), + ], +) +def test_normalize_reference(raw: str, expected: str | None) -> None: + assert normalize_reference(raw) == expected + + +def test_empty_and_broken_files(tmp_path: Path) -> None: + assert analyze_lean_file(str(tmp_path / "E.lean"), "", repo_path=str(tmp_path)) == ([], []) + nodes, _ = analyze_lean_file( + str(tmp_path / "B.lean"), "def broken (x : := ⟨\n#eval ¬¬\n", repo_path=str(tmp_path) + ) + assert [n.qualified_name for n in nodes] == ["broken"] From 362b5f13fef7047d9daff31016b930e308802f2d Mon Sep 17 00:00:00 2001 From: Fabricio Ceolin Date: Fri, 6 Nov 2026 17:53:20 -0300 Subject: [PATCH 4/5] style: ruff format the Lean masking test Co-Authored-By: Claude Opus 5.5 --- tests/test_lean_analyzer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_lean_analyzer.py b/tests/test_lean_analyzer.py index 096437f5..6fbba270 100644 --- a/tests/test_lean_analyzer.py +++ b/tests/test_lean_analyzer.py @@ -229,7 +229,7 @@ def last := top def test_masking_keeps_lines_and_byte_offsets() -> None: - source = 'def a := "-- not a comment" -- é comment\n/- block /- nested -/ -/ def b := \'x\'\n' + source = "def a := \"-- not a comment\" -- é comment\n/- block /- nested -/ -/ def b := 'x'\n" masked = mask_source(source) assert masked.text.count("\n") == source.count("\n") assert len(masked.text.encode("utf8")) == len(source.encode("utf8")) From f4e6ca9257e79ce3f546167a318ef26aad02e10d Mon Sep 17 00:00:00 2001 From: anhnh2002 Date: Mon, 9 Nov 2026 10:30:22 +0700 Subject: [PATCH 5/5] fix: correct Lean scopes, open resolution, and export aliases --- .../analysis/call_graph_analyzer.py | 130 +---- .../be/dependency_analyzer/analyzers/lean.py | 402 ++++++++++++--- .../analyzers/lean_resolution.py | 264 ++++++++++ tests/test_lean_analyzer.py | 471 +++++++++++++++++- 4 files changed, 1088 insertions(+), 179 deletions(-) create mode 100644 codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py diff --git a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py index 213720da..47bc73b4 100644 --- a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py +++ b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py @@ -15,6 +15,7 @@ from collections import defaultdict from contextlib import contextmanager from pathlib import Path +from typing import TYPE_CHECKING from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node from codewiki.src.be.dependency_analyzer.utils.external_symbols import ( @@ -26,10 +27,11 @@ from codewiki.src.be.dependency_analyzer.utils.patterns import CODE_EXTENSIONS from codewiki.src.be.dependency_analyzer.utils.security import safe_open_text -logger = logging.getLogger(__name__) +if TYPE_CHECKING: + from codewiki.src.be.dependency_analyzer.analyzers.lean import LeanFile, LeanScope + from codewiki.src.be.dependency_analyzer.analyzers.lean_resolution import LeanNameResolver -# `_resolve_lean_name` result: no declaration has this name (try a shorter one). -_LEAN_NO_MATCH = object() +logger = logging.getLogger(__name__) class TimeoutError(BaseException): @@ -676,16 +678,12 @@ def _analyze_vue_file(self, file_path: str, content: str, repo_dir: str): logger.exception(f"Failed to analyze Vue file {file_path}") def _reset_lean_state(self) -> None: - # Relative path -> imported module names, per Lean file. - self._lean_imports: dict[str, list[str]] = {} - # Node id -> (namespace prefixes innermost first, open namespaces). - self._lean_scopes: dict[str, tuple[list[str], list[str]]] = {} - # Constructor/field FQN -> ids of the structures or inductives it names. + self._lean_files: dict[str, LeanFile] = {} + self._lean_scopes: dict[str, LeanScope] = {} self._lean_aliases: dict[str, list[str]] = defaultdict(list) self._lean_private: set[str] = set() - self._lean_index: dict[str, list[str]] | None = None - self._lean_modules: dict[str, set[str]] = {} - self._lean_visible: dict[str, frozenset[str]] = {} + self._lean_protected: set[str] = set() + self._lean_resolver: LeanNameResolver | None = None def _analyze_lean_file(self, file_path: str, content: str, repo_dir: str): """ @@ -709,110 +707,28 @@ def _analyze_lean_file(self, file_path: str, content: str, repo_dir: str): self.functions[func.id] = func self.call_relationships.extend(analyzer.call_relationships) - self._lean_imports[analyzer.relative_path] = analyzer.imports + self._lean_files[analyzer.relative_path] = analyzer.scan self._lean_scopes.update(analyzer.scopes) for alias, node_id in analyzer.aliases.items(): self._lean_aliases[alias].append(node_id) self._lean_private.update(analyzer.private_ids) + self._lean_protected.update(analyzer.protected_ids) except Exception: logger.exception(f"Failed to analyze Lean file {file_path}") - def _build_lean_index(self) -> dict[str, list[str]]: - """FQN -> Lean node ids, and module name -> files. A module name maps - to every file whose path ends with it (`Geo.Point` -> `src/Geo/Point.lean`), - because the Lake source folder is not known.""" - if self._lean_index is None: - index: dict[str, list[str]] = defaultdict(list) - for func_id, func in self.functions.items(): - if func.language == "lean" and func.qualified_name: - index[func.qualified_name].append(func_id) - for alias, node_ids in self._lean_aliases.items(): - for node_id in node_ids: - if node_id not in index[alias]: - index[alias].append(node_id) - modules: dict[str, set[str]] = defaultdict(set) - for rel in self._lean_imports: - parts = rel[: -len(".lean")].split("/") - for k in range(len(parts)): - modules[".".join(parts[k:])].add(rel) - self._lean_index = index - self._lean_modules = modules - return self._lean_index - - def _lean_visible_files(self, rel: str) -> frozenset[str]: - """The file itself and every repository file it imports, transitively.""" - if rel not in self._lean_visible: - self._build_lean_index() - seen = {rel} - queue = [rel] - while queue: - current = queue.pop() - for module in self._lean_imports.get(current, []): - for target in self._lean_modules.get(module, ()): - if target not in seen: - seen.add(target) - queue.append(target) - self._lean_visible[rel] = frozenset(seen) - return self._lean_visible[rel] - def _resolve_lean_callee(self, callee: str, caller: Node) -> str | None: - """Resolve a Lean reference to one declaration by exact FQN, in Lean's - order: the caller's namespaces (innermost first), the root, then the - open namespaces. No match, or more than one, drops the reference. - - When the whole dotted name matches nothing, its longest prefix that - matches is the constant and the rest are projections, as in Lean - (`origin.x` is `origin` then the field `x`).""" - parts = callee.split(".") - for end in range(len(parts), 0, -1): - found = self._resolve_lean_name(".".join(parts[:end]), caller) - if found is not _LEAN_NO_MATCH: - return found - return None - - def _resolve_lean_name(self, callee: str, caller: Node): - """One exact name: a node id, None (ambiguous or recursion: stop), or - _LEAN_NO_MATCH (try a shorter prefix).""" - index = self._build_lean_index() - visible = self._lean_visible_files(caller.relative_path) - namespaces, opens = self._lean_scopes.get(caller.id, ([], [])) - - def qualify(prefix: str, name: str) -> str: - return f"{prefix}.{name}" if prefix else name - - def matches(fqn: str) -> list[str]: - return [ - node_id - for node_id in index.get(fqn, []) - if self.functions[node_id].relative_path in visible - and ( - node_id not in self._lean_private - or self.functions[node_id].relative_path == caller.relative_path - ) - ] - - def unique(found) -> str | None: - # A reference to the caller itself is recursion, not an edge. - found = list(found) - return found[0] if len(found) == 1 and found[0] != caller.id else None - - if callee.startswith("_root_."): - found = matches(callee[len("_root_.") :]) - return unique(found) if found else _LEAN_NO_MATCH - - prefixes = [*namespaces, ""] - for prefix in prefixes: - found = matches(qualify(prefix, callee)) - if found: - return unique(found) # the innermost match shadows the outer ones - - opened = { - node_id - for name in opens - for prefix in prefixes - for node_id in matches(qualify(qualify(prefix, name), callee)) - } - return unique(opened) if opened else _LEAN_NO_MATCH + from codewiki.src.be.dependency_analyzer.analyzers.lean_resolution import LeanNameResolver + + if self._lean_resolver is None: + self._lean_resolver = LeanNameResolver( + self.functions, + self._lean_files, + self._lean_scopes, + self._lean_aliases, + self._lean_private, + self._lean_protected, + ) + return self._lean_resolver.resolve(callee, caller) def _analyze_go_file(self, file_path: str, content: str, repo_dir: str): """ diff --git a/codewiki/src/be/dependency_analyzer/analyzers/lean.py b/codewiki/src/be/dependency_analyzer/analyzers/lean.py index 0bbec7f1..dd1e5e12 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/lean.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/lean.py @@ -10,7 +10,8 @@ and the Lean commands are read line by line: declarations (``def``, ``theorem``/``lemma``, ``abbrev``, ``instance``, ``structure``, ``class``, ``inductive``, ``axiom``, ``opaque``), the ``namespace``/``section``/ - ``mutual``/``end`` stack, ``open`` (also ``open X in``), ``import``, + ``mutual``/``end`` stack, ``variable``, scoped/restricted ``open`` + (also ``open X in``), ``export``, ``import``, ``/-- -/`` docstrings and spans. The scanner does not depend on the parse tree, so an ``ERROR`` region does not hide a declaration. 2. **References, from tree-sitter.** Every ``identifier`` node in a @@ -30,6 +31,11 @@ ``private`` name is visible only in its own file. A reference that matches no declaration, or more than one, is dropped. A bare name is never guessed. +Open directives are bound in their command context and retain selection, hiding, +renaming and protected-name rules. Scoped opens do not introduce ordinary names. +Exports persist as aliases of their original components, including across imports; +closing a section restores its open directives and variable bindings. + Known limitations: - Generalized field notation (``v.norm`` for ``Vec.norm``) needs type information, so it does not resolve. @@ -82,7 +88,9 @@ _SECTION_RE = re.compile(rf"^\s*(?:noncomputable\s+)?section\b\s*(?P{IDENT})?") _MUTUAL_RE = re.compile(r"^\s*mutual\b") _END_RE = re.compile(rf"^\s*end\b\s*(?P{IDENT})?\s*$") -_OPEN_RE = re.compile(r"^\s*(?:(?:public|scoped|local)\s+)*(?:open|export)\s+(?P.*)$") +_OPEN_RE = re.compile(r"^\s*open\b\s*(?P.*)$") +_EXPORT_RE = re.compile(r"^\s*export\b\s*(?P.*)$") +_VARIABLE_RE = re.compile(r"^\s*variable\b\s*(?P.*)$") # Commands that end the span of the declaration before them. _COMMAND_RE = re.compile( r"^\s*(?:@\[|#|(?:" @@ -101,12 +109,9 @@ # Local names introduced in a declaration's text. Tree rules cover the parsed # parts; these also cover `ERROR` regions and tactic blocks. _LOCAL_RES = [ - re.compile(r"\bfun\s+(?P[^=]*?)=>"), - re.compile(r"(?:∀|∃|∃!|Σ|Π|λ)\s*(?P[^,]*?),"), re.compile(rf"\blet\s+(?:mut\s+|rec\s+)?(?P{IDENT})"), re.compile(rf"\bhave\s+(?P{IDENT})\s*:"), re.compile(rf"\bfor\s+(?P{IDENT})\s+in\b"), - re.compile(rf"[({{⦃\[]\s*(?P{IDENT}(?:\s+{IDENT})*)\s*:(?!=)"), re.compile(rf"\(\s*(?P{IDENT})\s*:="), re.compile( r"\b(?:intro|intros|rintro|rename_i|next|case|obtain|rcases|cases'|induction')\s+(?P[^\n:=]*)" @@ -240,6 +245,35 @@ def _clean_docstring(block: str) -> str: # ---------------------------------------------------------------------- +@dataclass(frozen=True) +class LeanOpen: + """An opening captured at its command, before entering any later scope.""" + + namespace: str + line: int + context: str + preceding: tuple[int, ...] + mode: str = "all" + names: tuple[str, ...] = () + renames: tuple[tuple[str, str], ...] = () + + +@dataclass(frozen=True) +class LeanExport: + namespace: str + names: tuple[str, ...] + destination: str + line: int + opens: tuple[int, ...] + + +@dataclass(frozen=True) +class LeanScope: + namespaces: tuple[str, ...] + opens: tuple[int, ...] + line: int + + @dataclass class LeanDecl: fqn: str @@ -250,17 +284,29 @@ class LeanDecl: indent: int private: bool namespaces: list[str] # the declaration's namespace prefixes, innermost first - opens: list[str] + opens: list[int] + protected: bool = False + variables: frozenset[str] = frozenset() docstring: str = "" aliases: list[str] = field(default_factory=list) # constructor and field FQNs bases: list[str] = field(default_factory=list) +@dataclass +class LeanFile: + declarations: list[LeanDecl] = field(default_factory=list) + imports: list[str] = field(default_factory=list) + opens: list[LeanOpen] = field(default_factory=list) + exports: list[LeanExport] = field(default_factory=list) + namespaces: list[tuple[int, str]] = field(default_factory=list) + + @dataclass class _Scope: kind: str # "namespace", "section" or "mutual" name: str - opens: list[str] = field(default_factory=list) + opens: list[int] = field(default_factory=list) + variables: set[str] = field(default_factory=set) def _qualify(prefix: str, name: str) -> str: @@ -271,29 +317,125 @@ def _strip_guillemets(name: str) -> str: return name.replace("«", "").replace("»", "") -def _parse_open_names(body: str) -> list[str]: - """`open A B (x y) hiding z renaming p → q` -> ["A", "B"].""" - body = re.split(r"\b(?:hiding|renaming)\b", body)[0] - body = re.sub(r"\([^)]*\)", " ", body) - return [_strip_guillemets(m.group(0)) for m in _IDENT_RE.finditer(body)] +_TOKEN_RE = re.compile(rf"{IDENT}|:=|=>|->|[^\s]") +_DELIMITERS = {"(": ")", "{": "}", "⦃": "⦄", "[": "]"} + + +def _tokens(text: str) -> list[str]: + return _TOKEN_RE.findall(text) + + +def _group_end(tokens: list[str], start: int) -> int: + """Index after a balanced group, or the end of an incomplete group.""" + stack = [_DELIMITERS[tokens[start]]] + for i in range(start + 1, len(tokens)): + token = tokens[i] + if token in _DELIMITERS: + stack.append(_DELIMITERS[token]) + elif token == stack[-1]: + stack.pop() + if not stack: + return i + 1 + return len(tokens) + + +def _binder_names(tokens: list[str]) -> set[str]: + """Read binder names, never identifiers in their types or defaults.""" + names: set[str] = set() + i = 0 + while i < len(tokens): + token = tokens[i] + if token in (":", ":=", "=>", ",", "where", "extends", "deriving", "|"): + break + if token in _DELIMITERS: + end = _group_end(tokens, i) + body = tokens[i + 1 : end - 1] + # `[Class α]` binds an anonymous instance, not `Class` or `α`. + if token != "[" or (len(body) > 1 and _IDENT_RE.fullmatch(body[0]) and body[1] == ":"): + names.update(_binder_names(body)) + i = end + continue + if _IDENT_RE.fullmatch(token) and "." not in token: + names.add(_strip_guillemets(token)) + i += 1 + return names - _KEYWORDS + + +def _parse_open(body: str, line: int, context: str, preceding: list[int]) -> list[LeanOpen]: + tokens = _tokens(body) + mode = "all" + if tokens and tokens[0] == "scoped": + mode = tokens.pop(0) + if not tokens or not _IDENT_RE.fullmatch(tokens[0]): + return [] + if len(tokens) > 1 and tokens[1] in ("(", "hiding", "renaming"): + namespace = _strip_guillemets(tokens[0]) + shape = tokens[1] + names: tuple[str, ...] = () + renames: tuple[tuple[str, str], ...] = () + if shape == "(": + mode = "select" + if tokens[-1] != ")": + return [] + names = tuple(_strip_guillemets(t) for t in tokens[2:-1] if _IDENT_RE.fullmatch(t)) + elif shape == "hiding": + mode = "hiding" + names = tuple(_strip_guillemets(t) for t in tokens[2:] if _IDENT_RE.fullmatch(t)) + else: + mode = "renaming" + pairs = [] + rest = tokens[2:] + while rest: + if ( + len(rest) < 3 + or not _IDENT_RE.fullmatch(rest[0]) + or rest[1] not in ("→", "->") + or not _IDENT_RE.fullmatch(rest[2]) + ): + return [] + pairs.append((_strip_guillemets(rest[0]), _strip_guillemets(rest[2]))) + rest = rest[3:] + if rest: + if rest[0] != ",": + return [] + rest = rest[1:] + renames = tuple(pairs) + return [LeanOpen(namespace, line, context, tuple(preceding), mode, names, renames)] + if any(not _IDENT_RE.fullmatch(t) for t in tokens): + return [] + return [LeanOpen(_strip_guillemets(t), line, context, tuple(preceding), mode) for t in tokens] def scan_headers(masked: _Masked) -> tuple[list[LeanDecl], list[str]]: - """Return the declarations and the imported module names of a file.""" + """Compatibility interface for callers that only need headers and imports.""" + scanned = _scan_file(masked) + return scanned.declarations, scanned.imports + + +def _scan_file(masked: _Masked) -> LeanFile: lines = masked.text.split("\n") docs = {end: text for end, text in masked.docstrings} - decls: list[LeanDecl] = [] - imports: list[str] = [] + scanned = LeanFile() + decls = scanned.declarations stack: list[_Scope] = [] - file_opens: list[str] = [] - pending_open_in: list[str] = [] + root_scope = _Scope("section", "") + pending_open_in: list[int] = [] attr_start: int | None = None + consumed_until = -1 def namespace() -> str: return ".".join(s.name for s in stack if s.kind == "namespace") - def active_opens() -> list[str]: - return file_opens + [o for s in stack for o in s.opens] + def active_opens() -> list[int]: + return root_scope.opens + [o for s in stack for o in s.opens] + + def add_decl(match, idx: int, start: int, extra_opens: list[int]) -> None: + decl = _decl_from_match(match, idx, start, namespace(), extra_opens) + if decl is not None: + decl.opens = active_opens() + decl.opens + decl.variables = frozenset(root_scope.variables.union(*(s.variables for s in stack))) + decl.docstring = _docstring_before(docs, lines, start) + decls.append(decl) def close_previous(line_idx: int, indent: int) -> None: if decls and decls[-1].end < 0 and indent <= decls[-1].indent: @@ -302,67 +444,147 @@ def close_previous(line_idx: int, indent: int) -> None: end -= 1 decls[-1].end = end + def close_scope(name: str | None) -> None: + if not stack: + logger.warning("Unmatched Lean end %s", name or "") + return + if name is None: + if not stack[-1].name: + stack.pop() + return + else: + parts = _strip_guillemets(name).split(".") + kind = stack[-1].kind + if ( + len(stack) >= len(parts) + and all(s.kind == kind for s in stack[-len(parts) :]) + and [s.name for s in stack[-len(parts) :]] == parts + ): + del stack[-len(parts) :] + return + logger.warning("Mismatched Lean end %s; retaining scope", name or "") + + def process_command( + command: str, idx: int, start: int, indent: int, temporary_opens: list[int] + ) -> list[int]: + # Peel right-associative `open ... in` wrappers without changing the + # spelling of the inner command (attributes and quoted names included). + temporary = bool(temporary_opens) + while m := _OPEN_RE.match(command): + body = m.group("body") + in_token = next((t for t in _TOKEN_RE.finditer(body) if t.group() == "in"), None) + opening_text = body[: in_token.start()] if in_token else body + preceding = active_opens() + temporary_opens + opening_ids = [] + for opening in _parse_open(opening_text, idx, namespace(), preceding): + opening_id = len(scanned.opens) + scanned.opens.append( + LeanOpen( + opening.namespace, + opening.line, + opening.context, + tuple(preceding), + opening.mode, + opening.names, + opening.renames, + ) + ) + opening_ids.append(opening_id) + preceding.append(opening_id) + if in_token is None: + if not temporary: + (stack[-1] if stack else root_scope).opens.extend(opening_ids) + return [] + temporary = True + temporary_opens = temporary_opens + opening_ids + command = " " * indent + body[in_token.end() :].strip() + if not command.strip(): + return temporary_opens + + if m := _DECL_RE.match(command): + add_decl(m, idx, start, temporary_opens) + elif m := _IMPORT_RE.match(command): + scanned.imports.extend( + _strip_guillemets(x.group(0)) for x in _IDENT_RE.finditer(m.group("mods")) + ) + elif m := _NAMESPACE_RE.match(command): + for name in _strip_guillemets(m.group("name")).split("."): + stack.append(_Scope("namespace", name)) + scanned.namespaces.append((idx, namespace())) + stack[-1].opens.extend(temporary_opens) + elif _MUTUAL_RE.match(command): + stack.append(_Scope("mutual", "", opens=temporary_opens)) + elif m := _END_RE.match(command): + close_scope(m.group("name")) + elif m := _SECTION_RE.match(command): + for name in (m.group("name") or "").split("."): + stack.append(_Scope("section", _strip_guillemets(name))) + stack[-1].opens.extend(temporary_opens) + elif m := _VARIABLE_RE.match(command): + if not temporary: + (stack[-1] if stack else root_scope).variables.update( + _binder_names(_tokens(m.group("body"))) + ) + elif m := _EXPORT_RE.match(command): + tokens = _tokens(m.group("body")) + if ( + len(tokens) >= 3 + and _IDENT_RE.fullmatch(tokens[0]) + and tokens[1] == "(" + and tokens[-1] == ")" + and all(_IDENT_RE.fullmatch(t) for t in tokens[2:-1]) + ): + scanned.exports.append( + LeanExport( + _strip_guillemets(tokens[0]), + tuple(_strip_guillemets(t) for t in tokens[2:-1]), + namespace(), + idx, + tuple(active_opens() + temporary_opens), + ) + ) + return [] + for idx, line in enumerate(lines): - if not line.strip(): + if idx <= consumed_until or not line.strip(): continue indent = len(line) - len(line.lstrip()) if decls and decls[-1].end < 0 and indent > decls[-1].indent: - continue # a body line of the open declaration, even if it looks like a command + continue decl_m = _DECL_RE.match(line) if decl_m is None and not _COMMAND_RE.match(line): - continue # a continuation line at the declaration's indentation + continue close_previous(idx, indent) - if _ATTR_ONLY_RE.match(line): if attr_start is None: attr_start = idx continue start = attr_start if attr_start is not None else idx attr_start = None - if decl_m is not None: - decl = _decl_from_match(decl_m, idx, start, namespace(), pending_open_in) + add_decl(decl_m, idx, start, pending_open_in) pending_open_in = [] - if decl is None: - continue - decl.opens = active_opens() + decl.opens - decl.docstring = _docstring_before(docs, lines, start) - decls.append(decl) continue - # `open X in` covers only the next command, whatever it is. - pending_open_in = [] - if m := _IMPORT_RE.match(line): - imports.extend( - _strip_guillemets(x.group(0)) for x in _IDENT_RE.finditer(m.group("mods")) - ) - elif m := _NAMESPACE_RE.match(line): - stack.append(_Scope("namespace", _strip_guillemets(m.group("name")))) - elif _MUTUAL_RE.match(line): - stack.append(_Scope("mutual", "")) - elif m := _END_RE.match(line): - if stack: - stack.pop() - elif m := _SECTION_RE.match(line): - stack.append(_Scope("section", m.group("name") or "")) - elif m := _OPEN_RE.match(line): - body = m.group("body") - parts = re.split(r"\bin\b", body, maxsplit=1) - names = _parse_open_names(parts[0]) - if len(parts) == 2: - pending_open_in = names - rest = parts[1] - if rest.strip() and (inner := _DECL_RE.match(" " * indent + rest.strip())): - decl = _decl_from_match(inner, idx, start, namespace(), pending_open_in) - pending_open_in = [] - if decl is not None: - decl.opens = active_opens() + decl.opens - decl.docstring = _docstring_before(docs, lines, start) - decls.append(decl) - elif stack: - stack[-1].opens.extend(names) - else: - file_opens.extend(names) + # Gather indented continuations of directives, excluding a following command. + command = line + if _OPEN_RE.match(line) or _EXPORT_RE.match(line) or _VARIABLE_RE.match(line): + end = idx + 1 + while end < len(lines): + continuation = lines[end] + if continuation.strip(): + continuation_indent = len(continuation) - len(continuation.lstrip()) + if ( + continuation_indent <= indent + or _DECL_RE.match(continuation) + or _COMMAND_RE.match(continuation) + ): + break + command += " " + continuation.strip() + consumed_until = end + end += 1 + + pending_open_in = process_command(command, idx, start, indent, pending_open_in) if decls and decls[-1].end < 0: end = len(lines) - 1 @@ -372,7 +594,7 @@ def close_previous(line_idx: int, indent: int) -> None: for decl in decls: if decl.kind in ("structure", "class", "inductive", "class inductive"): decl.aliases = _member_aliases(decl, lines) - return decls, imports + return scanned def _decl_from_match(m, idx, start, ns, open_in) -> LeanDecl | None: @@ -411,6 +633,7 @@ def _decl_from_match(m, idx, start, ns, open_in) -> LeanDecl | None: end=-1, indent=len(m.group("indent")), private="private" in m.group("mods").split(), + protected="protected" in m.group("mods").split(), namespaces=namespaces, opens=list(open_in), bases=bases, @@ -556,6 +779,41 @@ def _tokens_in( def _text_locals(text: str) -> set[str]: names: set[str] = set() + tokens = _tokens(text) + declaration_kinds = { + "def", + "theorem", + "lemma", + "abbrev", + "instance", + "structure", + "class", + "inductive", + "axiom", + "opaque", + } + for start, token in enumerate(tokens): + if token in declaration_kinds: + binder_start = start + 1 + if token == "class" and tokens[binder_start : binder_start + 1] == ["inductive"]: + binder_start += 1 + if binder_start < len(tokens) and _IDENT_RE.fullmatch(tokens[binder_start]): + binder_start += 1 # the declaration name, if present + names.update(_binder_names(tokens[binder_start:])) + break + for start, token in enumerate(tokens): + if token not in ("fun", "λ", "∀", "∃", "Σ", "Π"): + continue + end = start + 1 + terminator = "=>" if token in ("fun", "λ") else "," + while end < len(tokens): + if tokens[end] == terminator: + names.update(_binder_names(tokens[start + 1 : end])) + break + if tokens[end] in _DELIMITERS: + end = _group_end(tokens, end) + else: + end += 1 for regex in _LOCAL_RES: for m in regex.finditer(text): for ident in _IDENT_RE.finditer(m.group("names")): @@ -590,11 +848,12 @@ def __init__(self, file_path: str, content: str, repo_path: str | None = None): self.nodes: list[Node] = [] self.call_relationships: list[CallRelationship] = [] self.imports: list[str] = [] - # node id -> (namespace prefixes innermost first, open namespaces) - self.scopes: dict[str, tuple[list[str], list[str]]] = {} + self.scopes: dict[str, LeanScope] = {} + self.scan = LeanFile() # constructor/field FQN -> node id of the structure or inductive self.aliases: dict[str, str] = {} self.private_ids: set[str] = set() + self.protected_ids: set[str] = set() self.relative_path = Path(self._relative_path()).as_posix() if is_skipped_lean_file(self.relative_path): return @@ -616,7 +875,8 @@ def _component_id(self, fqn: str) -> str: def _analyze(self) -> None: masked = mask_source(self.content) - decls, self.imports = scan_headers(masked) + self.scan = _scan_file(masked) + decls, self.imports = self.scan.declarations, self.scan.imports source_lines = self.content.split("\n") by_fqn: dict[str, Node] = {} owners: list[tuple[LeanDecl, Node]] = [] @@ -643,7 +903,7 @@ def _analyze(self) -> None: masked_lines = masked.text.split("\n") for decl, node in owners: text = "\n".join(masked_lines[decl.keyword_line : decl.end + 1]) - locals_by_decl[node.id] = _text_locals(text) + locals_by_decl[node.id] = _text_locals(text) | set(decl.variables) for line, name in tree_locals: if line in line_owner: locals_by_decl[line_owner[line][1].id].add(name) @@ -690,9 +950,13 @@ def _add_node(self, decl: LeanDecl, source_lines: list[str]) -> Node: qualified_name=decl.fqn, ) self.nodes.append(node) - self.scopes[component_id] = (decl.namespaces, decl.opens) + self.scopes[component_id] = LeanScope( + tuple(decl.namespaces), tuple(decl.opens), decl.keyword_line + ) if decl.private: self.private_ids.add(component_id) + if decl.protected: + self.protected_ids.add(component_id) return node diff --git a/codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py b/codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py new file mode 100644 index 00000000..81aedc42 --- /dev/null +++ b/codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py @@ -0,0 +1,264 @@ +"""Repository-local Lean name resolution, including scoped opens and exports. + +Commands are bound in import/source order. Bindings retain their defining file +and position: importing a target alone must not expose an unrelated export. +No type elaboration, external declarations, or notation expansion is attempted. +""" + +from collections import defaultdict +from dataclasses import dataclass + +from codewiki.src.be.dependency_analyzer.analyzers.lean import ( + LeanFile, + LeanOpen, + LeanScope, + _qualify, +) +from codewiki.src.be.dependency_analyzer.models.core import Node + +NO_MATCH = object() + + +@dataclass(frozen=True) +class _Binding: + target: str + origin: str + line: int + protected: bool = False + exported: bool = False + + +def _prefixes(name: str) -> list[str]: + parts = name.split(".") if name else [] + return [".".join(parts[:end]) for end in range(len(parts), 0, -1)] + [""] + + +class LeanNameResolver: + def __init__( + self, + functions: dict[str, Node], + files: dict[str, LeanFile], + scopes: dict[str, LeanScope], + aliases: dict[str, list[str]], + private: set[str], + protected: set[str], + ): + self.functions = functions + self.files = files + self.scopes = scopes + self.private = private + self.index: dict[str, list[_Binding]] = defaultdict(list) + # Earliest introduction per file is enough to test command visibility. + self.namespaces: dict[str, dict[str, int]] = defaultdict(dict) + self.modules: dict[str, set[str]] = defaultdict(set) + self.visible: dict[str, frozenset[str]] = {} + self.open_targets: dict[tuple[str, int], tuple[str, ...]] = {} + + for rel, scanned in files.items(): + parts = rel.removesuffix(".lean").split("/") + for start in range(len(parts)): + self.modules[".".join(parts[start:])].add(rel) + for line, namespace in scanned.namespaces: + self._add_namespace(namespace, rel, line) + + # Declaration names are indexed as before; directive binding additionally + # filters by position so a later namespace cannot change an earlier open. + for node_id, node in functions.items(): + if node.language != "lean" or not node.qualified_name: + continue + scope = scopes.get(node_id) + line = scope.line if scope else node.start_line - 1 + self._add_binding( + node.qualified_name, + _Binding(node_id, node.relative_path, line, node_id in protected), + ) + for name, targets in aliases.items(): + for target in targets: + node = functions[target] + scope = scopes.get(target) + self._add_binding( + name, + _Binding( + target, + node.relative_path, + scope.line if scope else node.start_line - 1, + target in protected, + ), + ) + + done: set[str] = set() + visiting: set[str] = set() + + def bind_file(rel: str) -> None: + if rel in done or rel in visiting: + return # invalid import cycles must not abort repository analysis + visiting.add(rel) + scanned = files[rel] + for module in scanned.imports: + for dependency in sorted(self.modules.get(module, ())): + bind_file(dependency) + events = [(opening.line, 0, i) for i, opening in enumerate(scanned.opens)] + events += [(export.line, 1, i) for i, export in enumerate(scanned.exports)] + for _, kind, i in sorted(events): + if kind == 0: + self._bind_open(rel, i) + else: + export = scanned.exports[i] + namespaces = self._namespace_targets( + export.namespace, export.destination, export.opens, rel, export.line + ) + for name in export.names: + targets = { + binding.target + for namespace in namespaces + for binding in self._matches( + _qualify(namespace, name), rel, export.line + ) + } + if len(targets) == 1: + self._add_binding( + _qualify(export.destination, name), + _Binding(next(iter(targets)), rel, export.line, exported=True), + ) + visiting.remove(rel) + done.add(rel) + + for rel in sorted(files): + bind_file(rel) + + def _add_binding(self, name: str, binding: _Binding) -> None: + if binding not in self.index[name]: + self.index[name].append(binding) + for namespace in _prefixes(name)[1:-1]: + self._add_namespace(namespace, binding.origin, binding.line) + + def _add_namespace(self, namespace: str, origin: str, line: int) -> None: + introductions = self.namespaces[namespace] + introductions[origin] = min(line, introductions.get(origin, line)) + + def _visible_files(self, rel: str) -> frozenset[str]: + if rel not in self.visible: + seen = {rel} + queue = [rel] + while queue: + scanned = self.files.get(queue.pop()) + if scanned is None: + continue + for module in scanned.imports: + for dependency in self.modules.get(module, ()): + if dependency not in seen: + seen.add(dependency) + queue.append(dependency) + self.visible[rel] = frozenset(seen) + return self.visible[rel] + + def _matches( + self, name: str, rel: str, line: int, *, declarations_after: bool = False + ) -> list[_Binding]: + visible = self._visible_files(rel) + return [ + binding + for binding in self.index.get(name, ()) + if binding.origin in visible + and self.functions[binding.target].relative_path in visible + and ( + binding.origin != rel + or binding.line <= line + or declarations_after + and not binding.exported + ) + and ( + binding.target not in self.private + or self.functions[binding.target].relative_path == rel + ) + ] + + @staticmethod + def _opened_name(opening: LeanOpen, name: str) -> str | None: + if opening.mode == "scoped": + return None + if opening.mode == "select": + return name if name in opening.names else None + if opening.mode == "hiding": + return None if name in opening.names else name + if opening.mode == "renaming": + return next((source for source, alias in opening.renames if alias == name), None) + return name + + def _namespace_targets( + self, name: str, context: str, opens: tuple[int, ...], rel: str, line: int + ) -> tuple[str, ...]: + visible = self._visible_files(rel) + + def exists(namespace: str) -> bool: + return any( + origin in visible and (origin != rel or position <= line) + for origin, position in self.namespaces.get(namespace, {}).items() + ) + + if name.startswith("_root_."): + target = name[len("_root_.") :] + return (target,) if exists(target) else () + candidates = {_qualify(prefix, name) for prefix in _prefixes(context)} + for opening_id in opens: + opening = self.files[rel].opens[opening_id] + suffix = self._opened_name(opening, name) + if suffix is not None: + candidates.update( + _qualify(namespace, suffix) + for namespace in self.open_targets.get((rel, opening_id), ()) + ) + return tuple(sorted(candidate for candidate in candidates if exists(candidate))) + + def _bind_open(self, rel: str, opening_id: int) -> None: + opening = self.files[rel].opens[opening_id] + targets = self._namespace_targets( + opening.namespace, opening.context, opening.preceding, rel, opening.line + ) + # `hiding` and `renaming` require one unambiguous namespace. + if opening.mode in ("hiding", "renaming") and len(targets) != 1: + targets = () + self.open_targets[rel, opening_id] = targets + + def resolve(self, callee: str, caller: Node) -> str | None: + """Find the longest constant prefix, preserving ambiguity as a stop.""" + parts = callee.split(".") + for end in range(len(parts), 0, -1): + found = self._resolve_name(".".join(parts[:end]), caller) + if found is not NO_MATCH: + return found + return None + + def _resolve_name(self, name: str, caller: Node): + rel = caller.relative_path + scope = self.scopes.get(caller.id, LeanScope((), (), caller.start_line - 1)) + + def matches(fqn: str) -> list[_Binding]: + # Keep the existing declaration lookup behavior (including mutual + # definitions), but never expose an export before its command. + return self._matches(fqn, rel, scope.line, declarations_after=True) + + def targets(fqn: str) -> set[str]: + return {binding.target for binding in matches(fqn)} + + def unique(found: set[str]) -> str | None: + return next(iter(found)) if len(found) == 1 and caller.id not in found else None + + if name.startswith("_root_."): + found = targets(name[len("_root_.") :]) + return unique(found) if found else NO_MATCH + for prefix in (*scope.namespaces, ""): + found = targets(_qualify(prefix, name)) + if found: + return unique(found) + opened: set[str] = set() + for opening_id in scope.opens: + opening = self.files[rel].opens[opening_id] + suffix = self._opened_name(opening, name) + if suffix is None: + continue + for namespace in self.open_targets.get((rel, opening_id), ()): + for binding in matches(_qualify(namespace, suffix)): + if opening.mode in ("select", "renaming") or not binding.protected: + opened.add(binding.target) + return unique(opened) if opened else NO_MATCH diff --git a/tests/test_lean_analyzer.py b/tests/test_lean_analyzer.py index 6fbba270..49709269 100644 --- a/tests/test_lean_analyzer.py +++ b/tests/test_lean_analyzer.py @@ -7,16 +7,16 @@ pytest.importorskip("tree_sitter_language_pack") -from codewiki.src.be.dependency_analyzer.analysis.call_graph_analyzer import ( +from codewiki.src.be.dependency_analyzer.analysis.call_graph_analyzer import ( # noqa: E402 CallGraphAnalyzer, ) -from codewiki.src.be.dependency_analyzer.analyzers.lean import ( +from codewiki.src.be.dependency_analyzer.analyzers.lean import ( # noqa: E402 analyze_lean_file, mask_source, normalize_reference, scan_headers, ) -from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_lean_file +from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_lean_file # noqa: E402 POINT = """\ /-! Geometry primitives. -/ @@ -272,3 +272,468 @@ def test_empty_and_broken_files(tmp_path: Path) -> None: str(tmp_path / "B.lean"), "def broken (x : := ⟨\n#eval ¬¬\n", repo_path=str(tmp_path) ) assert [n.qualified_name for n in nodes] == ["broken"] + + +def _analyze_sources(tmp_path: Path, sources: dict[str, str], *, reverse: bool = False): + files = [_write(tmp_path, rel, source) for rel, source in sources.items()] + analyzer = CallGraphAnalyzer() + analyzer.analyze_code_files( + [{"path": rel, "language": "lean"} for rel in (reversed(files) if reverse else files)], + str(tmp_path), + ) + return analyzer, {(r.caller, r.callee) for r in analyzer.call_relationships} + + +@pytest.mark.parametrize( + "binder", + [ + "fun (x : Foo) => x", + "λ (x : Foo) => x", + "fun x : Foo => x", + "fun (x y : Foo) => x", + "fun {x : Foo} => x", + "fun ⦃x : Foo⦄ => x", + "fun [inst : Foo] => inst", + "fun [Foo] => 1", + "∀ (x : Foo), x = x", + "∃ (x : Foo), x = x", + "∃! (x : Foo), x = x", + "Σ (x : Foo), Foo", + "Π (x : Foo), Foo", + ], +) +def test_typed_binders_preserve_type_dependencies(tmp_path: Path, binder: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": f"structure Foo where\n value : Nat\ndef use := {binder}\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo")} + + +def test_nested_binder_types_are_not_local_names(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\n" + "structure Bar where\n value : Nat\n" + "def use := fun (x : Foo × (List Bar)) (y : Bar := ⟨1⟩) => x\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo"), ("M.lean::use", "M.lean::Bar")} + + +def test_untyped_binders_suppress_globals(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "def x := 1\ndef y := 2\ndef use := fun x y => x + y\n", + }, + ) + assert not edges + + +@pytest.mark.parametrize( + "header, ending", + [ + ("namespace A\nnamespace B", "end A.B"), + ("namespace A.B", "end A.B"), + ("section A\nsection B", "end A.B"), + ("section A.B", "end A.B"), + ], +) +def test_named_end_closes_matching_suffix(tmp_path: Path, header: str, ending: str) -> None: + analyzer, _ = _analyze_sources( + tmp_path, + { + "M.lean": f"{header}\ndef inside := 1\n{ending}\ndef outside := 2\n", + }, + ) + assert "M.lean::outside" in analyzer.functions + + +def test_partial_namespace_end_and_section_barrier(tmp_path: Path, caplog) -> None: + analyzer, _ = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\nnamespace C\nend B.C\ndef first := 1\n" + "section\nnamespace D.E\nend A.D.E\ndef retained := 2\n" + "end D.E\nend\nend A\ndef last := 3\n", + }, + ) + assert set(analyzer.functions) == {"M.lean::A.first", "M.lean::A.D.E.retained", "M.lean::last"} + assert "Mismatched Lean end" in caplog.text + + +def test_section_variables_are_scoped_and_restored(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "def x := 1\ndef y := 2\nvariable (x : Nat)\n" + "section Outer.Inner\nvariable {y : Nat}\ndef inside := x + y\n" + "end Inner\ndef middle := x + y\nend Outer\ndef outside := x + y\n", + }, + ) + assert edges == {("M.lean::middle", "M.lean::y"), ("M.lean::outside", "M.lean::y")} + + +def test_namespace_variables_expire_and_types_are_not_binders(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\ndef x := 1\n" + "namespace A\nvariable\n (x : Foo)\n [Inhabited Foo]\n" + "def inside := (x, (⟨1⟩ : Foo))\nend A\ndef outside := x\n", + }, + ) + assert edges == {("M.lean::A.inside", "M.lean::Foo"), ("M.lean::outside", "M.lean::x")} + + +_OPEN_FIXTURE = "namespace A\ndef f := 1\ndef g := 2\nend A\nnamespace B\ndef g := 3\nend B\n" + + +@pytest.mark.parametrize("opening", ["open A (f)", "open A hiding g", "open scoped A"]) +def test_open_restrictions_do_not_create_ambiguity(tmp_path: Path, opening: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + f"{opening}\nopen B\ndef use := g\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::B.g")} + + +@pytest.mark.parametrize( + "opening, expected", + [ + ("open A (f)", {"A.f"}), + ("open A hiding g", {"A.f"}), + ("open scoped A", set()), + ("open A renaming f → h", {"A.f"}), + ("open A renaming f -> h, g -> k", {"A.f", "A.g"}), + ("open A\n renaming\n f → h,\n g → k", {"A.f", "A.g"}), + ("open A\n hiding g", {"A.f"}), + ("open A\n (f)", {"A.f"}), + ], +) +def test_open_modes_only_expose_the_selected_names( + tmp_path: Path, opening: str, expected: set +) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + f"{opening}\ndef use := (f, g, h, k)\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::" + name) for name in expected} + + +@pytest.mark.parametrize( + "opening, exposed", + [ + ("open A", False), + ("open A hiding other", False), + ("open A (secret)", True), + ("open A renaming secret → value", True), + ], +) +def test_protected_names_require_explicit_open(tmp_path: Path, opening: str, exposed: bool) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\nprotected def secret := 1\ndef other := 2\nend A\n" + + opening + + "\ndef use := (secret, value)\n", + }, + ) + assert edges == ({("M.lean::use", "M.lean::A.secret")} if exposed else set()) + + +@pytest.mark.parametrize("same_line", [False, True]) +def test_restricted_open_in_expires(tmp_path: Path, same_line: bool) -> None: + separator = " " if same_line else "\n" + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + + "open A renaming f → value in" + + separator + + "def inside := value\ndef outside := value\n", + }, + ) + assert edges == {("M.lean::inside", "M.lean::A.f")} + + +def test_open_in_applies_to_a_namespace_block(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "open A in\nnamespace Inner\n" + "def inside := f\nend Inner\ndef outside := f\n", + }, + ) + assert edges == {("M.lean::Inner.inside", "M.lean::A.f")} + + +def test_open_uses_the_command_namespace(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ndef f := 1\nend A\n" + "namespace B.A\ndef f := 2\nend B.A\nopen A\n" + "namespace B\ndef use := f\nend B\n", + }, + ) + assert edges == {("M.lean::B.use", "M.lean::A.f")} + + +def test_relative_and_sequential_opens(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B.C\ndef f := 1\nend A.B.C\n" + "namespace A\nopen B C\ndef inside := f\nend A\n" + "def outside := f\n", + }, + ) + assert edges == {("M.lean::A.inside", "M.lean::A.B.C.f")} + + +def test_later_namespace_does_not_rebind_an_open(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ndef f := 1\nend A\nnamespace B\nopen A\n" + "namespace A\ndef f := 2\nend A\ndef use := f\nend B\n", + }, + ) + assert edges == {("M.lean::B.use", "M.lean::A.f")} + + +@pytest.mark.parametrize("reverse", [False, True]) +def test_exports_are_persistent_visible_aliases(tmp_path: Path, reverse: bool) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "Base.lean": "namespace A\ndef f := 1\ndef g := 2\nend A\n", + "Exports.lean": "import Base\nnamespace B\nsection\nexport A (f)\nend\nend B\n", + "Reexport.lean": "import Exports\nnamespace C\nexport B (f)\n" + "def inside := f\nend C\ndef outside := C.f\n", + "Use.lean": "import Reexport\nopen C\ndef use := (f, B.f, C.f)\n", + "Island.lean": "import Base\ndef missing := (B.f, C.f)\n", + }, + reverse=reverse, + ) + assert edges == { + ("Reexport.lean::C.inside", "Base.lean::A.f"), + ("Reexport.lean::outside", "Base.lean::A.f"), + ("Use.lean::use", "Base.lean::A.f"), + } + assert len(analyzer.functions) == 6 # aliases introduce no components + + +def test_exports_only_expose_selected_names_and_keep_command_context(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "namespace Outer\nopen A\n" + "namespace Inner\nexport _root_.A (f)\nend Inner\nend Outer\n" + "open Outer.Inner\ndef use := (f, g, Outer.Inner.f)\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.f")} + + +def test_root_and_constructor_exports_resolve_to_original_components(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ninductive Choice where\n | first\n | second\n" + "export Choice (first)\nend A\nexport A (first)\ndef use := first\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.Choice")} + + +def test_export_is_not_visible_before_its_command(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "def before := f\nexport A (f)\ndef after := f\n", + }, + ) + assert edges == {("M.lean::after", "M.lean::A.f")} + + +def test_private_and_ambiguous_exports_are_not_linked(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "Base.lean": "namespace A\nprivate def secret := 1\ndef f := 2\nend A\n", + "Other.lean": "namespace A\ndef f := 3\nend A\n", + "Exports.lean": "import Base\nimport Other\nexport A (secret f)\ndef use := (secret, f)\n", + "Use.lean": "import Exports\ndef use := (secret, f)\n", + }, + ) + assert not edges + + +def test_lean_resolution_state_is_reset_between_runs(tmp_path: Path) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "export A (f)\ndef use := f\n", + }, + ) + assert edges + _write(tmp_path, "M.lean", "def use := f\n") + analyzer.analyze_code_files([{"path": "M.lean", "language": "lean"}], str(tmp_path)) + assert not analyzer.call_relationships + + +def test_malformed_directives_do_not_abort_the_file(tmp_path: Path) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "open A renaming f →\nexport A (\nend Nope\n" + "def after := A.f\n", + }, + ) + assert "M.lean::after" in analyzer.functions + assert edges == {("M.lean::after", "M.lean::A.f")} + + +@pytest.mark.parametrize("separator", [" ", "\n"]) +def test_nested_open_in_wrappers_expire(tmp_path: Path, separator: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\n" + + "open A in" + + separator + + "open B in" + + separator + + "@[simp] def inside := f\ndef outside := f\n", + }, + ) + assert edges == {("M.lean::inside", "M.lean::A.B.f")} + + +def test_open_in_does_not_leak_an_inner_open_command(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\nopen A in open B\ndef outside := f\n", + }, + ) + assert not edges + + +def test_open_in_is_consumed_by_a_non_declaration(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "open A in\n#check f\ndef outside := f\n", + }, + ) + assert not edges + + +def test_named_end_restores_opens_and_variables(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "def x := 1\nsection Outer.Inner\n" + "open A\nvariable (x : Nat)\ndef inside := (f, x)\n" + "end Outer.Inner\ndef outside := (f, x)\n", + }, + ) + assert edges == {("M.lean::inside", "M.lean::A.f"), ("M.lean::outside", "M.lean::x")} + + +def test_selective_open_disambiguates_each_selected_name(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\nnamespace B\ndef g := 2\nend B\n" + "open A\nopen B (f g)\ndef use := (f, g)\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.B.f"), ("M.lean::use", "M.lean::B.g")} + + +def test_duplicate_exports_to_the_same_target_are_not_ambiguous(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "Base.lean": _OPEN_FIXTURE, + "First.lean": "import Base\nnamespace Shared\nexport A (f)\nend Shared\n", + "Second.lean": "import Base\nnamespace Shared\nexport A (f)\nend Shared\n", + "Use.lean": "import First\nimport Second\ndef use := Shared.f\n", + }, + ) + assert edges == {("Use.lean::use", "Base.lean::A.f")} + + +def test_export_source_resolves_through_an_open_namespace(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\n" + "namespace C\nopen A\nexport B (f)\nend C\ndef use := C.f\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.B.f")} + + +def test_export_keeps_the_namespace_at_its_command(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ndef f := 1\nend A\nnamespace C\nexport A (f)\n" + "namespace A\ndef f := 2\nend A\ndef use := f\nend C\n", + }, + ) + assert edges == {("M.lean::C.use", "M.lean::A.f")} + + +def test_invalid_import_cycle_and_unresolved_exports_do_not_crash(tmp_path: Path) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "A.lean": "import B\nexport Missing (f)\ndef a := 1\n", + "B.lean": "import A\nopen Missing\ndef b := 2\n", + }, + ) + assert set(analyzer.functions) == {"A.lean::a", "B.lean::b"} + assert not edges + + +@pytest.mark.parametrize( + "body", + [ + "fun (x : (Foo : Type)) => x", + "fun [Inhabited (Foo : Type)] => (⟨1⟩ : Foo)", + "((⟨1⟩ : Foo), (Foo : Type))", + ], +) +def test_type_ascriptions_are_not_mistaken_for_binders(tmp_path: Path, body: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\ndef use := " + body + "\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo")} + + +def test_declaration_binders_still_work_in_error_regions(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\ndef x := 1\n" + "def use\n (x : Foo)\n : Foo := by\n exact x\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo")}