feat(monitor): targeted, tiered escalation for the ambiguous band

docs/crawl.md

@@ -105,16 +105,29 @@ Sweep the registry, fast-scan skills (rule engine + threat intel, escalating fla
 
 **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.
 
+**Targeted, tiered escalation.** The rules pass is confident at the two ends (obviously clean, obviously malicious) but there's an **ambiguous middle** where a sneaky skill hides — it scores just under the line. Rather than pay for a deep look at *every* flagged skill, an escalation **policy** selects only that middle band (rule risk ~8–74, below the confident-malicious line), ranks by suspicion, and caps it with `--escalate-budget`. A pluggable **backend** then gives those a second opinion:
+
+| `--escalate-backend` | What runs | Cost |
+|---|---|---|
+| `none` | nothing (policy still records candidates) | free |
+| `hf` | a free local Hugging Face classifier (CPU) — `pip install malwar[hf]` | free |
+| `anthropic` | the full LLM analyzer (rules + LLM + false-positive suppression) | paid API |
+| `tiered` | `hf` triages first; escalates to `anthropic` **only** on a hit | paid on the residual only |
+
+`tiered` is the money-saver: the free tier filters the band, and the LLM is spent only on what the free tier still finds suspicious. An `anthropic`/`tiered` result is *authoritative* — it can raise a sneaky skill or clear a rule-engine false positive; an `hf` result is recorded as a triage signal (`escalation_backend`/`escalation_verdict`/`escalation_score` on the snapshot) without overriding the rule verdict. `escalated_count` shows how many skills were sent to the backend.
+
+> **Note:** a "rule-clean but ML-anomalous" escalation path exists (`EscalationPolicy.ml_threshold`) but is **off by default** — the stock ML scorer saturates near 1.0 for both benign and malicious skills, so it can't discriminate yet. Enable it only once the model is recalibrated.
+
 ```bash
-malwar crawl monitor                     # incremental sweep -> snapshot -> diff
-malwar crawl monitor --full              # re-scan every skill (catches same-version tampering)
-malwar crawl monitor --digest            # also print a shareable digest + draft post
-malwar crawl monitor --publish           # post the digest to X when skills newly turn malicious
-malwar crawl monitor --fail-on-malicious # exit non-zero when newly-flagged skills are found (CI/cron)
-malwar crawl monitor --max 100 --no-escalate   # quick partial run, rules only
+malwar crawl monitor                          # incremental sweep -> snapshot -> diff (LLM on the ambiguous band)
+malwar crawl monitor --no-escalate            # rules only, no second opinion (fast, free)
+malwar crawl monitor --escalate-backend hf    # free local classifier on the ambiguous band
+malwar crawl monitor --escalate-backend tiered --escalate-budget 50  # hf triage -> LLM on hits, capped
+malwar crawl monitor --full                   # re-scan every skill (catches same-version tampering)
+malwar crawl monitor --fail-on-malicious      # exit non-zero when newly-flagged skills are found (CI/cron)
 ```
 
-**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)).
+**Options:** `--snapshot-dir`, `--full`, `--max-scans`, `--max`, `--escalate-backend` (`none`|`hf`|`anthropic`|`tiered`), `--escalate-budget`, `--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. It runs **rules-only (`--no-escalate`)** so a whole-registry sweep stays fast and cheap — the rule engine alone catches every malicious sample in the benchmark; use `malwar crawl scan <slug>` for an LLM deep-dive on an individual flagged skill.
 

pyproject.toml

@@ -43,6 +43,12 @@ postgres = [
 cache = [
     "redis[hiredis]>=5.0.0",
 ]
+hf = [
+    # Free, local second-opinion escalation tier (monitor.escalation). Heavy —
+    # only needed for `malwar crawl monitor --escalate-backend hf|tiered`.
+    "transformers>=4.44.0",
+    "torch>=2.2.0",
+]
 dev = [
     "pytest>=8.3.0",
     "pytest-asyncio>=0.24.0",
@@ -115,6 +121,11 @@ module = ["frontmatter", "frontmatter.*", "aiofiles", "aiofiles.*"]
 ignore_missing_imports = true
 follow_untyped_imports = true
 
+[[tool.mypy.overrides]]
+# Optional 'hf' extra — imported lazily in monitor.escalation.
+module = ["transformers"]
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_mode = "auto"

src/malwar/cli/commands/crawl.py

@@ -478,8 +478,24 @@ def crawl_monitor(
     ] = None,
     no_escalate: Annotated[
         bool,
-        typer.Option("--no-escalate", help="Never escalate flagged skills to the LLM"),
+        typer.Option("--no-escalate", help="Disable escalation (shorthand for --escalate-backend none)"),
     ] = False,
+    escalate_backend: Annotated[
+        str,
+        typer.Option(
+            "--escalate-backend",
+            help="Second-opinion backend for the ambiguous band: "
+            "none | hf (free local classifier) | anthropic (LLM) | "
+            "tiered (hf triage -> anthropic on hits)",
+        ),
+    ] = "anthropic",
+    escalate_budget: Annotated[
+        int | None,
+        typer.Option(
+            "--escalate-budget",
+            help="Cap how many skills are escalated per run (most-suspicious first)",
+        ),
+    ] = None,
     concurrency: Annotated[
         int, typer.Option("--concurrency", help="Skills scanned in parallel")
     ] = 8,
@@ -547,7 +563,8 @@ def crawl_monitor(
         _async_monitor(
             snapshot_dir=snapshot_dir,
             max_skills=max_skills,
-            escalate=not no_escalate,
+            escalate_backend="none" if no_escalate else escalate_backend,
+            escalate_budget=escalate_budget,
             concurrency=concurrency,
             fmt=fmt,
             output=output,
@@ -566,7 +583,8 @@ async def _async_monitor(
     *,
     snapshot_dir: Path,
     max_skills: int | None,
-    escalate: bool,
+    escalate_backend: str,
+    escalate_budget: int | None,
     concurrency: int,
     fmt: CrawlFormat,
     output: Path | None,
@@ -579,11 +597,24 @@ async def _async_monitor(
 ) -> int:
     from rich.progress import BarColumn, Progress, TextColumn
 
-    from malwar.monitor import SnapshotStore, build_snapshot, diff_snapshots
+    from malwar.monitor import (
+        EscalationPolicy,
+        SnapshotStore,
+        build_snapshot,
+        diff_snapshots,
+        make_backend,
+    )
 
     store = SnapshotStore(snapshot_dir)
     previous = store.load_latest()
 
+    try:
+        backend = make_backend(escalate_backend)
+    except ValueError as exc:
+        console.print(f"[red]{exc}[/red]")
+        return 2
+    policy = EscalationPolicy(budget=escalate_budget)
+
     mode = "full re-scan" if full else ("incremental" if previous else "first run")
     scope = f" (max {max_skills})" if max_skills else ""
     console.print(f"Crawling ClawHub registry{scope} [{mode}]...", style="dim")
@@ -607,7 +638,8 @@ def _on_progress(done: int, total: int, slug: str) -> None:
             force_rescan=full,
             max_skills=max_skills,
             scan_budget=max_scans,
-            escalate=escalate,
+            escalation=backend,
+            escalation_policy=policy,
             concurrency=concurrency,
             on_progress=_on_progress,
         )
@@ -622,6 +654,8 @@ def _on_progress(done: int, total: int, slug: str) -> None:
     )
     if snapshot.pending_count:
         summary += f", deferred {snapshot.pending_count} to next run"
+    if snapshot.escalated_count:
+        summary += f", escalated {snapshot.escalated_count} to '{backend.name}'"
     console.print(summary + ".[/dim]")
 
     diff = diff_snapshots(previous, snapshot)

src/malwar/monitor/__init__.py

@@ -4,6 +4,12 @@
 from __future__ import annotations
 
 from malwar.monitor.diff import diff_snapshots
+from malwar.monitor.escalation import (
+    EscalationPolicy,
+    EscalationResult,
+    make_backend,
+    select_candidates,
+)
 from malwar.monitor.models import (
     RegistrySnapshot,
     SkillChange,
@@ -14,13 +20,17 @@
 from malwar.monitor.snapshot import SnapshotStore, build_snapshot
 
 __all__ = [
+    "EscalationPolicy",
+    "EscalationResult",
     "RegistrySnapshot",
     "SkillChange",
     "SkillRecord",
     "SnapshotDiff",
     "SnapshotStore",
     "build_snapshot",
     "diff_snapshots",
+    "make_backend",
     "render_digest",
     "render_tweet",
+    "select_candidates",
 ]