diff --git a/BACKLOG.md b/BACKLOG.md
index d5825cc..215ffbd 100644
--- a/BACKLOG.md
+++ b/BACKLOG.md
@@ -190,23 +190,26 @@ monorepo; fetch each via raw (free, per ①). Replaces 1 Contents call per file.
 
 ---
 
-### ⑤ Batch repo metadata via GraphQL (≤100 repos/call)
-**Priority: medium**
-Replace per-repo `/repos/{o}/{r}` REST calls with a GraphQL query (stars/
-pushed_at/default_branch) for up to 100 repos at once — ~1–2 points of the
-separate 5,000-point/hr GraphQL pool.
-
-**Files:** `crawlers/base.py` (new GraphQL helper), crawler call sites.
+### ⑤ Batch repo metadata via GraphQL (≤100 repos/call) — DONE
+**Done.** `fetch_repo_metadata_batch` in `crawlers/base.py` fetches
+stars/pushedAt/defaultBranch/topics for ≤100 repos per GraphQL POST (separate
+5,000-point/hr pool); wired into the topic crawler with a per-repo REST fallback.
 
 ---
 
-### ④ Reduce search/code reliance (the 10/min stall)
-**Priority: medium**
-Cache the discovered repo list; run discovery with date filters
-(`pushed:>last-run`) so search surfaces only new repos. Prefer repo/topic search
-(30/min) over code search (10/min) where possible.
+### ④ Reduce search/code reliance (the 10/min stall) — REVERTED, needs redesign
+**Status: attempted and reverted.** The date-filter approach
+(`topic:X pushed:>last-run`) is **unsound for topic discovery**: a repo that
+*adds a skill topic without a new commit* keeps its old `pushed_at`, so it is
+filtered out and the advancing watermark makes it permanently undiscoverable —
+and GitHub search has no "topic-added-since" qualifier. A correct incremental
+discovery needs a different mechanism (e.g. enumerate the full topic result set
+each run and diff against a persisted repo-list, never a `pushed_at` watermark),
+or accept that discovery search is cheap enough (repo-search is 30/min, never the
+bottleneck) and leave it full. For code-search-heavy crawlers (skillsmp shards),
+the separate lever is to lean on repo-search over code-search where possible.
 
-**Files:** `crawlers/skillsmp_crawler.py`, `crawlers/topic_crawler.py`, `crawlers/marketplace_crawler.py`.
+**Files:** `crawlers/skillsmp_crawler.py`, `crawlers/marketplace_crawler.py`.
 
 ---
 
@@ -218,13 +221,10 @@ overhead — only if ①–⑥ are insufficient.
 
 **Files:** `crawlers/base.py` (session/token rotation).
 
-### ⑥b Cache find_skill_md_paths by repo + HEAD SHA
-**Priority: medium** *(follow-up from the ①②③ impact measurement)*
-After ①②③, the residual metered cost per repo is the recursive Trees call in
-`find_skill_md_paths` (one metered call per repo every run, even when nothing
-changed). Cache the `{path: blob_sha}` result keyed by `repo + HEAD commit SHA`
-(one cheap `fetch_commit_sha` call, or reuse the ETag-cached metadata's known
-HEAD), and skip the Trees call when HEAD is unchanged. This is what makes a warm
-run approach zero metered cost *per repo*, not just per skill.
-
-**Files:** `crawlers/base.py` (`find_skill_md_paths`, `fetch_commit_sha`).
+### ⑥b Cache find_skill_md_paths — DONE
+**Done.** `find_skill_md_paths_cached` in `crawlers/base.py` skips the recursive
+Trees call when a repo's `(pushed_at, default_branch)` is unchanged, reusing the
+cached `{path: blob_sha}` map (keyed on default branch because a branch change
+re-resolves `HEAD` to a different tree without bumping `pushed_at`; empty/failed
+lookups are never cached). Wired into the topic crawler. Makes a warm run
+approach zero metered cost *per unchanged repo*.
diff --git a/crawlers/base.py b/crawlers/base.py
index 1ea7cd1..84f375c 100644
--- a/crawlers/base.py
+++ b/crawlers/base.py
@@ -9,6 +9,7 @@
   fetch_repo_metadata()         - Fetch stars/pushed_at/topics/branch for a repo
   fetch_repo_metadata_with_etag() - ETag-aware version; returns (dict|None, etag|None)
   fetch_repo_metadata_cached()   - Like fetch_repo_metadata_with_etag but with a persistent cache (304 = zero quota)
+  fetch_repo_metadata_batch()   - Batch-fetch metadata for up to 100 repos via GraphQL (one POST per 100)
   fetch_commit_sha()            - Fetch HEAD commit SHA for a repo (one API call)
   find_skill_md_paths()         - Find all SKILL.md paths → {path: blob_sha} dict
   load_filter_cache()           - Load set of filtered-out canonical URLs
@@ -25,6 +26,9 @@
   fetch_skill_md_cached()       - Fetch SKILL.md, skipping when blob SHA is cached
   load_content_cache()          - Load persistent blob-SHA -> content cache
   save_content_cache()          - Persist blob-SHA -> content cache
