From 140c5ccfebf31575eb29aa8928a620e76ebc19e7 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Mon, 18 May 2026 21:04:18 +0900 Subject: [PATCH 1/3] Add partial port audit --- .github/workflows/partial-port-audit.yml | 37 +++ python/lean_pool/partial_port_audit.py | 353 +++++++++++++++++++++++ python/tests/test_partial_port_audit.py | 72 +++++ 3 files changed, 462 insertions(+) create mode 100644 .github/workflows/partial-port-audit.yml create mode 100644 python/lean_pool/partial_port_audit.py create mode 100644 python/tests/test_partial_port_audit.py diff --git a/.github/workflows/partial-port-audit.yml b/.github/workflows/partial-port-audit.yml new file mode 100644 index 0000000000..7e2f0ef661 --- /dev/null +++ b/.github/workflows/partial-port-audit.yml @@ -0,0 +1,37 @@ +name: Partial Port Audit + +on: + pull_request: + paths: + - 'LeanPool.lean' + - 'LeanPool/**/*.lean' + - 'LeanPool/projects.yml' + +permissions: + contents: read + +jobs: + audit: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd + with: + fetch-depth: 0 + + - uses: astral-sh/setup-uv@d0d8abe699bfb85fec6de9f7adb5ae17292296ff + with: + enable-cache: true + + - name: Sync Python dependencies + working-directory: python + run: uv sync + + - name: Compare imported project with upstream + working-directory: python + run: | + set -euo pipefail + git fetch origin "${{ github.base_ref }}" --depth=1 + uv run python -m lean_pool.partial_port_audit \ + --repo .. \ + --base-ref "origin/${{ github.base_ref }}" \ + --head-ref HEAD diff --git a/python/lean_pool/partial_port_audit.py b/python/lean_pool/partial_port_audit.py new file mode 100644 index 0000000000..e3283706a5 --- /dev/null +++ b/python/lean_pool/partial_port_audit.py @@ -0,0 +1,353 @@ +"""Detect suspiciously partial imports against an upstream GitHub project.""" + +from __future__ import annotations + +import argparse +import json +import re +import shutil +import subprocess +import sys +import tempfile +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import yaml + +from lean_pool.quality import _strip_lean_comments + +DEFAULT_LOC_TOLERANCE = 0.10 +DEFAULT_MIN_UPSTREAM_LOC = 200 +MAX_MISSING_FILES = 8 +PROJECTS_YML = "LeanPool/projects.yml" +GITHUB_REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") + + +@dataclass(frozen=True) +class LeanFile: + """Summary for one Lean file.""" + + path: str + loc: int + normalized_stem: str + + +@dataclass(frozen=True) +class LeanStats: + """Aggregate Lean file stats.""" + + files: tuple[LeanFile, ...] + + @property + def file_count(self) -> int: + """Return the number of Lean files.""" + return len(self.files) + + @property + def loc(self) -> int: + """Return non-comment, non-blank Lean lines.""" + return sum(file.loc for file in self.files) + + @property + def normalized_stems(self) -> set[str]: + """Return normalized file stems for filename comparison.""" + return {file.normalized_stem for file in self.files} + + +@dataclass(frozen=True) +class AuditFinding: + """A partial-port warning for one LeanPool project.""" + + project: str + upstream_repo: str + imported_files: int + upstream_files: int + imported_loc: int + upstream_loc: int + file_ratio: float + loc_ratio: float + missing_files: tuple[str, ...] + + def format(self) -> str: + """Format the finding as a concise human-readable paragraph.""" + missing = ", ".join(self.missing_files) if self.missing_files else "none" + return ( + f"{self.project}: imported {self.imported_files}/{self.upstream_files} " + f"Lean files and {self.imported_loc}/{self.upstream_loc} non-comment " + f"LOC from {self.upstream_repo}; missing large upstream filenames: " + f"{missing}" + ) + + +def run_git(repo: Path, *args: str, allow_failure: bool = False) -> str: + """Run git in ``repo`` and return stdout.""" + process = subprocess.run( + ["git", *args], + cwd=repo, + check=False, + capture_output=True, + text=True, + ) + if process.returncode and not allow_failure: + raise RuntimeError(process.stderr.strip() or process.stdout.strip()) + return process.stdout + + +def git_show(repo: Path, revision: str, path: str) -> str: + """Return a file from ``revision``.""" + return run_git(repo, "show", f"{revision}:{path}") + + +def changed_paths(repo: Path, base_ref: str, head_ref: str) -> list[str]: + """Return paths changed between ``base_ref`` and ``head_ref``.""" + output = run_git(repo, "diff", "--name-only", f"{base_ref}...{head_ref}") + return [line for line in output.splitlines() if line] + + +def load_projects(repo: Path, head_ref: str) -> list[dict[str, Any]]: + """Load ``LeanPool/projects.yml`` from ``head_ref``.""" + data = yaml.safe_load(git_show(repo, head_ref, PROJECTS_YML)) or {} + projects = data.get("projects") or [] + return [project for project in projects if isinstance(project, dict)] + + +def project_prefix(entry_module: str) -> str | None: + """Return the top-level ``LeanPool/`` path component.""" + parts = entry_module.split(".") + if len(parts) != 2 or parts[0] != "LeanPool": + return None + return parts[1] + + +def touched_prefixes(paths: list[str]) -> set[str]: + """Return top-level LeanPool project prefixes touched by a diff.""" + prefixes: set[str] = set() + for path in paths: + parts = path.split("/") + if len(parts) == 2 and parts[0] == "LeanPool" and parts[1].endswith(".lean"): + prefixes.add(parts[1].removesuffix(".lean")) + elif len(parts) >= 3 and parts[0] == "LeanPool": + prefixes.add(parts[1]) + return prefixes + + +def github_repo(project: dict[str, Any]) -> str | None: + """Return ``source.github_repo`` if present and syntactically safe.""" + source = project.get("source") + if not isinstance(source, dict): + return None + repo = source.get("github_repo") + if not isinstance(repo, str) or not GITHUB_REPO_RE.fullmatch(repo): + return None + return repo + + +def count_lean_loc(text: str) -> int: + """Count non-comment, non-blank Lean source lines.""" + stripped = _strip_lean_comments(text) + return sum(1 for line in stripped.splitlines() if line.strip()) + + +def normalize_stem(path: str | Path) -> str: + """Normalize a Lean filename stem for rough source/import matching.""" + return re.sub(r"[^a-z0-9]", "", Path(path).stem.lower()) + + +def ignored_lean_path(path: Path) -> bool: + """Return true for Lean files that should not count as source content.""" + if path.name == "lakefile.lean": + return True + ignored_parts = {".git", ".lake", "build", "lake-packages"} + return any(part in ignored_parts for part in path.parts) + + +def stats_from_worktree(root: Path) -> LeanStats: + """Collect Lean stats from an upstream checkout.""" + files: list[LeanFile] = [] + for path in sorted(root.rglob("*.lean")): + relative = path.relative_to(root) + if ignored_lean_path(relative): + continue + text = path.read_text() + files.append( + LeanFile( + path=relative.as_posix(), + loc=count_lean_loc(text), + normalized_stem=normalize_stem(relative), + ) + ) + return LeanStats(tuple(files)) + + +def stats_from_git_project(repo: Path, head_ref: str, prefix: str) -> LeanStats: + """Collect Lean stats for ``LeanPool/`` at ``head_ref``.""" + paths = [ + path + for path in run_git( + repo, + "ls-tree", + "-r", + "--name-only", + head_ref, + "--", + f"LeanPool/{prefix}.lean", + f"LeanPool/{prefix}", + ).splitlines() + if path.endswith(".lean") + ] + files = [ + LeanFile( + path=path, + loc=count_lean_loc(git_show(repo, head_ref, path)), + normalized_stem=normalize_stem(path), + ) + for path in paths + ] + return LeanStats(tuple(files)) + + +def clone_upstream(repo: str, clone_root: Path) -> Path | None: + """Clone ``repo`` under ``clone_root`` and return the checkout path.""" + target = clone_root / repo.replace("/", "__") + if target.exists(): + shutil.rmtree(target) + process = subprocess.run( + ["git", "clone", "--depth=1", f"https://github.com/{repo}.git", str(target)], + check=False, + capture_output=True, + text=True, + ) + if process.returncode: + print( + f"partial-port-audit: warning: could not clone {repo}: " + f"{process.stderr.strip() or process.stdout.strip()}", + file=sys.stderr, + ) + return None + return target + + +def evaluate_stats( + project: str, + upstream_repo: str, + imported: LeanStats, + upstream: LeanStats, + loc_tolerance: float = DEFAULT_LOC_TOLERANCE, + min_upstream_loc: int = DEFAULT_MIN_UPSTREAM_LOC, +) -> AuditFinding | None: + """Return a finding when imported stats look too small.""" + if upstream.loc < min_upstream_loc or not upstream.file_count: + return None + loc_ratio = imported.loc / upstream.loc if upstream.loc else 1.0 + file_ratio = ( + imported.file_count / upstream.file_count if upstream.file_count else 1.0 + ) + cutoff = 1.0 - loc_tolerance + imported_stems = imported.normalized_stems + large_file_cutoff = max(25, int(upstream.loc * loc_tolerance)) + missing_files = tuple( + file.path + for file in sorted(upstream.files, key=lambda item: item.loc, reverse=True) + if file.loc >= large_file_cutoff + and file.normalized_stem not in imported_stems + )[:MAX_MISSING_FILES] + file_count_suspicious = upstream.file_count >= 3 and file_ratio < cutoff + if loc_ratio >= cutoff and not file_count_suspicious and not missing_files: + return None + return AuditFinding( + project=project, + upstream_repo=upstream_repo, + imported_files=imported.file_count, + upstream_files=upstream.file_count, + imported_loc=imported.loc, + upstream_loc=upstream.loc, + file_ratio=file_ratio, + loc_ratio=loc_ratio, + missing_files=missing_files, + ) + + +def audit_projects( + repo: Path, + base_ref: str, + head_ref: str, + clone_root: Path, + loc_tolerance: float = DEFAULT_LOC_TOLERANCE, + min_upstream_loc: int = DEFAULT_MIN_UPSTREAM_LOC, +) -> list[AuditFinding]: + """Audit changed LeanPool projects for partial upstream imports.""" + touched = touched_prefixes(changed_paths(repo, base_ref, head_ref)) + if not touched: + return [] + findings: list[AuditFinding] = [] + for project in load_projects(repo, head_ref): + entry_module = project.get("entry_module") + if not isinstance(entry_module, str): + continue + prefix = project_prefix(entry_module) + if prefix not in touched: + continue + upstream_repo = github_repo(project) + if upstream_repo is None: + continue + upstream_path = clone_upstream(upstream_repo, clone_root) + if upstream_path is None: + continue + imported_stats = stats_from_git_project(repo, head_ref, prefix) + upstream_stats = stats_from_worktree(upstream_path) + finding = evaluate_stats( + project=entry_module, + upstream_repo=upstream_repo, + imported=imported_stats, + upstream=upstream_stats, + loc_tolerance=loc_tolerance, + min_upstream_loc=min_upstream_loc, + ) + if finding is not None: + findings.append(finding) + return findings + + +def parse_args(argv: list[str]) -> argparse.Namespace: + """Parse CLI arguments.""" + parser = argparse.ArgumentParser( + description="Compare changed LeanPool projects with upstream GitHub repos." + ) + parser.add_argument("--repo", type=Path, default=Path.cwd()) + parser.add_argument("--base-ref", default="origin/main") + parser.add_argument("--head-ref", default="HEAD") + parser.add_argument("--loc-tolerance", type=float, default=DEFAULT_LOC_TOLERANCE) + parser.add_argument( + "--min-upstream-loc", type=int, default=DEFAULT_MIN_UPSTREAM_LOC + ) + parser.add_argument("--json", action="store_true") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + """Run the partial-port audit CLI.""" + args = parse_args(sys.argv[1:] if argv is None else argv) + repo = args.repo.resolve() + with tempfile.TemporaryDirectory(prefix="lean-pool-upstream-") as temp_dir: + findings = audit_projects( + repo=repo, + base_ref=args.base_ref, + head_ref=args.head_ref, + clone_root=Path(temp_dir), + loc_tolerance=args.loc_tolerance, + min_upstream_loc=args.min_upstream_loc, + ) + if args.json: + print(json.dumps([asdict(finding) for finding in findings], indent=2)) + elif findings: + print("Suspicious partial upstream imports detected:") + for finding in findings: + print(f"- {finding.format()}") + else: + print("No suspicious partial upstream imports detected.") + return 1 if findings else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tests/test_partial_port_audit.py b/python/tests/test_partial_port_audit.py new file mode 100644 index 0000000000..846d9281d5 --- /dev/null +++ b/python/tests/test_partial_port_audit.py @@ -0,0 +1,72 @@ +"""Tests for deterministic partial-port auditing.""" + +from lean_pool.partial_port_audit import ( + LeanFile, + LeanStats, + count_lean_loc, + evaluate_stats, + normalize_stem, +) + + +def test_count_lean_loc_ignores_comments_and_blank_lines() -> None: + """Comment-only and blank lines do not inflate LOC counts.""" + text = """ +-- a line comment + +/- a block +comment -/ +def x := 1 +theorem y : x = 1 := rfl +""" + assert count_lean_loc(text) == 2 + + +def test_normalize_stem_matches_renamed_camel_case_file() -> None: + """Snake-case upstream files can match CamelCase imported files.""" + assert normalize_stem("Lean4/directed_van_kampen.lean") == normalize_stem( + "LeanPool/DirectedTopologyLean4/DirectedVanKampen.lean" + ) + + +def test_evaluate_stats_flags_large_loc_gap() -> None: + """A small imported slice of a larger upstream project is suspicious.""" + imported = LeanStats((LeanFile("LeanPool/Fineqs/Main.lean", 90, "main"),)) + upstream = LeanStats((LeanFile("fineqs/Main.lean", 570, "main"),)) + finding = evaluate_stats("LeanPool.Fineqs", "owner/fineqs", imported, upstream) + assert finding is not None + assert finding.imported_loc == 90 + assert finding.upstream_loc == 570 + + +def test_evaluate_stats_flags_missing_large_file() -> None: + """A headline-sized upstream file missing from the import is suspicious.""" + imported = LeanStats( + ( + LeanFile("LeanPool/DirectedTopologyLean4/Dipath.lean", 500, "dipath"), + LeanFile( + "LeanPool/DirectedTopologyLean4/Dihomotopy.lean", + 500, + "dihomotopy", + ), + ) + ) + upstream = LeanStats( + ( + LeanFile("Lean4/dipath.lean", 500, "dipath"), + LeanFile("Lean4/dihomotopy.lean", 500, "dihomotopy"), + LeanFile("Lean4/directed_van_kampen.lean", 1000, "directedvankampen"), + ) + ) + finding = evaluate_stats( + "LeanPool.DirectedTopologyLean4", "owner/directed", imported, upstream + ) + assert finding is not None + assert finding.missing_files == ("Lean4/directed_van_kampen.lean",) + + +def test_evaluate_stats_accepts_close_import() -> None: + """A near-complete import passes the tolerance check.""" + imported = LeanStats((LeanFile("LeanPool/Foo/Main.lean", 920, "main"),)) + upstream = LeanStats((LeanFile("Foo/Main.lean", 1000, "main"),)) + assert evaluate_stats("LeanPool.Foo", "owner/foo", imported, upstream) is None From 25471caea8fc3e7845ad0895c7f464ba9ce9d312 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Mon, 18 May 2026 21:08:05 +0900 Subject: [PATCH 2/3] Format partial port audit --- python/lean_pool/partial_port_audit.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/python/lean_pool/partial_port_audit.py b/python/lean_pool/partial_port_audit.py index e3283706a5..05da0f8b18 100644 --- a/python/lean_pool/partial_port_audit.py +++ b/python/lean_pool/partial_port_audit.py @@ -249,8 +249,7 @@ def evaluate_stats( missing_files = tuple( file.path for file in sorted(upstream.files, key=lambda item: item.loc, reverse=True) - if file.loc >= large_file_cutoff - and file.normalized_stem not in imported_stems + if file.loc >= large_file_cutoff and file.normalized_stem not in imported_stems )[:MAX_MISSING_FILES] file_count_suspicious = upstream.file_count >= 3 and file_ratio < cutoff if loc_ratio >= cutoff and not file_count_suspicious and not missing_files: From 3f46f3790740d70e7c44d65cf853acbc5cfbd411 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Mon, 18 May 2026 21:11:30 +0900 Subject: [PATCH 3/3] Improve partial port audit findings --- python/lean_pool/partial_port_audit.py | 15 +++++++++++---- python/tests/test_partial_port_audit.py | 20 ++++++++++++++++++++ 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/python/lean_pool/partial_port_audit.py b/python/lean_pool/partial_port_audit.py index 05da0f8b18..a82cf23c5a 100644 --- a/python/lean_pool/partial_port_audit.py +++ b/python/lean_pool/partial_port_audit.py @@ -75,7 +75,7 @@ def format(self) -> str: return ( f"{self.project}: imported {self.imported_files}/{self.upstream_files} " f"Lean files and {self.imported_loc}/{self.upstream_loc} non-comment " - f"LOC from {self.upstream_repo}; missing large upstream filenames: " + f"LOC from {self.upstream_repo}; largest unmatched upstream filenames: " f"{missing}" ) @@ -245,15 +245,22 @@ def evaluate_stats( ) cutoff = 1.0 - loc_tolerance imported_stems = imported.normalized_stems + missing_upstream_files = tuple( + file + for file in sorted(upstream.files, key=lambda item: item.loc, reverse=True) + if file.normalized_stem not in imported_stems + ) large_file_cutoff = max(25, int(upstream.loc * loc_tolerance)) missing_files = tuple( - file.path - for file in sorted(upstream.files, key=lambda item: item.loc, reverse=True) - if file.loc >= large_file_cutoff and file.normalized_stem not in imported_stems + file.path for file in missing_upstream_files if file.loc >= large_file_cutoff )[:MAX_MISSING_FILES] file_count_suspicious = upstream.file_count >= 3 and file_ratio < cutoff if loc_ratio >= cutoff and not file_count_suspicious and not missing_files: return None + if not missing_files: + missing_files = tuple(file.path for file in missing_upstream_files)[ + :MAX_MISSING_FILES + ] return AuditFinding( project=project, upstream_repo=upstream_repo, diff --git a/python/tests/test_partial_port_audit.py b/python/tests/test_partial_port_audit.py index 846d9281d5..6962649072 100644 --- a/python/tests/test_partial_port_audit.py +++ b/python/tests/test_partial_port_audit.py @@ -65,6 +65,26 @@ def test_evaluate_stats_flags_missing_large_file() -> None: assert finding.missing_files == ("Lean4/directed_van_kampen.lean",) +def test_evaluate_stats_lists_largest_missing_files_for_ratio_gap() -> None: + """Suspicious partial imports report useful filenames even for small files.""" + imported = LeanStats((LeanFile("LeanPool/Foo/Main.lean", 100, "main"),)) + upstream = LeanStats( + ( + LeanFile("Foo/Main.lean", 100, "main"), + LeanFile("Foo/SkippedA.lean", 80, "skippeda"), + LeanFile("Foo/SkippedB.lean", 60, "skippedb"), + LeanFile("Foo/SkippedC.lean", 40, "skippedc"), + ) + ) + finding = evaluate_stats("LeanPool.Foo", "owner/foo", imported, upstream) + assert finding is not None + assert finding.missing_files == ( + "Foo/SkippedA.lean", + "Foo/SkippedB.lean", + "Foo/SkippedC.lean", + ) + + def test_evaluate_stats_accepts_close_import() -> None: """A near-complete import passes the tolerance check.""" imported = LeanStats((LeanFile("LeanPool/Foo/Main.lean", 920, "main"),))