perf(monitor): one request per skill so the full baseline fits in CI

docs/crawl.md

@@ -101,6 +101,8 @@ Sweep the registry, fast-scan skills (rule engine + threat intel, escalating fla
 
 **Incremental by default.** The monitor only re-fetches and re-scans skills whose `version`/`updated_at` changed since the last snapshot; unchanged skills are carried forward untouched. The **first run scans everything**; every run after only pays for what actually changed — so a daily sweep is fast and cheap. Because incremental detection trusts the registry's version metadata, run a periodic **`--full`** sweep (e.g. weekly) to also catch *silent same-version content swaps* — trojanized updates that keep the same version number. Each snapshot records `scanned_count` vs `reused_count` so you can see how much a run actually did.
 
+**One request per skill.** Skill metadata (version, updated_at, display name, install count) is taken from the enumeration listing, so scanning a skill costs a single request — the `SKILL.md` file — rather than a separate detail lookup. This roughly halves crawl time and keeps a full sweep within the ClawHub rate limit (120 req/min). The trade-off: per-skill publisher and moderation flags aren't captured by the sweep (the scan verdict, version, and installs are).
+
 ```bash
 malwar crawl monitor                     # incremental sweep -> snapshot -> diff
 malwar crawl monitor --full              # re-scan every skill (catches same-version tampering)

src/malwar/monitor/snapshot.py

@@ -25,6 +25,7 @@
 import logging
 from collections.abc import Callable
 from pathlib import Path
+from typing import NamedTuple
 
 from malwar.crawl.client import ClawHubClient
 from malwar.monitor.models import RegistrySnapshot, SkillRecord
@@ -39,26 +40,37 @@
 
 ProgressCallback = Callable[[int, int, str], None]
 
-# A lightweight (slug, version, updated_at) reference gathered from the cheap
-# list/search endpoints — enough to decide whether a skill needs re-scanning
-# without fetching its file.
-SkillRef = tuple[str, str | None, int | None]
+
+class SkillMeta(NamedTuple):
+    """Per-skill metadata harvested from the cheap list/search endpoints.
+
+    Carrying this through means the sweep never needs a separate per-skill
+    detail request: it decides whether a skill changed from ``version`` /
+    ``updated_at``, and populates the record's display fields directly — so a
+    scanned skill costs a single request (the file) instead of two.
+    """
+
+    slug: str
+    version: str | None
+    updated_at: int | None
+    display_name: str = ""
+    installs: int = 0
 
 
 async def _enumerate_skills(
     client: ClawHubClient,
     *,
     max_skills: int | None,
     page_size: int,
-) -> list[SkillRef]:
-    """Return ``(slug, version, updated_at)`` for every skill in the registry.
+) -> list[SkillMeta]:
+    """Return :class:`SkillMeta` for every skill in the registry.
 
     Pages through the list endpoint; if that yields nothing (the endpoint is
     known to return empty at times), falls back to fanning search queries out
-    across seed terms and de-duplicating. The version/updated_at metadata comes
-    free with the listing, so change detection needs no per-skill fetch.
+    across seed terms and de-duplicating. The metadata comes free with the
+    listing, so change detection and record population need no per-skill fetch.
     """
-    refs: list[SkillRef] = []
+    metas: list[SkillMeta] = []
     seen: set[str] = set()
 
     cursor: str | None = None
@@ -72,16 +84,24 @@ async def _enumerate_skills(
             if item.slug not in seen:
                 seen.add(item.slug)
                 version = item.latest_version.version if item.latest_version else None
-                refs.append((item.slug, version, item.updated_at))
-        if max_skills is not None and len(refs) >= max_skills:
-            return refs[:max_skills]
+                metas.append(
+                    SkillMeta(
+                        slug=item.slug,
+                        version=version,
+                        updated_at=item.updated_at,
+                        display_name=item.display_name,
+                        installs=item.stats.installs_all_time,
+                    )
+                )
+        if max_skills is not None and len(metas) >= max_skills:
+            return metas[:max_skills]
         if not cursor:
             break
 
-    if refs:
-        return refs
+    if metas:
+        return metas
 
-    # Fallback: broad search fan-out.
+    # Fallback: broad search fan-out (search results carry no install stats).
     for term in _SEED_TERMS:
         try:
             results = await client.search(term, limit=page_size)
@@ -91,11 +111,18 @@ async def _enumerate_skills(
         for r in results:
             if r.slug not in seen:
                 seen.add(r.slug)
-                refs.append((r.slug, r.version, r.updated_at))
-        if max_skills is not None and len(refs) >= max_skills:
-            return refs[:max_skills]
+                metas.append(
+                    SkillMeta(
+                        slug=r.slug,
+                        version=r.version,
+                        updated_at=r.updated_at,
+                        display_name=r.display_name,
+                    )
+                )
+        if max_skills is not None and len(metas) >= max_skills:
+            return metas[:max_skills]
 
-    return refs
+    return metas
 
 
 def _is_unchanged(prev: SkillRecord | None, version: str | None, updated_at: int | None) -> bool:
@@ -110,33 +137,29 @@ def _is_unchanged(prev: SkillRecord | None, version: str | None, updated_at: int
     return prev.version == version and prev.updated_at == updated_at
 
 
-async def _scan_slug(client: ClawHubClient, slug: str, *, escalate: bool) -> SkillRecord:
-    """Fetch and scan a single skill, escalating to the LLM only if flagged."""
-    try:
-        detail = await client.get_skill(slug)
-    except Exception as exc:  # ClawHubError, httpx timeouts, etc.
-        return SkillRecord(slug=slug, verdict="UNKNOWN", error=f"detail: {exc}")
+async def _scan_slug(client: ClawHubClient, meta: SkillMeta, *, escalate: bool) -> SkillRecord:
+    """Fetch and scan a single skill, escalating to the LLM only if flagged.
 
+    Metadata (version, updated_at, display name, installs) comes from the
+    enumeration listing, so this makes just one request per skill — the file.
+    """
     record = SkillRecord(
-        slug=slug,
-        display_name=detail.display_name,
-        publisher=detail.owner.username if detail.owner else "",
-        version=detail.latest_version.version if detail.latest_version else None,
-        updated_at=detail.updated_at,
-        installs=detail.stats.installs_all_time,
-        moderation_blocked=bool(detail.moderation and detail.moderation.is_malware_blocked),
-        moderation_suspicious=bool(detail.moderation and detail.moderation.is_suspicious),
+        slug=meta.slug,
+        display_name=meta.display_name,
+        version=meta.version,
+        updated_at=meta.updated_at,
+        installs=meta.installs,
     )
 
     try:
-        content = await client.get_skill_file(slug)
+        content = await client.get_skill_file(meta.slug)
     except Exception as exc:  # ClawHubError, httpx timeouts, etc.
         record.error = f"file: {exc}"
         return record
 
     record.content_sha256 = hashlib.sha256(content.encode("utf-8")).hexdigest()
 
-    file_name = f"{slug}/SKILL.md"
+    file_name = f"{meta.slug}/SKILL.md"
     try:
         result = await scan(content, file_name=file_name, use_llm=False, use_urls=False)
         if escalate and result.verdict != "CLEAN":
@@ -186,23 +209,23 @@ async def build_snapshot(
         Optional callback ``(done, total, slug)`` for progress reporting.
     """
     client = client or ClawHubClient()
-    refs = await _enumerate_skills(client, max_skills=max_skills, page_size=page_size)
+    metas = await _enumerate_skills(client, max_skills=max_skills, page_size=page_size)
 
     snapshot = RegistrySnapshot(registry=client.base_url)
-    total = len(refs)
+    total = len(metas)
     if total == 0:
         return snapshot
 
     prev_skills = previous.skills if (previous is not None and not force_rescan) else {}
 
     # Split the registry into "carry forward unchanged" and "must re-scan".
-    to_scan: list[str] = []
-    for slug, version, updated_at in refs:
-        prev = prev_skills.get(slug)
-        if _is_unchanged(prev, version, updated_at):
-            snapshot.skills[slug] = prev.model_copy(deep=True)  # type: ignore[union-attr]
+    to_scan: list[SkillMeta] = []
+    for meta in metas:
+        prev = prev_skills.get(meta.slug)
+        if _is_unchanged(prev, meta.version, meta.updated_at):
+            snapshot.skills[meta.slug] = prev.model_copy(deep=True)  # type: ignore[union-attr]
         else:
-            to_scan.append(slug)
+            to_scan.append(meta)
 
     snapshot.reused_count = len(snapshot.skills)
     snapshot.scanned_count = len(to_scan)
@@ -217,22 +240,22 @@ async def build_snapshot(
     done = len(snapshot.skills)
     lock = asyncio.Lock()
 
-    async def _worker(slug: str) -> None:
+    async def _worker(meta: SkillMeta) -> None:
         nonlocal done
         async with semaphore:
             try:
-                record = await _scan_slug(client, slug, escalate=escalate)
+                record = await _scan_slug(client, meta, escalate=escalate)
             except Exception as exc:  # last-resort guard; never abort the sweep
-                record = SkillRecord(slug=slug, verdict="UNKNOWN", error=f"worker: {exc}")
+                record = SkillRecord(slug=meta.slug, verdict="UNKNOWN", error=f"worker: {exc}")
         async with lock:
-            snapshot.skills[slug] = record
+            snapshot.skills[meta.slug] = record
             if record.error:
-                snapshot.errors[slug] = record.error
+                snapshot.errors[meta.slug] = record.error
             done += 1
             if on_progress is not None:
-                on_progress(done, total, slug)
+                on_progress(done, total, meta.slug)
 
-    await asyncio.gather(*(_worker(slug) for slug in to_scan))
+    await asyncio.gather(*(_worker(meta) for meta in to_scan))
     return snapshot
 
 

tests/unit/test_monitor.py

@@ -53,6 +53,9 @@ def __init__(
         self._updated_ats = updated_ats or dict.fromkeys(skills, 1000)
         # Slugs whose file was actually fetched — lets tests prove reuse.
         self.file_fetches: list[str] = []
+        # Slugs whose per-skill detail endpoint was hit — should stay empty,
+        # since the sweep now derives metadata from the listing.
+        self.detail_fetches: list[str] = []
 
     async def list_skills(self, limit: int = 20, cursor: str | None = None):
         items = [
@@ -61,6 +64,7 @@ async def list_skills(self, limit: int = 20, cursor: str | None = None):
                 displayName=slug.replace("-", " ").title(),
                 latestVersion=VersionInfo(version=self._versions[slug]),
                 updatedAt=self._updated_ats.get(slug),
+                stats=SkillStats(installsAllTime=1234),
             )
             for slug in self._skills
         ]
@@ -70,6 +74,7 @@ async def search(self, query: str, limit: int = 20) -> list[SearchResult]:
         return []
 
     async def get_skill(self, slug: str) -> SkillDetail:
+        self.detail_fetches.append(slug)
         if slug not in self._skills:
             raise ClawHubError(f"not found: {slug}", status_code=404)
         return SkillDetail(
@@ -112,6 +117,16 @@ async def test_records_content_hash_and_version(self):
         assert rec.version == "1.0.0"
         assert rec.installs == 1234
 
+    async def test_one_request_per_skill_no_detail_fetch(self):
+        # Metadata comes from the listing, so a scanned skill costs a single
+        # request (the file) — the per-skill detail endpoint is never hit.
+        client = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY})
+        snap = await build_snapshot(client, escalate=False)
+        assert snap.skill_count == 2
+        assert client.detail_fetches == []
+        assert sorted(client.file_fetches) == ["a", "b"]
+        assert snap.skills["a"].display_name == "A"
+
     async def test_max_skills_cap(self):
         client = FakeClawHubClient({f"s{i}": BENIGN_BODY for i in range(10)})
         snap = await build_snapshot(client, escalate=False, max_skills=3)