+  find_skill_md_paths_cached()  - Find SKILL.md paths, skipping Trees API when pushed_at is unchanged
+  load_tree_cache()             - Load persistent pushed_at -> {path: sha} tree cache
+  save_tree_cache()             - Persist pushed_at -> {path: sha} tree cache
 """
 
 from __future__ import annotations
@@ -562,6 +566,80 @@ def find_skill_md_paths(session, repo_full_name: str) -> dict[str, str]:
     return paths
 
 
+def find_skill_md_paths_cached(
+    session,
+    repo_full_name: str,
+    pushed_at: str,
+    default_branch: str,
+    tree_cache: dict,
+) -> dict[str, str]:
+    """Return SKILL.md paths for a repo, skipping the Trees API call when unchanged.
+
+    Uses ``(pushed_at, default_branch)`` as the freshness key.  If both match what is
+    stored in ``tree_cache``, the cached ``{path: sha}`` mapping is returned
+    immediately without any API call.  Otherwise ``find_skill_md_paths`` is called and
+    the result is stored back into ``tree_cache`` (mutated in place).
+
+    ``default_branch`` is part of the key because ``find_skill_md_paths`` walks
+    ``git/trees/HEAD`` (the default branch): a default-branch change resolves HEAD to a
+    different tree even when ``pushed_at`` is unchanged, so keying on ``pushed_at``
+    alone would return stale paths/blob SHAs from the old branch.
+
+    An empty or falsy ``pushed_at`` always calls the API and never caches the result
+    because we cannot prove freshness without a timestamp.
+
+    Args:
+        session:        A requests.Session from make_session().
+        repo_full_name: "{owner}/{repo}" string.
+        pushed_at:      The repo's ``pushed_at`` ISO-8601 string from the metadata API.
+                        Pass ``""`` (or any falsy value) to force a live fetch.
+        default_branch: The repo's current default branch name.
+        tree_cache:     Mutable dict that persists across calls within a crawl run.
+                        Shape: ``{repo: {"pushed_at": str, "default_branch": str, "paths": dict}}``.
+
+    Returns:
+        Dict mapping SKILL.md path → blob SHA (same contract as ``find_skill_md_paths``).
+    """
+    entry = tree_cache.get(repo_full_name, {})
+    if (
+        pushed_at
+        and entry.get("pushed_at") == pushed_at
+        and entry.get("default_branch") == default_branch
+    ):
+        return entry["paths"]
+
+    paths = find_skill_md_paths(session, repo_full_name)
+    # Only cache a NON-empty result. find_skill_md_paths returns {} both for a
+    # repo with no SKILL.md and for a transient Trees/Search API failure; caching
+    # {} under the unchanged pushed_at would pin a repo that actually has skills to
+    # "empty" forever (it never re-fetches until pushed again). Genuinely empty
+    # repos are cheaply re-checked and short-circuited by the crawler's filter cache.
+    if pushed_at and paths:
+        tree_cache[repo_full_name] = {
+            "pushed_at": pushed_at,
+            "default_branch": default_branch,
+            "paths": paths,
+        }
+    return paths
+
+
+def load_tree_cache(path: str) -> dict:
+    """Load the persistent Trees-API path cache, or {} if absent/corrupt.
+
+    Thin alias for ``load_meta_cache`` — same JSON format, different file.
+    Cache shape: ``{repo_full_name: {"pushed_at": str, "paths": {path: sha}}}``.
+    """
+    return load_meta_cache(path)
+
+
+def save_tree_cache(cache: dict, path: str) -> None:
+    """Persist the Trees-API path cache atomically.
+
+    Thin alias for ``save_meta_cache`` — same atomic-write semantics, different file.
+    """
+    save_meta_cache(cache, path)
+
+
 def _find_skill_md_via_search(session, repo_full_name: str) -> dict[str, str]:
     """Find SKILL.md paths via Code Search API, scoped to one repo.
 
@@ -823,6 +901,140 @@ def fetch_repo_metadata_cached(session, repo_full_name: str, cache: dict) -> dic
     return meta
 
 
