feat(monitor): budgeted, resumable baseline (--max-scans)

.github/workflows/registry-monitor.yml

@@ -97,10 +97,15 @@ jobs:
           else
             echo "Mode: incremental (only changed skills)."
           fi
+          # ClawHub is rate-limited (~2 req/s), so a full ~6k-skill registry
+          # can't be swept in one CI run. Cap scans per run; the overflow is
+          # deferred and picked up on subsequent runs, so the baseline builds
+          # up over a few days and then daily runs only touch what changed.
           malwar crawl monitor \
             --snapshot-dir data/registry-snapshots \
             --format json \
             --output data/registry-snapshots/latest-diff.json \
+            --max-scans 1500 \
             $FULL
 
       - name: Publish snapshot to data branch

docs/crawl.md

@@ -101,7 +101,9 @@ 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).
+**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. The trade-off: per-skill publisher and moderation flags aren't captured by the sweep (the scan verdict, version, and installs are).
+
+**Budgeted, resumable baseline.** ClawHub is rate-limited (~120 req/min), so a full ~6k-skill registry can't be swept in one shot. `--max-scans N` caps how many skills are actually scanned per run; any overflow is recorded as an `UNKNOWN` placeholder and picked up on the next run. So the **first full baseline builds up over several runs** rather than dying to a timeout, and once complete, daily incremental runs only touch what changed. `scanned_count`/`reused_count`/`pending_count` on each snapshot show the split.
 
 ```bash
 malwar crawl monitor                     # incremental sweep -> snapshot -> diff
@@ -112,7 +114,7 @@ malwar crawl monitor --fail-on-malicious # exit non-zero when newly-flagged skil
 malwar crawl monitor --max 100 --no-escalate   # quick partial run, rules only
 ```
 
