fix(monitor): resilient registry enumeration — stop silent truncation of the 66k baseline

src/malwar/monitor/models.py

@@ -85,6 +85,14 @@ class RegistrySnapshot(BaseModel):
     # Fragile single-rule MALICIOUS verdicts downgraded to SUSPICIOUS this run
     # because no authoritative second opinion confirmed them.
     downgraded_count: int = 0
+    # Registry enumeration health. ``enumerated_count`` is how many skills the
+    # listing returned this run; ``enumeration_complete`` is False when paging
+    # aborted early (e.g. a rate-limit/5xx mid-listing). On an incomplete
+    # enumeration, previously-known skills are carried forward instead of being
+    # dropped, so a transient listing failure can't silently shrink the baseline.
+    enumerated_count: int = 0
+    enumeration_complete: bool = True
+    carried_forward_count: int = 0
 
     @property
     def skill_count(self) -> int:

src/malwar/monitor/snapshot.py

@@ -69,29 +69,63 @@ class SkillMeta(NamedTuple):
     installs: int = 0
 
 
+# How many times to retry a single failed listing page before giving up on the
+# enumeration. The registry rate-limits, so a long (~66k skill) sweep will hit a
+# transient 429/5xx mid-listing; retrying the page keeps one hiccup from
+# truncating the whole enumeration.
+_PAGE_RETRIES = 4
+# Backoff (seconds) between page retries, indexed by attempt. Capped and finite.
+_PAGE_BACKOFF = (2, 4, 8, 16)
+
+
 async def _enumerate_skills(
     client: ClawHubClient,
     *,
     max_skills: int | None,
     page_size: int,
-) -> 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 metadata comes free with the
-    listing, so change detection and record population need no per-skill fetch.
+) -> tuple[list[SkillMeta], bool]:
+    """Return ``(skills, complete)`` for the registry listing.
+
+    Pages through the list endpoint, retrying each page across transient
+    failures (the registry rate-limits, so a long sweep will hit a 429/5xx
+    mid-listing). ``complete`` is True only when paging reached the natural end
+    with every page succeeding; it is False when a page still failed after
+    retries, so the caller knows the list is partial and must not treat absent
+    skills as removed. If the listing yields nothing at all, falls back to
+    fanning search queries across seed terms (best-effort, always ``complete``
+    False since it can't guarantee coverage).
     """
     metas: list[SkillMeta] = []
     seen: set[str] = set()
 
     cursor: str | None = None
     while True:
-        try:
-            items, cursor = await client.list_skills(limit=page_size, cursor=cursor)
-        except Exception as exc:  # ClawHubError, httpx timeouts after retries, etc.
-            logger.warning("list_skills failed: %s", exc)
-            break
+        items = None
+        for attempt in range(_PAGE_RETRIES):
+            try:
+                items, cursor = await client.list_skills(limit=page_size, cursor=cursor)
+                break
+            except Exception as exc:  # ClawHubError, httpx timeouts after retries, etc.
+                logger.warning(
+                    "list_skills page failed (attempt %d/%d): %s",
+                    attempt + 1,
+                    _PAGE_RETRIES,
+                    exc,
+                )
+                if attempt + 1 < _PAGE_RETRIES:
+                    await asyncio.sleep(_PAGE_BACKOFF[min(attempt, len(_PAGE_BACKOFF) - 1)])
+        if items is None:
+            # Page still failing after retries: stop, but report the list as
+            # PARTIAL so the caller preserves the baseline instead of dropping
+            # everything we couldn't re-list this run.
+            logger.error(
+                "enumeration aborted after %d failed attempts on one page; "
+                "returning a PARTIAL list of %d skills (baseline will be "
+                "carried forward)",
+                _PAGE_RETRIES,
+                len(metas),
+            )
+            return metas, False
         for item in items:
             if item.slug not in seen:
                 seen.add(item.slug)
@@ -106,14 +140,22 @@ async def _enumerate_skills(
                     )
                 )
         if max_skills is not None and len(metas) >= max_skills:
-            return metas[:max_skills]
+            return metas[:max_skills], True
         if not cursor:
-            break
+            return metas, True  # reached the natural end with every page ok
+
+    # (unreachable: every branch above returns)
 
-    if metas:
-        return metas
 
-    # Fallback: broad search fan-out (search results carry no install stats).
+async def _enumerate_via_search(
+    client: ClawHubClient,
+    *,
+    max_skills: int | None,
+    page_size: int,
+    seen: set[str],
+) -> list[SkillMeta]:
+    """Best-effort fallback: fan search queries across seed terms, de-duplicated."""
+    metas: list[SkillMeta] = []
     for term in _SEED_TERMS:
         try:
             results = await client.search(term, limit=page_size)
@@ -133,7 +175,6 @@ async def _enumerate_skills(
                 )
         if max_skills is not None and len(metas) >= max_skills:
             return metas[:max_skills]