+_GRAPHQL_BATCH_SIZE = 100
+
+
+def fetch_repo_metadata_batch(
+    session,
+    repo_full_names: list[str],
+) -> dict[str, dict]:
+    """Batch-fetch metadata for up to 100 repos per request via the GitHub GraphQL API.
+
+    Uses one GraphQL POST per chunk of ≤100 repos (aliased fields), consuming from
+    the separate 5,000-point/hr GraphQL quota rather than the REST quota.
+
+    Args:
+        session:         A requests.Session from make_session() (carries Authorization header).
+        repo_full_names: List of "{owner}/{repo}" strings to fetch.
+
+    Returns:
+        Dict mapping full_name → {stargazers_count, pushed_at, topics, description,
+        default_branch} for every repo that resolved successfully.  Repos that GitHub
+        returns as null (deleted, renamed, private, or otherwise inaccessible) are
+        simply absent from the result — callers should fall back to the per-repo REST
+        path for missing entries.
+
+    Error handling:
+        - network / non-200 / unparseable JSON: logs a warning, skips that chunk.
+        - GraphQL ``errors`` array: resolved aliases in ``data`` are still returned;
+          only the null aliases are omitted.
+    """
+    graphql_url = f"{GITHUB_API}/graphql"
+    result: dict[str, dict] = {}
+
+    # Filter entries that don't have a "/" — they cannot be split into owner/name.
+    valid: list[tuple[int, str, str, str]] = []  # (original_index, alias, owner, name)
+    for i, full_name in enumerate(repo_full_names):
+        if "/" not in full_name:
+            logger.warning("fetch_repo_metadata_batch: skipping malformed entry %r", full_name)
+            continue
+        owner, name = full_name.split("/", 1)
+        valid.append((i, f"r{i}", owner, name))
+
+    # Process in chunks of _GRAPHQL_BATCH_SIZE
+    for chunk_start in range(0, len(valid), _GRAPHQL_BATCH_SIZE):
+        chunk = valid[chunk_start: chunk_start + _GRAPHQL_BATCH_SIZE]
+
+        # Build variable declarations and field aliases for this chunk
+        var_decls: list[str] = []
+        field_aliases: list[str] = []
+        variables: dict[str, str] = {}
+
+        for _orig_idx, alias, owner, name in chunk:
+            var_decls.append(f"${alias}Owner:String! ${alias}Name:String!")
+            field_aliases.append(
+                f"{alias}: repository(owner:${alias}Owner, name:${alias}Name) {{"
+                f" stargazerCount pushedAt description"
+                f" defaultBranchRef {{ name }}"
+                f" repositoryTopics(first:20) {{ nodes {{ topic {{ name }} }} }}"
+                f" }}"
+            )
+            variables[f"{alias}Owner"] = owner
+            variables[f"{alias}Name"] = name
+
+        query = "query(" + " ".join(var_decls) + ") { " + " ".join(field_aliases) + " }"
+
+        try:
+            resp = session.post(
+                graphql_url,
+                json={"query": query, "variables": variables},
+                timeout=30,
+            )
+        except requests.RequestException as exc:
+            logger.warning(
+                "fetch_repo_metadata_batch: network error on chunk starting at index %d: %s",
+                chunk_start,
+                exc,
+            )
+            continue
+
+        record_request(graphql_url, resp.status_code)
+
+        if resp.status_code != 200:
+            logger.warning(
+                "fetch_repo_metadata_batch: HTTP %s for chunk at index %d",
+                resp.status_code,
+                chunk_start,
+            )
+            continue
+
+        try:
+            payload = resp.json()
+        except ValueError as exc:
+            logger.warning(
+                "fetch_repo_metadata_batch: unparseable JSON for chunk at index %d: %s",
+                chunk_start,
+                exc,
+            )
+            continue
+
+        if payload.get("errors"):
+            logger.debug(
+                "fetch_repo_metadata_batch: GraphQL errors in chunk (still using resolved data): %s",
+                payload["errors"],
+            )
+
+        data = payload.get("data") or {}
+
+        # Map resolved aliases back to full_name
+        alias_to_full: dict[str, str] = {
+            alias: repo_full_names[orig_idx]
+            for orig_idx, alias, _owner, _name in chunk
+        }
+
+        for alias, node in data.items():
+            if node is None:
+                continue  # repo not found / inaccessible
+            full_name = alias_to_full.get(alias)
+            if full_name is None:
+                continue
+            default_branch_ref = node.get("defaultBranchRef") or {}
+            topics = [
+                n["topic"]["name"]
+                for n in (node.get("repositoryTopics") or {}).get("nodes", [])
+                if n and n.get("topic")
+            ]
+            result[full_name] = {
+                "stargazers_count": node.get("stargazerCount") or 0,
+                "pushed_at": node.get("pushedAt") or "",
+                "description": node.get("description") or "",
+                "default_branch": default_branch_ref.get("name") or "main",
+                "topics": topics,
+            }
+
+    return result
+
+
 def fetch_commit_sha(session, repo_full_name: str) -> str | None:
     """Fetch the HEAD commit SHA for a repo (one API call).
 
diff --git a/crawlers/topic_crawler.py b/crawlers/topic_crawler.py
index 9186d8a..bcd3d23 100644
--- a/crawlers/topic_crawler.py
+++ b/crawlers/topic_crawler.py
@@ -35,18 +35,21 @@
 from crawlers.base import (
     GITHUB_API,
     add_to_filter_cache,
+    fetch_repo_metadata_batch,
     fetch_repo_metadata_cached,
     fetch_skill_md_cached,
-    find_skill_md_paths,
+    find_skill_md_paths_cached,
     github_get,
     infer_platforms,
     load_content_cache,
     load_filter_cache,
     load_meta_cache,
+    load_tree_cache,
     make_session,
     parse_frontmatter,
     save_content_cache,
     save_meta_cache,
+    save_tree_cache,
     write_jsonl,
 )
 
@@ -100,12 +103,19 @@ def _discover_topic_repos(session, limit: int = 1000) -> list[str]:
     Paginates each query (up to 1000 results per query as GitHub allows).
     Returns a deduplicated list of "{owner}/{repo}" full names, at most `limit`.
 
+    Note: every run searches all topics in full. Topic membership changes
+    independently of ``pushed_at`` and GitHub search has no "topic-added-since"
+    qualifier, so a date-filtered incremental discovery cannot be correct here —
+    a repo that gains a skill topic without a new commit would be missed. The
+    crawl stays cheap via the metadata/tree/content caches downstream, not by
+    narrowing discovery.
+
     Args:
         session: A requests.Session from make_session().
         limit:   Total maximum unique repos to return across all queries.
 
     Returns:
-        Deduplicated list of full repo names.
+        Deduplicated list of full repo names, at most `limit`.
     """
     seen: set[str] = set()
     results: list[str] = []
