From 3f2d43f504febd27b26b46e92b3dbbd6886c58cd Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 04:08:30 +0000 Subject: [PATCH] perf(monitor): one request per skill (drop redundant detail fetch) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The full registry sweep made two requests per skill (metadata + file), so at ClawHub's 120 req/min rate limit a full baseline of a few thousand skills could not finish inside a 60-min CI window — and since the snapshot only commits after scanning everything, timed-out runs saved nothing. Harvest per-skill metadata (version, updated_at, display name, installs) from the enumeration listing instead and drop the per-skill get_skill() detail call, so a scanned skill costs a single request (the file). This roughly halves crawl time and keeps a full sweep within the rate budget. Trade-off: publisher and moderation flags are no longer captured by the sweep (verdict, version, and installs are); nothing in the diff/report depended on them. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01DNoTXU8k3pfSBzR7aJubqL --- docs/crawl.md | 2 + src/malwar/monitor/snapshot.py | 123 +++++++++++++++++++-------------- tests/unit/test_monitor.py | 15 ++++ 3 files changed, 90 insertions(+), 50 deletions(-) diff --git a/docs/crawl.md b/docs/crawl.md index 4d2c004..d80b906 100644 --- a/docs/crawl.md +++ b/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) diff --git a/src/malwar/monitor/snapshot.py b/src/malwar/monitor/snapshot.py index f87e156..896f858 100644 --- a/src/malwar/monitor/snapshot.py +++ b/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,10 +40,21 @@ 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( @@ -50,15 +62,15 @@ async def _enumerate_skills( *, 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 diff --git a/tests/unit/test_monitor.py b/tests/unit/test_monitor.py index ee282e4..52c6fe9 100644 --- a/tests/unit/test_monitor.py +++ b/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)