-
     return metas
 
 
@@ -259,11 +300,43 @@ async def build_snapshot(
         Optional callback ``(done, total, slug)`` for progress reporting.
     """
     client = client or ClawHubClient()
-    metas = await _enumerate_skills(client, max_skills=max_skills, page_size=page_size)
+    metas, complete = await _enumerate_skills(client, max_skills=max_skills, page_size=page_size)
+    if not metas:
+        # Listing returned nothing at all — try the search fan-out (best-effort,
+        # cannot guarantee coverage, so the result stays "incomplete").
+        metas = await _enumerate_via_search(
+            client, max_skills=max_skills, page_size=page_size, seen=set()
+        )
+        complete = False
 
     snapshot = RegistrySnapshot(registry=client.base_url)
+    snapshot.enumerated_count = len(metas)
+    snapshot.enumeration_complete = complete
+
+    # Carry-forward guard: when the listing is incomplete, a previously-known
+    # skill that simply wasn't re-listed this run must NOT be treated as removed.
+    # Preserve its last record so one transient listing failure can't shrink the
+    # baseline (this is the bug that collapsed 66k -> a few hundred after 07-06).
+    enumerated_slugs = {m.slug for m in metas}
+    if previous is not None and not complete:
+        carried = 0
+        for slug, rec in previous.skills.items():
+            if slug not in enumerated_slugs:
+                snapshot.skills[slug] = rec.model_copy(deep=True)
+                carried += 1
+        snapshot.carried_forward_count = carried
+        if carried:
+            logger.warning(
+                "enumeration INCOMPLETE (%d listed vs %d previously known): "
+                "carried forward %d skills to protect the baseline",
+                len(metas),
+                len(previous.skills),
+                carried,
+            )
+
     total = len(metas)
     if total == 0:
+        # Nothing listed; any carried-forward baseline above still stands.
         return snapshot
 
     prev_skills = previous.skills if (previous is not None and not force_rescan) else {}
@@ -288,11 +361,14 @@ async def build_snapshot(
     snapshot.scanned_count = len(to_scan)
     snapshot.pending_count = len(pending)
     logger.info(
-        "sweep: %d skills total, %d to scan, %d reused unchanged, %d deferred (budget)",
+        "sweep: %d listed (complete=%s), %d to scan, %d reused unchanged, "
+        "%d deferred (budget), %d carried forward",
         total,
+        complete,
         snapshot.scanned_count,
         snapshot.reused_count,
         snapshot.pending_count,
+        snapshot.carried_forward_count,
     )
 
     backend = escalation or NoneBackend()
@@ -369,13 +445,13 @@ async def _escalate(slug: str) -> None:
     # when escalation is disabled, so a rules-only sweep never over-convicts.
     downgraded = 0
     for meta in to_scan:
-        rec = snapshot.skills.get(meta.slug)
-        if rec is None or not is_fragile_malicious(rec):
+        record = snapshot.skills.get(meta.slug)
+        if record is None or not is_fragile_malicious(record):
             continue
-        confirmed = rec.llm_escalated and rec.escalation_verdict == "MALICIOUS"
+        confirmed = record.llm_escalated and record.escalation_verdict == "MALICIOUS"
         if not confirmed:
-            rec.verdict = "SUSPICIOUS"
-            rec.risk_score = min(rec.risk_score, _FRAGILE_DOWNGRADE_RISK)
+            record.verdict = "SUSPICIOUS"
+            record.risk_score = min(record.risk_score, _FRAGILE_DOWNGRADE_RISK)
             downgraded += 1
     if downgraded:
         logger.info(

tests/unit/test_monitor.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import pytest
+
 from malwar.crawl.client import ClawHubError
 from malwar.crawl.models import (
     ModerationInfo,
@@ -25,6 +27,12 @@
 )
 from malwar.monitor.report import TWEET_LIMIT
 
+
+@pytest.fixture(autouse=True)
+def _instant_enumeration_backoff(monkeypatch):
+    """Zero the page-retry backoff so enumeration-failure tests don't sleep."""
+    monkeypatch.setattr("malwar.monitor.snapshot._PAGE_BACKOFF", (0, 0, 0, 0))
+
 # A skill body that trips MALWAR-FRAUD-001 (agentic affiliate injection).
 MALICIOUS_BODY = (
     "# Money Radar\n"
@@ -170,6 +178,7 @@ async def _boom(limit=20, cursor=None):
         # search fallback returns [] in the fake → empty snapshot, no exception.
         snap = await build_snapshot(client)
         assert snap.skill_count == 0
+        assert snap.enumeration_complete is False
 
 
 class TestIncrementalSweep:
@@ -634,3 +643,112 @@ async def test_authoritative_clear_becomes_clean(self):
         rec = snap.skills["evil"]
         assert rec.verdict == "CLEAN"  # authoritatively cleared
         assert snap.downgraded_count == 0
+
+
+# ---------------------------------------------------------------------------
+# Enumeration resilience: retry pages, mark partial, carry forward baseline
+# ---------------------------------------------------------------------------
+
+class PaginatedClient(FakeClawHubClient):
+    """Serves skills across pages; can fail a chosen page to test retry/partial.
+
+    ``fail_page`` (0-indexed) fails its first ``fail_times`` attempts; set
+    ``fail_times`` high to fail permanently.
+    """
+
+    def __init__(self, skills, *, page_size=2, fail_page=None, fail_times=99, **kw):
+        super().__init__(skills, **kw)
+        self._order = list(skills)
+        self._psize = page_size
+        self._fail_page = fail_page
+        self._fail_times = fail_times
+        self._attempts: dict[int, int] = {}
+
+    async def list_skills(self, limit: int = 20, cursor: str | None = None):
+        page = int(cursor) if cursor else 0
+        if page == self._fail_page:
+            n = self._attempts.get(page, 0)
+            if n < self._fail_times:
+                self._attempts[page] = n + 1
+                raise ClawHubError("page boom", status_code=503)
+        start = page * self._psize
+        chunk = self._order[start:start + self._psize]
+        items = [
+            SkillSummary(
+                slug=s,
+                displayName=s,
+                latestVersion=VersionInfo(version=self._versions[s]),
+                updatedAt=self._updated_ats.get(s),
+                stats=SkillStats(installsAllTime=1),
+            )
+            for s in chunk
+        ]
+        nxt = str(page + 1) if start + self._psize < len(self._order) else None
+        return items, nxt
+
+
+def _prev_snapshot(slugs):
+    snap = RegistrySnapshot(registry="https://clawhub.test")
+    for s in slugs:
+        snap.skills[s] = SkillRecord(slug=s, verdict="CLEAN", version="1.0.0", updated_at=1000)
+    return snap
+
+
+class TestEnumerationResilience:
+    async def test_transient_page_failure_is_retried_then_completes(self):
+        # Page 1 fails twice, succeeds on the third attempt (< _PAGE_RETRIES).
+        client = PaginatedClient(
+            {"a": BENIGN_BODY, "b": BENIGN_BODY, "c": BENIGN_BODY, "d": BENIGN_BODY},
+            page_size=2, fail_page=1, fail_times=2,
+        )
+        snap = await build_snapshot(client)
+        assert snap.enumeration_complete is True
+        assert snap.enumerated_count == 4
+        assert set(snap.skills) == {"a", "b", "c", "d"}
+
+    async def test_persistent_page_failure_marks_incomplete(self):
+        client = PaginatedClient(
+            {"a": BENIGN_BODY, "b": BENIGN_BODY, "z": BENIGN_BODY, "y": BENIGN_BODY},
+            page_size=2, fail_page=1, fail_times=99,
+        )
+        snap = await build_snapshot(client)
+        assert snap.enumeration_complete is False
+        # Only page 0 (a, b) was listed before the permanent failure.
+        assert snap.enumerated_count == 2
+
+    async def test_incomplete_enumeration_carries_forward_baseline(self):
+        # Previous run knew a, b, c. This run only lists a (page 1 dies), so b and
+        # c must be carried forward, not dropped.
+        prev = _prev_snapshot(["a", "b", "c"])
+        client = PaginatedClient(
+            {"a": BENIGN_BODY, "z": BENIGN_BODY},
+            page_size=1, fail_page=1, fail_times=99,
+        )
+        snap = await build_snapshot(client, previous=prev)
+        assert snap.enumeration_complete is False
+        assert snap.carried_forward_count == 2
+        assert {"a", "b", "c"}.issubset(set(snap.skills))  # baseline preserved
+
+    async def test_complete_enumeration_allows_genuine_removal(self):
+        # When enumeration is complete, a skill absent from the listing is
+        # genuinely gone and must NOT be carried forward.
+        prev = _prev_snapshot(["a", "b", "gone"])
+        client = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY})  # single full page
+        snap = await build_snapshot(client, previous=prev)
+        assert snap.enumeration_complete is True
+        assert snap.carried_forward_count == 0
+        assert "gone" not in snap.skills
+
+    async def test_empty_listing_with_previous_preserves_baseline(self):
+        prev = _prev_snapshot(["a", "b", "c"])
+
+        async def _empty(limit=20, cursor=None):
+            return [], None
+
+        client = FakeClawHubClient({"a": BENIGN_BODY})
+        client.list_skills = _empty  # type: ignore[assignment]
+        snap = await build_snapshot(client, previous=prev)
+        # Search fallback also yields nothing in the fake -> incomplete, but the
+        # previous baseline is carried forward rather than wiped.
+        assert snap.enumeration_complete is False
+        assert {"a", "b", "c"}.issubset(set(snap.skills))