@@ -218,9 +228,10 @@ def run(
 
     session = make_session(token=token)
 
-    # Load ETag metadata cache and blob-SHA content cache
+    # Load ETag metadata cache, blob-SHA content cache, and Trees-path cache
     meta_cache = load_meta_cache("data/crawl_state/repo_meta_cache.json")
     content_cache = load_content_cache("data/crawl_state/content_cache.json")
+    tree_cache = load_tree_cache("data/crawl_state/tree_cache.json")
 
     # Load filter cache (repos known to have no SKILL.md)
     filter_cache: set[str] = set()
@@ -253,9 +264,30 @@ def run(
                     pass
         log.info("Resume mode: %d skill keys already in output", len(existing_skill_keys))
 
-    # Discover repos via topic search
+    # Discover repos via topic search (full each run — see _discover_topic_repos note).
     discovered = _discover_topic_repos(session, limit=1000)
 
+    # Pre-filter discovered repos before the GraphQL batch to avoid wasting quota on
+    # repos that will be skipped anyway (already_covered or in filter_cache).
+    to_process = []
+    for full_name in discovered:
+        repo_url = f"https://github.com/{full_name}"
+        canon_url = repo_url.lower()
+        if canon_url in already_covered:
+            log.debug("Skipping %s: already covered by another crawler", full_name)
+            continue
+        if canon_url in filter_cache or repo_url in filter_cache:
+            log.debug("Skipping %s: in filter cache", full_name)
+            continue
+        to_process.append(full_name)
+
+    # Lazily batch-fetch metadata in chunks of 100 as the loop consumes repos.
+    # A small --limit should not pay GraphQL for repos that are never processed.
+    # Each chunk of ≤100 repos costs one GraphQL POST (separate 5k-point/hr pool).
+    # Repos absent from any chunk's result fall back to per-repo REST below.
+    batch_meta: dict = {}
+    _batched_upto = 0
+
     # Cache (meta, skill_md_paths) per repo to avoid repeated API calls
     _repo_cache: dict[str, tuple[dict, list[str]]] = {}
 
@@ -264,32 +296,37 @@ def run(
 
     records: list[dict] = []
 
-    for full_name in discovered:
+    for idx, full_name in enumerate(to_process):
         if limit is not None and len(records) >= limit:
             log.info("Reached limit of %d records; stopping.", limit)
             break
 
-        repo_url = f"https://github.com/{full_name}"
-        canon_url = repo_url.lower()
+        # Lazily batch-fetch metadata in chunks of 100 as the loop reaches them, so a
+        # small --limit doesn't pay GraphQL for repos that are never processed.
+        if idx >= _batched_upto:
+            chunk = to_process[_batched_upto:_batched_upto + 100]
+            batch_meta.update(fetch_repo_metadata_batch(session, chunk))
+            _batched_upto += 100
 
-        # Skip repos already covered by other crawlers
-        if canon_url in already_covered:
-            log.debug("Skipping %s: already covered by another crawler", full_name)
-            continue
-
-        # Skip repos with no SKILL.md (from filter cache)
-        if canon_url in filter_cache or repo_url in filter_cache:
-            log.debug("Skipping %s: in filter cache", full_name)
-            continue
+        repo_url = f"https://github.com/{full_name}"
 
         # Fetch metadata + SKILL.md paths
         if full_name not in _repo_cache:
             try:
-                meta = fetch_repo_metadata_cached(session, full_name, meta_cache)
+                # Use GraphQL batch result when available; fall back to per-repo REST
+                meta = batch_meta.get(full_name) or fetch_repo_metadata_cached(
+                    session, full_name, meta_cache
+                )
             except RuntimeError as exc:
                 log.warning("Could not fetch metadata for %s: %s", full_name, exc)
                 continue
-            skill_md_paths = find_skill_md_paths(session, full_name)
+            skill_md_paths = find_skill_md_paths_cached(
+                session,
+                full_name,
+                meta.get("pushed_at", ""),
+                meta.get("default_branch", "main"),
+                tree_cache,
+            )
             _repo_cache[full_name] = (meta, skill_md_paths)
             if not skill_md_paths:
                 if filter_cache_path:
@@ -360,6 +397,7 @@ def run(
 
     save_meta_cache(meta_cache, "data/crawl_state/repo_meta_cache.json")
     save_content_cache(content_cache, "data/crawl_state/content_cache.json")
+    save_tree_cache(tree_cache, "data/crawl_state/tree_cache.json")
 
     written = write_jsonl(records, output_path, append=resume)
     log.info("Topic crawler done: %d records written to %s", written, output_path)
@@ -398,7 +436,7 @@ def _build_parser() -> argparse.ArgumentParser:
         "--mode",
         choices=["full", "incremental", "metadata", "discover"],
         default="full",
-        help="Crawl mode: full=complete re-crawl, incremental=changed repos only, metadata=stars/ETags only, discover=new repos since last run",
+        help="Crawl mode: full=complete re-crawl, incremental=changed repos only, metadata=stars/ETags only, discover=alias for full discovery",
     )
     p.add_argument(
         "--resume",
diff --git a/docs/crawler-rate-limit-strategy.md b/docs/crawler-rate-limit-strategy.md
index a827c76..ae98f8a 100644
--- a/docs/crawler-rate-limit-strategy.md
+++ b/docs/crawler-rate-limit-strategy.md
@@ -74,11 +74,19 @@ canonical URL → fetch each **once**. Today a repo found by marketplace + topic
 skillsmp is fetched 3×. Pair with a per-run in-memory (and optionally on-disk)
 cache of fetched metadata/content keyed by canonical repo URL.
 
-### ④ Cut `search/code` reliance — **medium impact, medium effort**
-Cache the discovered repo list and run discovery with date filters
-(`pushed:>last-run`) so search only surfaces *new* repos (few calls). Prefer
-repo/topic search (30/min) over code search (10/min) where the query allows.
-Removes the 60s-cooldown stalls that dominate wall-clock.
+### ④ Cut `search/code` reliance — **attempted, reverted (date-filter is unsound for topic discovery)**
+The obvious approach — date-filter discovery with `pushed:>last-run` so search
+only surfaces *new* repos — was implemented and then **reverted**. It is
+fundamentally wrong for *topic* discovery: a repo that adds a skill topic without
+a new commit keeps its old `pushed_at`, so `pushed:>since` excludes it, the
+advancing watermark moves past it, and it becomes permanently undiscoverable —
+and GitHub search has no "topic-added-since" qualifier to compensate. (~10 codex
+findings chased the watermark before identifying the root flaw.) A correct
+incremental discovery must enumerate the full topic result set each run and diff
+against a persisted repo-list, not rely on a `pushed_at` watermark. Topic
+discovery is left full (repo-search at 30/min was never the bottleneck); the
+remaining lever is preferring repo-search over code-search in the
+code-search-heavy crawlers (skillsmp shards).
 
 ### ⑤ Batch metadata via GraphQL — **medium impact, medium effort**
 Replace per-repo `/repos/{o}/{r}` REST calls with one GraphQL query for ≤100
diff --git a/tests/crawlers/test_base.py b/tests/crawlers/test_base.py
index d320bd6..5e383d9 100644
--- a/tests/crawlers/test_base.py
+++ b/tests/crawlers/test_base.py
@@ -12,6 +12,7 @@
     decode_b64_utf8,
     extract_github_url,
     fetch_commit_sha,
+    fetch_repo_metadata_batch,
     fetch_skill_md,
     load_crawl_state,
     load_existing_records,
@@ -21,8 +22,9 @@
     save_crawl_state,
     write_jsonl,
     fetch_skill_md_cached,
-    load_content_cache,
-    save_content_cache,
+    find_skill_md_paths_cached,
+    load_tree_cache,
+    save_tree_cache,
 )
 
 
@@ -1029,3 +1031,251 @@ def test_empty_sha_always_fetches(self):
             content = fetch_skill_md_cached(MagicMock(), "u/r", "SKILL.md", "", "main", cache)
         assert content == "real"
         mock_fetch.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# TestTreeCache (load_tree_cache / save_tree_cache aliases)
+# ---------------------------------------------------------------------------
+
+class TestTreeCacheAliases:
+    def test_load_tree_cache_returns_empty_for_missing_file(self, tmp_path):
+        result = load_tree_cache(str(tmp_path / "nonexistent.json"))
+        assert result == {}
+
+    def test_save_and_load_round_trip(self, tmp_path):
+        path = str(tmp_path / "tree_cache.json")
+        cache = {"u/r": {"pushed_at": "2026-01-01T00:00:00Z", "paths": {"SKILL.md": "sha1"}}}
+        save_tree_cache(cache, path)
+        result = load_tree_cache(path)
+        assert result == cache
+
+    def test_load_tree_cache_returns_empty_on_corrupt_file(self, tmp_path):
+        p = tmp_path / "corrupt.json"
+        p.write_text("not valid json")
+        result = load_tree_cache(str(p))
+        assert result == {}
+
+
+# ---------------------------------------------------------------------------
+# TestFindSkillMdPathsCached
+# ---------------------------------------------------------------------------
+
+class TestFindSkillMdPathsCached:
+    def test_cache_hit_skips_tree_call(self):
+        """Pre-seeded cache with matching pushed_at and default_branch → no API call made."""
+        tree_cache = {
+            "u/r": {"pushed_at": "2026-01-01T00:00:00Z", "default_branch": "main", "paths": {"SKILL.md": "sha1"}}
+        }
+        with patch("crawlers.base.find_skill_md_paths") as mock_tree:
+            result = find_skill_md_paths_cached(
+                MagicMock(), "u/r", "2026-01-01T00:00:00Z", "main", tree_cache
+            )
+        assert result == {"SKILL.md": "sha1"}
+        mock_tree.assert_not_called()
+
+    def test_cache_miss_calls_and_stores(self):
+        """Empty cache → calls find_skill_md_paths and stores result."""
+        tree_cache = {}
+        with patch("crawlers.base.find_skill_md_paths", return_value={"SKILL.md": "sha9"}) as mock_tree:
+            result = find_skill_md_paths_cached(
+                MagicMock(), "u/r", "2026-02-02T00:00:00Z", "main", tree_cache
+            )
+        assert result == {"SKILL.md": "sha9"}
+        mock_tree.assert_called_once()
+        assert tree_cache["u/r"] == {
+            "pushed_at": "2026-02-02T00:00:00Z",
+            "default_branch": "main",
+            "paths": {"SKILL.md": "sha9"},
+        }
+
+    def test_empty_result_not_cached(self):
+        """An empty result (no SKILL.md OR a transient Trees failure, both {}) must
+        NOT be cached — caching it would pin a repo with skills to empty forever."""
+        tree_cache = {}
+        with patch("crawlers.base.find_skill_md_paths", return_value={}) as mock_tree:
+            result = find_skill_md_paths_cached(
+                MagicMock(), "u/r", "2026-02-02T00:00:00Z", "main", tree_cache
+            )
+        assert result == {}
+        mock_tree.assert_called_once()
+        assert tree_cache == {}  # not cached
+
+    def test_changed_pushed_at_refetches(self):
+        """Stale pushed_at → refetches and updates cache entry."""
+        tree_cache = {
+            "u/r": {"pushed_at": "2026-01-01T00:00:00Z", "default_branch": "main", "paths": {"SKILL.md": "old_sha"}}
+        }
+        new_paths = {"SKILL.md": "new_sha", "sub/SKILL.md": "abc"}
+        with patch("crawlers.base.find_skill_md_paths", return_value=new_paths) as mock_tree:
+            result = find_skill_md_paths_cached(
+                MagicMock(), "u/r", "2026-03-15T12:00:00Z", "main", tree_cache
+            )
+        mock_tree.assert_called_once()
+        assert result == new_paths
+        assert tree_cache["u/r"]["pushed_at"] == "2026-03-15T12:00:00Z"
+        assert tree_cache["u/r"]["paths"] == new_paths
+
+    def test_empty_pushed_at_always_calls_and_does_not_cache(self):
+        """Falsy pushed_at → always calls API and never caches the result."""
+        tree_cache = {}
+        with patch("crawlers.base.find_skill_md_paths", return_value={"SKILL.md": "x"}) as mock_tree:
+            result = find_skill_md_paths_cached(
+                MagicMock(), "u/r", "", "main", tree_cache
+            )
+        assert result == {"SKILL.md": "x"}
+        mock_tree.assert_called_once()
+        assert tree_cache == {}  # nothing cached
+
+    def test_changed_default_branch_refetches(self):
+        """Same pushed_at but different default_branch → cache miss; refetches and updates."""
+        tree_cache = {
+            "u/r": {"pushed_at": "2026-01-01T00:00:00Z", "default_branch": "main", "paths": {"SKILL.md": "old"}}
+        }
+        with patch("crawlers.base.find_skill_md_paths", return_value={"SKILL.md": "new"}) as mock_tree:
+            result = find_skill_md_paths_cached(
+                MagicMock(), "u/r", "2026-01-01T00:00:00Z", "develop", tree_cache
+            )
+        mock_tree.assert_called_once()
+        assert result == {"SKILL.md": "new"}
+        assert tree_cache["u/r"]["default_branch"] == "develop"
+        assert tree_cache["u/r"]["paths"] == {"SKILL.md": "new"}
+
+
+# ---------------------------------------------------------------------------
+# TestFetchRepoMetadataBatch
+# ---------------------------------------------------------------------------
+
+def _make_graphql_node(
+    stargazer_count=5,
+    pushed_at="2026-01-01T00:00:00Z",
+    description="A skill.",
+    default_branch="main",
+    topics=("python", "ai"),
+):
+    """Build a fake GraphQL repository node dict."""
+    return {
+        "stargazerCount": stargazer_count,
+        "pushedAt": pushed_at,
+        "description": description,
+        "defaultBranchRef": {"name": default_branch},
+        "repositoryTopics": {
+            "nodes": [{"topic": {"name": t}} for t in topics],
+        },
+    }
+
+
+def _make_graphql_response(aliases: dict):
+    """Wrap alias→node mapping in a fake GraphQL response envelope."""
+    return {"data": aliases}
+
+
+class TestFetchRepoMetadataBatch:
+    """Unit tests for fetch_repo_metadata_batch() — no network."""
+
+    def _mock_session(self, json_body, status_code=200):
+        """Return a MagicMock session whose .post() returns a fake response."""
+        resp = MagicMock()
+        resp.status_code = status_code
+        resp.json.return_value = json_body
+        session = MagicMock()
+        session.post.return_value = resp
+        return session
+
+    def test_parses_batch_response(self):
+        """Two repos resolved → both are in the result with correct flattened fields."""
+        payload = _make_graphql_response({
+            "r0": _make_graphql_node(
+                stargazer_count=10,
+                pushed_at="2026-03-01T00:00:00Z",
+                description="skill a",
+                default_branch="main",
+                topics=["python"],
+            ),
+            "r1": _make_graphql_node(
+                stargazer_count=20,
+                pushed_at="2026-04-01T00:00:00Z",
+                description="skill b",
+                default_branch="develop",
+                topics=["go", "cli"],
+            ),
+        })
+        session = self._mock_session(payload)
+        result = fetch_repo_metadata_batch(session, ["o/a", "o/b"])
+
+        assert set(result.keys()) == {"o/a", "o/b"}
+
+        a = result["o/a"]
+        assert a["stargazers_count"] == 10
+        assert a["pushed_at"] == "2026-03-01T00:00:00Z"
+        assert a["description"] == "skill a"
+        assert a["default_branch"] == "main"
+        assert a["topics"] == ["python"]
+
+        b = result["o/b"]
+        assert b["stargazers_count"] == 20
+        assert b["default_branch"] == "develop"
+        assert b["topics"] == ["go", "cli"]
+
+    def test_null_repo_is_omitted(self):
+        """An alias whose value is null is absent from the result."""
+        payload = _make_graphql_response({
+            "r0": _make_graphql_node(stargazer_count=5),
+            "r1": None,
+        })
+        session = self._mock_session(payload)
+        result = fetch_repo_metadata_batch(session, ["o/a", "o/b"])
+
+        assert "o/a" in result
+        assert "o/b" not in result
+
+    def test_partial_errors_still_returns_resolved(self):
+        """data with one resolved repo AND a top-level errors list → resolved repo returned."""
+        payload = {
+            "data": {
+                "r0": _make_graphql_node(stargazer_count=7),
+                "r1": None,
+            },
+            "errors": [{"message": "Could not resolve to a Repository with the name 'o/missing'."}],
+        }
+        session = self._mock_session(payload)
+        result = fetch_repo_metadata_batch(session, ["o/a", "o/missing"])
+
+        assert "o/a" in result
+        assert result["o/a"]["stargazers_count"] == 7
+        assert "o/missing" not in result
+
+    def test_non_200_returns_empty(self):
+        """HTTP 500 → {} with no exception raised."""
+        session = self._mock_session({}, status_code=500)
+        result = fetch_repo_metadata_batch(session, ["o/a", "o/b"])
+
+        assert result == {}
+        # No exception propagated
+        session.post.assert_called_once()
+
+    def test_chunks_over_100(self):
+        """150 repos → two POST calls (100 + 50)."""
+        # Both chunks return empty data (all null / absent) — we just count POSTs.
+        payload = {"data": {}}
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = payload
+        session = MagicMock()
+        session.post.return_value = resp
+
+        names = [f"owner/repo{i}" for i in range(150)]
+        fetch_repo_metadata_batch(session, names)
+
+        assert session.post.call_count == 2
+
+    def test_counts_graphql_request(self):
+        """A successful batch POST increments the 'graphql' API counter."""
+        from crawlers.base import reset_api_counters, get_api_counters
+
+        payload = _make_graphql_response({"r0": _make_graphql_node()})
+        session = self._mock_session(payload)
+
+        reset_api_counters()
+        fetch_repo_metadata_batch(session, ["o/a"])
+
+        assert get_api_counters()["graphql"] >= 1
diff --git a/tests/crawlers/test_topic_crawler.py b/tests/crawlers/test_topic_crawler.py
index 728dbad..66b8d23 100644
--- a/tests/crawlers/test_topic_crawler.py
+++ b/tests/crawlers/test_topic_crawler.py
@@ -87,26 +87,32 @@ class TestTopicCrawlerRun:
     def _patch_run(self):
         return (
             patch("crawlers.topic_crawler._discover_topic_repos"),
+            patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}),
             patch("crawlers.topic_crawler.fetch_repo_metadata_cached"),
-            patch("crawlers.topic_crawler.find_skill_md_paths"),
+            patch("crawlers.topic_crawler.find_skill_md_paths_cached"),
             patch("crawlers.topic_crawler.fetch_skill_md_cached"),
             patch("crawlers.topic_crawler.load_meta_cache", return_value={}),
             patch("crawlers.topic_crawler.save_meta_cache"),
             patch("crawlers.topic_crawler.load_content_cache", return_value={}),
             patch("crawlers.topic_crawler.save_content_cache"),
+            patch("crawlers.topic_crawler.load_tree_cache", return_value={}),
+            patch("crawlers.topic_crawler.save_tree_cache"),
         )
 
     def test_writes_records_for_discovered_repos(self, tmp_path):
         from crawlers.topic_crawler import run
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/skill-a", "user/skill-b"]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -121,13 +127,16 @@ def test_output_has_required_fields(self, tmp_path):
         from crawlers.topic_crawler import run
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/skill-a"]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -144,13 +153,16 @@ def test_source_tag_is_topic(self, tmp_path):
         from crawlers.topic_crawler import run
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/skill-a"]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -166,13 +178,16 @@ def test_respects_limit(self, tmp_path):
         from crawlers.topic_crawler import run
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = [f"user/skill-{i}" for i in range(10)]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -187,13 +202,16 @@ def test_skips_repos_with_no_skill_md(self, tmp_path):
         from crawlers.topic_crawler import run
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/no-skill-md", "user/has-skill-md"]
             mock_meta.return_value = _mock_meta()
             mock_paths.side_effect = [{}, {"SKILL.md": "sha1"}]
@@ -216,13 +234,16 @@ def test_skips_already_covered_repos(self, tmp_path):
         )
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/already-covered", "user/new-skill"]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -249,13 +270,16 @@ def test_resume_skips_existing_keys(self, tmp_path):
         )
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/skill-a", "user/skill-b"]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -269,13 +293,16 @@ def test_name_falls_back_to_repo_name_when_no_frontmatter(self, tmp_path):
         from crawlers.topic_crawler import run
 
         with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch", return_value={}), \
              patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
