From ea045db24f4836da46993fa2a9e7bb5076f61f2c Mon Sep 17 00:00:00 2001 From: justincasher Date: Sat, 9 May 2026 15:07:33 -0700 Subject: [PATCH] Add aggregator clone subcommand for shallow candidate clones Shallow-blobless clones every Reservoir + manual candidate repo into aggregator/candidates/raw_data/clones///, deduped by canonical GitHub key, parallelised with a thread pool. Downstream classification (mathlib/cslib/physlean candidate vs. tooling) needs local source files to build a per-repo gist for an LLM. --- python/lean_pool/aggregator/__main__.py | 59 +++++++++ python/lean_pool/aggregator/cloner.py | 169 ++++++++++++++++++++++++ python/tests/aggregator/test_cloner.py | 166 +++++++++++++++++++++++ 3 files changed, 394 insertions(+) create mode 100644 python/lean_pool/aggregator/cloner.py create mode 100644 python/tests/aggregator/test_cloner.py diff --git a/python/lean_pool/aggregator/__main__.py b/python/lean_pool/aggregator/__main__.py index acc99a9c8c..774734f6dd 100644 --- a/python/lean_pool/aggregator/__main__.py +++ b/python/lean_pool/aggregator/__main__.py @@ -5,6 +5,9 @@ - ``fetch``: download the Reservoir manifest and the manual package list into ``raw_data/``. +- ``clone``: shallow-clone every candidate repo into + ``raw_data/clones/`` so downstream classification can read source + files locally. - ``render``: regenerate the candidates README table from those files. """ @@ -13,6 +16,10 @@ import click +from lean_pool.aggregator.cloner import ( + DEFAULT_PARALLELISM, + clone_all, +) from lean_pool.aggregator.manual import ( fetch_manual_packages, load_manual_packages, @@ -32,6 +39,7 @@ DEFAULT_MANUAL_LIST = CANDIDATES_DIR / "manual.txt" DEFAULT_MANUAL_DATA = CANDIDATES_DIR / "raw_data" / "manual_packages.json" DEFAULT_MANUAL_CACHE = CANDIDATES_DIR / "raw_data" / "manual_cache" +DEFAULT_CLONES_DIR = CANDIDATES_DIR / "raw_data" / "clones" DEFAULT_README = CANDIDATES_DIR / "README.md" @@ -96,6 +104,57 @@ def fetch( click.echo(f"No manual list at {manual_list}; skipping manual fetch.") +@cli.command() +@click.option( + "--manifest", + "manifest_path", + type=click.Path(exists=True, dir_okay=False, path_type=Path), + default=DEFAULT_MANIFEST, + show_default=True, + help="Manifest JSON produced by `fetch`.", +) +@click.option( + "--manual-data", + "manual_data_path", + type=click.Path(dir_okay=False, path_type=Path), + default=DEFAULT_MANUAL_DATA, + show_default=True, + help="Manual package metadata produced by `fetch`.", +) +@click.option( + "--clones-dir", + type=click.Path(file_okay=False, path_type=Path), + default=DEFAULT_CLONES_DIR, + show_default=True, + help="Where to write shallow blobless clones.", +) +@click.option( + "--parallelism", + type=int, + default=DEFAULT_PARALLELISM, + show_default=True, + help="Maximum concurrent `git clone` processes.", +) +def clone( + manifest_path: Path, + manual_data_path: Path, + clones_dir: Path, + parallelism: int, +) -> None: + """Shallow-clone every candidate repo into the local cache.""" + with manifest_path.open() as manifest_file: + manifest = json.load(manifest_file) + manual_packages = load_manual_packages(manual_data_path) + packages = list(manifest["packages"]) + manual_packages + cloned_now, already_present, failed = clone_all( + packages, clones_dir, parallelism=parallelism + ) + click.echo( + f"Clones at {clones_dir}: " + f"{cloned_now} new, {already_present} cached, {failed} failed." + ) + + @cli.command() @click.option( "--manifest", diff --git a/python/lean_pool/aggregator/cloner.py b/python/lean_pool/aggregator/cloner.py new file mode 100644 index 0000000000..398fbd288f --- /dev/null +++ b/python/lean_pool/aggregator/cloner.py @@ -0,0 +1,169 @@ +"""Shallow-clone every candidate repo into a local cache directory. + +Each repo is cloned with ``git clone --depth=1 --filter=blob:none``, +which fetches the latest tree without history or blob contents. Files +are downloaded on demand the first time they are read, so an idle +clone is typically a few hundred KB. + +The cache lives at +``aggregator/candidates/raw_data/clones///``, keyed by the +package's canonical GitHub key (lowercased, ``.lean``/``.git`` suffix +stripped) so two manifest entries pointing at the same repo (e.g. +Reservoir's ``leanprover-community/mathlib`` and a manual entry's +``mathlib4``) clone once. +""" + +from __future__ import annotations + +import logging +import shutil +import subprocess +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path + +from lean_pool.aggregator.render import package_key +from lean_pool.aggregator.reservoir import Package + +logger = logging.getLogger(__name__) + +DEFAULT_PARALLELISM = 8 +DEFAULT_CLONE_TIMEOUT_SECONDS = 120.0 + + +def _repo_url(package: Package) -> str | None: + """Return the package's primary repository URL, if available.""" + sources = package.get("sources") or [] + if not sources: + return None + return sources[0].get("repoUrl") or sources[0].get("gitUrl") + + +def _git_clone( + url: str, dest: Path, timeout: float = DEFAULT_CLONE_TIMEOUT_SECONDS +) -> None: + """Run a shallow blobless clone with a timeout.""" + subprocess.run( + [ + "git", + "clone", + "--depth=1", + "--filter=blob:none", + "--quiet", + url, + str(dest), + ], + check=True, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def _target_for_key(clones_dir: Path, key: str) -> Path: + """Map an ``owner/name`` canonical key to a directory under ``clones_dir``.""" + owner, _, name = key.partition("/") + return clones_dir / owner / name + + +def clone_package( + package: Package, + clones_dir: Path, + *, + timeout: float = DEFAULT_CLONE_TIMEOUT_SECONDS, +) -> Path | None: + """Clone one package's repo into ``clones_dir`` if not already cloned. + + Args: + package: A package dict from the Reservoir manifest or the + manual list. + clones_dir: Root directory holding all clones. + timeout: Per-clone timeout in seconds. + + Returns: + The clone path on success or already-present, or ``None`` if + the package has no usable URL or the clone failed. + """ + key = package_key(package) + url = _repo_url(package) + if not key or not url: + logger.debug("Skipping %s: no canonical key or URL", package.get("fullName")) + return None + target = _target_for_key(clones_dir, key) + if target.exists(): + return target + target.parent.mkdir(parents=True, exist_ok=True) + try: + _git_clone(url, target, timeout=timeout) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + stderr = (getattr(exc, "stderr", "") or "").strip() + logger.warning("Clone failed for %s: %s", key, stderr[:200] or exc) + if target.exists(): + shutil.rmtree(target, ignore_errors=True) + return None + return target + + +def _dedupe_by_key(packages: list[Package]) -> list[Package]: + """First-seen-wins dedup by canonical GitHub key.""" + seen: set[str] = set() + deduped: list[Package] = [] + for package in packages: + key = package_key(package) + if key is None or key in seen: + continue + seen.add(key) + deduped.append(package) + return deduped + + +def clone_all( + packages: list[Package], + clones_dir: Path, + *, + parallelism: int = DEFAULT_PARALLELISM, + timeout: float = DEFAULT_CLONE_TIMEOUT_SECONDS, +) -> tuple[int, int, int]: + """Shallow-clone every package in parallel. + + Packages are deduplicated by canonical GitHub key before cloning so + Reservoir + manual overlaps don't clone twice. + + Args: + packages: Combined list of packages to clone. + clones_dir: Root directory for the clones; created if missing. + parallelism: Maximum concurrent ``git clone`` processes. + timeout: Per-clone timeout in seconds. + + Returns: + ``(cloned_now, already_present, failed)``. ``cloned_now`` counts + repos a fresh clone produced this run; ``already_present`` counts + repos that were cached on disk and skipped; ``failed`` counts + repos that errored or had no usable URL. + """ + clones_dir.mkdir(parents=True, exist_ok=True) + queue = _dedupe_by_key(packages) + + cloned_now = 0 + already_present = 0 + failed = 0 + + def _process(package: Package) -> tuple[Path | None, bool]: + key = package_key(package) + url = _repo_url(package) + if not key or not url: + return None, False + existed = _target_for_key(clones_dir, key).exists() + result = clone_package(package, clones_dir, timeout=timeout) + return result, existed + + with ThreadPoolExecutor(max_workers=parallelism) as pool: + futures = [pool.submit(_process, p) for p in queue] + for future in as_completed(futures): + result, existed = future.result() + if result is None: + failed += 1 + elif existed: + already_present += 1 + else: + cloned_now += 1 + return cloned_now, already_present, failed diff --git a/python/tests/aggregator/test_cloner.py b/python/tests/aggregator/test_cloner.py new file mode 100644 index 0000000000..46010af899 --- /dev/null +++ b/python/tests/aggregator/test_cloner.py @@ -0,0 +1,166 @@ +"""Tests for the candidate repo cloner.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, patch + +from lean_pool.aggregator.cloner import ( + _dedupe_by_key, + _repo_url, + _target_for_key, + clone_all, + clone_package, +) +from lean_pool.aggregator.reservoir import Package + + +def _package( + *, + full_name: str = "acme/demo", + repo_url: str | None = "https://github.com/acme/demo", +) -> Package: + """Build a minimal package dict shaped like a Reservoir entry.""" + owner, name = full_name.split("/") + sources = [{"repoUrl": repo_url}] if repo_url else [] + return { # type: ignore[typeddict-item] + "name": name, + "owner": owner, + "fullName": full_name, + "sources": sources, + } + + +def test_repo_url_prefers_repo_url_then_git_url() -> None: + """``repoUrl`` wins; ``gitUrl`` is the fallback when missing.""" + with_repo = _package(repo_url="https://github.com/acme/demo") + only_git: Package = { # type: ignore[typeddict-item] + "fullName": "acme/demo", + "sources": [{"gitUrl": "https://github.com/acme/demo.git"}], + } + no_sources: Package = {"fullName": "acme/demo"} # type: ignore[typeddict-item] + + assert _repo_url(with_repo) == "https://github.com/acme/demo" + assert _repo_url(only_git) == "https://github.com/acme/demo.git" + assert _repo_url(no_sources) is None + + +def test_target_for_key_splits_owner_and_name(tmp_path: Path) -> None: + """The canonical key maps to ``///``.""" + assert _target_for_key(tmp_path, "acme/demo") == tmp_path / "acme" / "demo" + + +def test_dedupe_by_key_keeps_first_occurrence() -> None: + """Two URLs that differ only by case or ``.lean`` suffix collapse to one.""" + first = _package(full_name="Acme/Demo", repo_url="https://github.com/Acme/Demo") + second = _package( + full_name="acme/demo.lean", repo_url="https://github.com/acme/Demo.lean" + ) + third = _package(full_name="other/repo", repo_url="https://github.com/other/repo") + + deduped = _dedupe_by_key([first, second, third]) + + assert [p["fullName"] for p in deduped] == ["Acme/Demo", "other/repo"] + + +def test_clone_package_invokes_git_with_blobless_flags(tmp_path: Path) -> None: + """``clone_package`` runs git with the expected shallow blobless args.""" + package = _package() + + with patch("lean_pool.aggregator.cloner.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + result = clone_package(package, tmp_path) + + assert result == tmp_path / "acme" / "demo" + args, kwargs = mock_run.call_args + cmd = args[0] + assert cmd[:2] == ["git", "clone"] + assert "--depth=1" in cmd + assert "--filter=blob:none" in cmd + assert cmd[-2] == "https://github.com/acme/demo" + assert cmd[-1].endswith("acme/demo") + assert kwargs.get("check") is True + + +def test_clone_package_skips_when_target_exists(tmp_path: Path) -> None: + """An existing clone directory short-circuits before ``git`` runs.""" + package = _package() + existing = tmp_path / "acme" / "demo" + existing.mkdir(parents=True) + + with patch("lean_pool.aggregator.cloner.subprocess.run") as mock_run: + result = clone_package(package, tmp_path) + + assert result == existing + mock_run.assert_not_called() + + +def test_clone_package_returns_none_without_url(tmp_path: Path) -> None: + """Packages without a repo URL are skipped, not cloned.""" + package = _package(repo_url=None) + + with patch("lean_pool.aggregator.cloner.subprocess.run") as mock_run: + result = clone_package(package, tmp_path) + + assert result is None + mock_run.assert_not_called() + + +def test_clone_package_cleans_partial_dir_on_failure(tmp_path: Path) -> None: + """A failed clone leaves no stray directory behind.""" + package = _package() + + def _fake_run(*args: object, **kwargs: object) -> None: + # Simulate git creating the dest dir, then failing mid-clone. + target = tmp_path / "acme" / "demo" + target.mkdir(parents=True, exist_ok=True) + raise subprocess.CalledProcessError( + returncode=128, + cmd=args[0], + stderr="fatal: repository not found", + ) + + with patch("lean_pool.aggregator.cloner.subprocess.run", side_effect=_fake_run): + result = clone_package(package, tmp_path) + + assert result is None + assert not (tmp_path / "acme" / "demo").exists() + + +def test_clone_all_dedupes_and_counts(tmp_path: Path) -> None: + """Run summary counts cloned, cached, and failed buckets correctly.""" + duplicate_a = _package( + full_name="Acme/Demo", repo_url="https://github.com/Acme/Demo" + ) + duplicate_b = _package( + full_name="acme/demo", repo_url="https://github.com/acme/demo" + ) + cached = _package( + full_name="cached/repo", repo_url="https://github.com/cached/repo" + ) + failing = _package(full_name="bad/repo", repo_url="https://github.com/bad/repo") + no_url = _package(full_name="missing/url", repo_url=None) + + # Pre-create the cached entry so it shows up in already_present. + (tmp_path / "cached" / "repo").mkdir(parents=True) + + def _fake_run(args: list[str], **_: object) -> MagicMock: + url = args[-2] + if "bad/repo" in url: + raise subprocess.CalledProcessError( + returncode=128, cmd=args, stderr="fatal: not found" + ) + Path(args[-1]).mkdir(parents=True, exist_ok=True) + return MagicMock(returncode=0) + + with patch("lean_pool.aggregator.cloner.subprocess.run", side_effect=_fake_run): + cloned_now, already_present, failed = clone_all( + [duplicate_a, duplicate_b, cached, failing, no_url], + tmp_path, + parallelism=2, + ) + + assert cloned_now == 1 # the deduped Acme/Demo pair clones once + assert already_present == 1 # cached/repo + assert failed == 2 # bad/repo + missing/url