-**Options:** `--snapshot-dir`, `--full`, `--max`, `--no-escalate`, `--concurrency`, `--format`/`-f` (`console`|`json`), `--output`/`-o`, `--no-save`, `--digest`, `--publish`, `--fail-on-malicious`. Publishing to X requires the `MALWAR_X_*` credentials (see [Configuration](deployment/configuration.md#x-twitter-publishing)).
+**Options:** `--snapshot-dir`, `--full`, `--max-scans`, `--max`, `--no-escalate`, `--concurrency`, `--format`/`-f` (`console`|`json`), `--output`/`-o`, `--no-save`, `--digest`, `--publish`, `--fail-on-malicious`. Publishing to X requires the `MALWAR_X_*` credentials (see [Configuration](deployment/configuration.md#x-twitter-publishing)).
 
 The bundled GitHub Actions workflow (`.github/workflows/registry-monitor.yml`) runs this on two cadences: **daily incremental** and a **weekly `--full`** re-scan, committing each snapshot to the `registry-snapshots` branch.
 

src/malwar/cli/commands/crawl.py

@@ -523,6 +523,15 @@ def crawl_monitor(
             "(defends against silent same-version content swaps)",
         ),
     ] = False,
+    max_scans: Annotated[
+        int | None,
+        typer.Option(
+            "--max-scans",
+            help="Cap skills scanned per run; the overflow is deferred and "
+            "picked up next run. Lets a large registry build up over runs "
+            "instead of timing out.",
+        ),
+    ] = None,
 ) -> None:
     """Scan the registry incrementally and diff against the last snapshot.
 
@@ -547,6 +556,7 @@ def crawl_monitor(
             publish=publish,
             fail_on_malicious=fail_on_malicious,
             full=full,
+            max_scans=max_scans,
         )
     )
     raise typer.Exit(exit_code)
@@ -565,6 +575,7 @@ async def _async_monitor(
     publish: bool,
     fail_on_malicious: bool,
     full: bool = False,
+    max_scans: int | None = None,
 ) -> int:
     from rich.progress import BarColumn, Progress, TextColumn
 
@@ -595,6 +606,7 @@ def _on_progress(done: int, total: int, slug: str) -> None:
             previous=previous,
             force_rescan=full,
             max_skills=max_skills,
+            scan_budget=max_scans,
             escalate=escalate,
             concurrency=concurrency,
             on_progress=_on_progress,
@@ -604,10 +616,13 @@ def _on_progress(done: int, total: int, slug: str) -> None:
         console.print("[yellow]No skills found in the registry.[/yellow]")
         return 0
 
-    console.print(
+    summary = (
         f"[dim]Scanned {snapshot.scanned_count} changed/new, "
-        f"reused {snapshot.reused_count} unchanged.[/dim]"
+        f"reused {snapshot.reused_count} unchanged"
     )
+    if snapshot.pending_count:
+        summary += f", deferred {snapshot.pending_count} to next run"
+    console.print(summary + ".[/dim]")
 
     diff = diff_snapshots(previous, snapshot)
 

src/malwar/monitor/models.py

@@ -65,9 +65,12 @@ class RegistrySnapshot(BaseModel):
     # Slugs the crawl knew about but failed to fetch/scan, for transparency.
     errors: dict[str, str] = Field(default_factory=dict)
     # How this snapshot was built: skills actually fetched + scanned this run
-    # vs. carried forward unchanged from the previous snapshot (incremental).
+    # vs. carried forward unchanged from the previous snapshot (incremental)
+    # vs. deferred to a later run because a per-run scan budget was hit
+    # (recorded as UNKNOWN placeholders so the next run picks them up).
     scanned_count: int = 0
     reused_count: int = 0
+    pending_count: int = 0
 
     @property
     def skill_count(self) -> int:

src/malwar/monitor/snapshot.py

@@ -175,12 +175,30 @@ async def _scan_slug(client: ClawHubClient, meta: SkillMeta, *, escalate: bool)
     return record
 
 
+def _pending_record(meta: SkillMeta) -> SkillRecord:
+    """A metadata-only placeholder for a skill deferred to a later run.
+
+    Verdict is UNKNOWN with no error, so ``_is_unchanged`` returns False and the
+    next run re-scans it — this is how a budgeted baseline converges over
+    successive runs.
+    """
+    return SkillRecord(
+        slug=meta.slug,
+        display_name=meta.display_name,
+        version=meta.version,
+        updated_at=meta.updated_at,
+        installs=meta.installs,
+        verdict="UNKNOWN",
+    )
+
+
 async def build_snapshot(
     client: ClawHubClient | None = None,
     *,
     previous: RegistrySnapshot | None = None,
     force_rescan: bool = False,
     max_skills: int | None = None,
+    scan_budget: int | None = None,
     escalate: bool = True,
     concurrency: int = 8,
     page_size: int = 50,
@@ -200,7 +218,12 @@ async def build_snapshot(
         Ignore ``previous`` and re-scan every skill (a full sweep). Use this on
         a periodic cadence to catch silent same-version content swaps.
     max_skills:
-        Cap the number of skills scanned (for testing / partial runs).
+        Cap the number of skills enumerated (for testing / partial runs).
+    scan_budget:
+        Cap how many skills are actually fetched + scanned this run. Any excess
+        is recorded as an UNKNOWN placeholder and picked up on the next run, so a
+        registry too large to sweep within one run's time limit is built up over
+        successive runs rather than lost to an all-or-nothing timeout.
     escalate:
         Re-scan flagged skills with the full LLM + URL pipeline.
     concurrency:
@@ -227,13 +250,22 @@ async def build_snapshot(
         else:
             to_scan.append(meta)
 
-    snapshot.reused_count = len(snapshot.skills)
+    # Enforce the per-run scan budget: defer the overflow to a later run.
+    pending: list[SkillMeta] = []
+    if scan_budget is not None and len(to_scan) > scan_budget:
+        to_scan, pending = to_scan[:scan_budget], to_scan[scan_budget:]
+        for meta in pending:
+            snapshot.skills[meta.slug] = _pending_record(meta)
+
+    snapshot.reused_count = total - len(to_scan) - len(pending)
     snapshot.scanned_count = len(to_scan)
+    snapshot.pending_count = len(pending)
     logger.info(
-        "incremental sweep: %d skills total, %d changed/new to scan, %d reused unchanged",
+        "sweep: %d skills total, %d to scan, %d reused unchanged, %d deferred (budget)",
         total,
         snapshot.scanned_count,
         snapshot.reused_count,
+        snapshot.pending_count,
     )
 
     semaphore = asyncio.Semaphore(max(1, concurrency))

tests/unit/test_monitor.py

@@ -229,6 +229,35 @@ async def test_force_rescan_scans_everything(self):
         assert day2.scanned_count == 2
         assert day2.reused_count == 0
 
+    async def test_scan_budget_defers_overflow(self):
+        client = FakeClawHubClient({f"s{i}": BENIGN_BODY for i in range(5)})
+        snap = await build_snapshot(client, escalate=False, scan_budget=2)
+        # Only 2 scanned this run; the other 3 recorded as UNKNOWN placeholders.
+        assert snap.skill_count == 5
+        assert snap.scanned_count == 2
+        assert snap.pending_count == 3
+        assert len(client.file_fetches) == 2
+        unknown = [s for s, r in snap.skills.items() if r.verdict == "UNKNOWN"]
+        assert len(unknown) == 3
+        assert all(snap.skills[s].error is None for s in unknown)
+
+    async def test_budgeted_baseline_converges_over_runs(self):
+        skills = {f"s{i}": BENIGN_BODY for i in range(5)}
+        prev = None
+        # Three runs of budget 2 must fully cover a 5-skill registry.
+        for _ in range(3):
+            client = FakeClawHubClient(skills)
+            prev = await build_snapshot(client, previous=prev, escalate=False, scan_budget=2)
+        assert prev is not None
+        assert prev.pending_count == 0
+        assert all(r.verdict == "CLEAN" for r in prev.skills.values())
+
+    async def test_no_budget_scans_all(self):
+        client = FakeClawHubClient({f"s{i}": BENIGN_BODY for i in range(4)})
+        snap = await build_snapshot(client, escalate=False)
+        assert snap.scanned_count == 4
+        assert snap.pending_count == 0
+
     async def test_previously_errored_skill_is_retried(self):
         # Day 1: the file fetch fails, so the record carries an error.
         client1 = FakeClawHubClient({"a": BENIGN_BODY})