-             patch("crawlers.topic_crawler.find_skill_md_paths") as mock_paths, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
              patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
              patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
              patch("crawlers.topic_crawler.save_meta_cache"), \
              patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
-             patch("crawlers.topic_crawler.save_content_cache"):
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
             mock_disc.return_value = ["user/my-cool-skill"]
             mock_meta.return_value = _mock_meta()
             mock_paths.return_value = {"SKILL.md": "sha1"}
@@ -288,13 +315,15 @@ def test_name_falls_back_to_repo_name_when_no_frontmatter(self, tmp_path):
         assert record["name"] == "my-cool-skill"
 
     def test_topic_crawl_uses_caches(self, tmp_path, monkeypatch):
-        """Crawl loads both caches at start and saves both caches at end."""
+        """Crawl loads meta, content, and tree caches at start and saves all three at end."""
         import crawlers.topic_crawler as tc
 
-        calls = {"saved_meta": 0, "saved_content": 0}
+        calls = {"saved_meta": 0, "saved_content": 0, "saved_tree": 0}
 
+        monkeypatch.setattr(tc, "fetch_repo_metadata_batch", lambda s, names: {})
         monkeypatch.setattr(tc, "load_meta_cache", lambda p: {})
         monkeypatch.setattr(tc, "load_content_cache", lambda p: {})
