From da565488f5f0a6eef5540889169eb4ea2f6fa911 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 05:10:17 +0000 Subject: [PATCH] feat(monitor): budgeted, resumable baseline via --max-scans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ClawHub is rate-limited to ~2 req/s, so a full ~6k-skill registry sweep takes >60 min and can't finish in one CI run — and because the snapshot only commits after scanning everything, every timed-out baseline saved nothing. (Confirmed empirically: five runs, none completed.) Make the baseline resumable: - build_snapshot() gains scan_budget: cap how many skills are actually fetched+scanned per run; the overflow is recorded as UNKNOWN placeholders (metadata only, no error). Because UNKNOWN records aren't 'unchanged', the next run re-scans them — so the baseline builds up over successive runs and then daily incremental only touches what changed. - crawl monitor --max-scans N wires it; snapshot records pending_count. - Workflow passes --max-scans 1500 (~one CI run's worth) so the baseline converges over a few runs instead of timing out. Tests: budget defers overflow, baseline converges over runs, no-budget scans all. Docs updated. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01DNoTXU8k3pfSBzR7aJubqL --- .github/workflows/registry-monitor.yml | 5 ++++ docs/crawl.md | 6 ++-- src/malwar/cli/commands/crawl.py | 19 +++++++++++-- src/malwar/monitor/models.py | 5 +++- src/malwar/monitor/snapshot.py | 38 ++++++++++++++++++++++++-- tests/unit/test_monitor.py | 29 ++++++++++++++++++++ 6 files changed, 94 insertions(+), 8 deletions(-) diff --git a/.github/workflows/registry-monitor.yml b/.github/workflows/registry-monitor.yml index 331a5ff..e06c027 100644 --- a/.github/workflows/registry-monitor.yml +++ b/.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 diff --git a/docs/crawl.md b/docs/crawl.md index d80b906..70c6ce2 100644 --- a/docs/crawl.md +++ b/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. diff --git a/src/malwar/cli/commands/crawl.py b/src/malwar/cli/commands/crawl.py index 508ea8c..3c15b46 100644 --- a/src/malwar/cli/commands/crawl.py +++ b/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) diff --git a/src/malwar/monitor/models.py b/src/malwar/monitor/models.py index ebb448b..e4a7bba 100644 --- a/src/malwar/monitor/models.py +++ b/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: diff --git a/src/malwar/monitor/snapshot.py b/src/malwar/monitor/snapshot.py index 896f858..22f3eb7 100644 --- a/src/malwar/monitor/snapshot.py +++ b/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)) diff --git a/tests/unit/test_monitor.py b/tests/unit/test_monitor.py index 52c6fe9..f9fdf9e 100644 --- a/tests/unit/test_monitor.py +++ b/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})