+        monkeypatch.setattr(tc, "load_tree_cache", lambda p: {})
         monkeypatch.setattr(
             tc, "save_meta_cache",
             lambda c, p: calls.__setitem__("saved_meta", calls["saved_meta"] + 1),
@@ -303,17 +332,24 @@ def test_topic_crawl_uses_caches(self, tmp_path, monkeypatch):
             tc, "save_content_cache",
             lambda c, p: calls.__setitem__("saved_content", calls["saved_content"] + 1),
         )
+        monkeypatch.setattr(
+            tc, "save_tree_cache",
+            lambda c, p: calls.__setitem__("saved_tree", calls["saved_tree"] + 1),
+        )
         monkeypatch.setattr(
             tc, "fetch_repo_metadata_cached",
             lambda s, r, c: {
                 "stargazers_count": 50,
                 "default_branch": "main",
-                "pushed_at": "",
+                "pushed_at": "2026-01-01T00:00:00Z",
                 "topics": [],
                 "description": "",
             },
         )
-        monkeypatch.setattr(tc, "find_skill_md_paths", lambda s, r: {"SKILL.md": "sha1"})
+        monkeypatch.setattr(
+            tc, "find_skill_md_paths_cached",
+            lambda s, r, p, db, c: {"SKILL.md": "sha1"},
+        )
         monkeypatch.setattr(tc, "fetch_skill_md_cached", lambda *a, **k: "---\nname: t\n---")
         monkeypatch.setattr(tc, "_discover_topic_repos", lambda s, limit=1000: ["user/skill-a"])
 
@@ -323,3 +359,153 @@ def test_topic_crawl_uses_caches(self, tmp_path, monkeypatch):
         assert count == 1
         assert calls["saved_meta"] == 1
         assert calls["saved_content"] == 1
+        assert calls["saved_tree"] == 1
+
+
+
+# ---------------------------------------------------------------------------
+# TestTopicCrawlerBatchMetaIntegration
+# ---------------------------------------------------------------------------
+
+class TestTopicCrawlerBatchMetaIntegration:
+    """Verify that batch GraphQL metadata is used (and REST is skipped) when available."""
+
+    def test_batch_result_used_and_rest_not_called(self, tmp_path):
+        """When fetch_repo_metadata_batch returns metadata for a repo, the per-repo
+        fetch_repo_metadata_cached should NOT be called for that repo."""
+        from crawlers.topic_crawler import run
+
+        batch_meta = {
+            "user/skill-a": {
+                "stargazers_count": 99,
+                "pushed_at": "2026-05-01T00:00:00Z",
+                "description": "from graphql",
+                "default_branch": "main",
+                "topics": ["graphql"],
+            }
+        }
+
+        with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch",
+                   return_value=batch_meta), \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_rest_meta, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
+             patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
+             patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_meta_cache"), \
+             patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
+
+            mock_disc.return_value = ["user/skill-a"]
+            mock_paths.return_value = {"SKILL.md": "sha1"}
+            mock_skill_md.return_value = None
+
+            out = str(tmp_path / "out.jsonl")
+            count = run(out)
+
+        assert count == 1
+        # The batch provided metadata — per-repo REST fallback must NOT have been called
+        mock_rest_meta.assert_not_called()
+
+        # Confirm the record used the batch-sourced star count
+        import json as _json
+        record = _json.loads(Path(out).read_text().strip())
+        assert record["raw_metadata"]["stars"] == 99
+
+
+# ---------------------------------------------------------------------------
+# P2 Fix tests: batch pre-filtering and clean-sweep watermark
+# ---------------------------------------------------------------------------
+
+class TestBatchPreFilter:
+    """P2 #1: batch should only fetch metadata for repos that survive skip filters."""
+
+    def test_batch_excludes_already_covered_and_filtered(self, tmp_path):
+        """fetch_repo_metadata_batch should only be called with repos not in
+        already_covered or filter_cache."""
+        from crawlers.topic_crawler import run
+
+        with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch") as mock_batch, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
+             patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
+             patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_meta_cache"), \
+             patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"), \
+             patch("crawlers.topic_crawler._load_existing_repo_urls",
+                   return_value={"https://github.com/u/a"}), \
+             patch("crawlers.topic_crawler.load_filter_cache",
+                   return_value={"https://github.com/u/b"}):
+
+            mock_disc.return_value = ["u/a", "u/b", "u/c"]
+            mock_batch.return_value = {}
+            mock_meta.return_value = _mock_meta()
+            mock_paths.return_value = {}
+            mock_skill_md.return_value = None
+
+            out = str(tmp_path / "out.jsonl")
+            run(out, filter_cache_path="fake_filter_cache.json",
+                existing_raw_dirs=["fake_raw_dir"])
+
+        # batch must be called exactly once
+        mock_batch.assert_called_once()
+        called_names = mock_batch.call_args.args[1]
+        assert "u/c" in called_names, f"Expected u/c in batch call, got {called_names}"
+        assert "u/a" not in called_names, "u/a (already_covered) must be excluded"
+        assert "u/b" not in called_names, "u/b (filter_cache) must be excluded"
+
+
+
+# ---------------------------------------------------------------------------
+# P2 fix: lazy batch-fetch respects --limit
+# ---------------------------------------------------------------------------
+
+class TestLazyBatchRespectsLimit:
+    """P2: GraphQL batch should not eagerly fetch all repos when limit is small."""
+
+    def test_lazy_batch_respects_limit(self, tmp_path):
+        """With limit=1 and 250 repos, fetch_repo_metadata_batch should be called
+        AT MOST twice (one chunk of 100 at most, possibly two if boundary is hit),
+        not 3 chunks for all 250 repos."""
+        from crawlers.topic_crawler import run
+
+        repos_250 = [f"u/r{i}" for i in range(250)]
+
+        with patch("crawlers.topic_crawler._discover_topic_repos") as mock_disc, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_batch") as mock_batch, \
+             patch("crawlers.topic_crawler.fetch_repo_metadata_cached") as mock_meta, \
+             patch("crawlers.topic_crawler.find_skill_md_paths_cached") as mock_paths, \
+             patch("crawlers.topic_crawler.fetch_skill_md_cached") as mock_skill_md, \
+             patch("crawlers.topic_crawler.load_meta_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_meta_cache"), \
+             patch("crawlers.topic_crawler.load_content_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_content_cache"), \
+             patch("crawlers.topic_crawler.load_tree_cache", return_value={}), \
+             patch("crawlers.topic_crawler.save_tree_cache"):
+
+            mock_disc.return_value = repos_250
+            mock_batch.return_value = {}  # empty -> REST fallback fills meta
+            mock_meta.return_value = _mock_meta()
+            mock_paths.return_value = {"SKILL.md": "sha1"}
+            mock_skill_md.return_value = SAMPLE_SKILL_MD
+
+            out = str(tmp_path / "out.jsonl")
+            run(out, limit=1)
+
+        # With limit=1, only the first chunk (repos 0-99) should be fetched,
+        # so call count must be at most 2 (NOT 3 for all 250 repos).
+        assert mock_batch.call_count <= 2, (
+            f"Expected at most 2 batch calls with limit=1, got {mock_batch.call_count}"
+        )
+        # Each individual call must be a chunk of at most 100 repos (not all 250 at once).
+        for i, c in enumerate(mock_batch.call_args_list):
+            chunk = c.args[1]  # second positional arg is the list of repo names
+            assert len(chunk) <= 100, (
+                f"Batch call {i} sent {len(chunk)} repos — expected ≤100 (lazy chunking)"
+            )