Add variant alerts to alerter and daily token budget to KernelSense

Alerter now consumes both semantic_deltas and kernelsense tasks, sending
Telegram alerts for confirmed cross-driver variants. LLM client tracks
daily token usage in Redis with auto-expire and blocks calls when a
configurable budget ceiling is reached.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: splintersfury

docker-compose.yml

@@ -436,6 +436,7 @@ services:
       KERNELSENSE_FP_THRESHOLD: "4.0"
       KERNELSENSE_VARIANT_ENABLED: "true"
       KERNELSENSE_VARIANT_CONFIDENCE: "0.70"
+      KERNELSENSE_DAILY_TOKEN_BUDGET: ${KERNELSENSE_DAILY_TOKEN_BUDGET:-0}
       KERNELSENSE_EMBEDDINGS_DIR: /data/embeddings
     volumes:
       - ./karton.ini:/etc/karton/karton.ini:ro

services/autopiff-alerter/karton_alerter.py

@@ -1,9 +1,13 @@
 #!/usr/bin/env python3
 """
-AutoPiff Alerter — Karton consumer that sends Telegram alerts for high-scoring findings.
+AutoPiff Alerter — Karton consumer that sends Telegram alerts.
+
+Consumes:
+  - {type: autopiff, kind: semantic_deltas} — high-scoring patch findings
+  - {type: autopiff, kind: kernelsense}     — confirmed variant vulnerabilities
 
-Consumes: {type: autopiff, kind: semantic_deltas}
 Filters:  final_score >= 8.0 AND surface_area in [ioctl, irp, filesystem]
+          variant_candidates where is_variant=true AND confidence >= threshold
 Sends:    Telegram alerts via HTTP API
 Stores:   Recent alerts in Redis sorted set (30-day TTL)
 """
@@ -28,25 +32,45 @@
 TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID")
 REDIS_HOST = os.environ.get("KARTON_REDIS_HOST", "karton-redis")
 SCORE_THRESHOLD = float(os.environ.get("AUTOPIFF_SCORE_THRESHOLD", "8.0"))
+VARIANT_CONFIDENCE_THRESHOLD = float(
+    os.environ.get("AUTOPIFF_VARIANT_CONFIDENCE", "0.65")
+)
 ALERTABLE_SURFACES = {"ioctl", "irp", "filesystem"}
 
 # Redis keys
 ALERTS_KEY = "autopiff:alerts:recent"
 ALERTS_FAILED_KEY = "autopiff:alerts:failed"
+VARIANT_ALERTS_KEY = "autopiff:alerts:variants"
 ALERTS_TTL_SECONDS = 30 * 24 * 3600  # 30 days
 
 
 class AutoPiffAlerter(Karton):
-    """Karton consumer that filters high-scoring AutoPiff findings and sends Telegram alerts."""
+    """Karton consumer that sends Telegram alerts for high-scoring findings and variants."""
 
     identity = "karton.autopiff.alerter"
-    filters = [{"type": "autopiff", "kind": "semantic_deltas"}]
+    filters = [
+        {"type": "autopiff", "kind": "semantic_deltas"},
+        {"type": "autopiff", "kind": "kernelsense"},
+    ]
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.rdb = redis.Redis(host=REDIS_HOST, port=6379, decode_responses=True)
 
     def process(self, task: Task) -> None:
+        kind = task.headers.get("kind")
+        if kind == "semantic_deltas":
+            self._process_semantic_deltas(task)
+        elif kind == "kernelsense":
+            self._process_kernelsense(task)
+        else:
+            logger.warning(f"Unknown task kind: {kind}")
+
+    # ------------------------------------------------------------------
+    # Semantic Deltas (original patch-based alerting)
+    # ------------------------------------------------------------------
+
+    def _process_semantic_deltas(self, task: Task) -> None:
         semantic_deltas = task.get_payload("semantic_deltas")
         if not semantic_deltas:
             logger.warning("No semantic_deltas payload in task")
@@ -89,15 +113,121 @@ def process(self, task: Task) -> None:
 
         logger.info(f"Found {len(alertable)} alertable findings")
 
-        # Build and send alert
         msg = self._build_alert_message(
             alertable, driver_new_sha, driver_new_ver, driver_old_ver, summary
         )
         self._send_telegram_alert(msg)
-
-        # Store in Redis for /findings command
         self._store_alerts(alertable, driver_new_sha)
 
+    # ------------------------------------------------------------------
+    # KernelSense (variant alerting)
+    # ------------------------------------------------------------------
+
+    def _process_kernelsense(self, task: Task) -> None:
+        ks_raw = task.headers.get("kernelsense")
+        if isinstance(ks_raw, str):
+            ks_data = json.loads(ks_raw)
+        else:
+            ks_data = ks_raw
+
+        if not ks_data:
+            logger.warning("No kernelsense data in task")
+            return
+
+        findings = ks_data.get("findings", [])
+        driver_new = ks_data.get("driver_new", {})
+        driver_name = driver_new.get("name", driver_new.get("sha256", "unknown"))
+
+        # Collect all confirmed variants across all findings
+        all_variants = []
+        for finding in findings:
+            candidates = finding.get("variant_candidates", [])
+            if not candidates:
+                continue
+
+            assessment = finding.get("llm_assessment", {})
+            if not assessment.get("is_security_fix"):
+                continue
+
+            confirmed = [
+                c for c in candidates
+                if c.get("is_variant")
+                and c.get("confidence", 0) >= VARIANT_CONFIDENCE_THRESHOLD
+            ]
+
+            for variant in confirmed:
+                all_variants.append({
+                    "source_function": finding.get("function", "unknown"),
+                    "source_driver": driver_name,
+                    "bug_class": assessment.get("bug_class", "unknown"),
+                    "source_confidence": assessment.get("confidence", 0),
+                    **variant,
+                })
+
+        if not all_variants:
+            logger.info("No confirmed variants above threshold")
+            return
+
+        logger.info(f"Found {len(all_variants)} confirmed variant(s)")
+
+        msg = self._build_variant_alert(all_variants)
+        self._send_telegram_alert(msg)
+        self._store_variant_alerts(all_variants)
+
+    def _build_variant_alert(self, variants: list[dict]) -> str:
+        count = len(variants)
+        source = variants[0]
+
+        msg = (
+            f"*AutoPiff Variant Alert* — "
+            f"{count} potential variant{'s' if count > 1 else ''} found\n\n"
+        )
+        msg += (
+            f"Known vulnerability: {source['bug_class']} in "
+            f"{source['source_driver']}/`{source['source_function']}`\n"
+            f"Source confidence: *{source['source_confidence']:.2f}*\n\n"
+        )
+
+        for i, v in enumerate(variants[:5]):
+            msg += (
+                f"*{i+1}.* {v['driver']} / `{v['function']}` — "
+                f"similarity *{v['similarity']:.2f}*\n"
+            )
+            msg += f"   {v.get('match_type', 'unknown')} | confidence: {v['confidence']:.2f}\n"
+            reasoning = v.get("reasoning", "")
+            if reasoning:
+                msg += f"   _{reasoning[:100]}_\n"
+            msg += "\n"
+
+        if count > 5:
+            msg += f"_...and {count - 5} more variant(s)_\n"
+
+        return msg
+
+    def _store_variant_alerts(self, variants: list[dict]) -> None:
+        now = time.time()
+        pipe = self.rdb.pipeline()
+        for v in variants:
+            entry = {
+                "source_driver": v["source_driver"],
+                "source_function": v["source_function"],
+                "bug_class": v["bug_class"],
+                "variant_driver": v["driver"],
+                "variant_function": v["function"],
+                "similarity": v["similarity"],
+                "confidence": v["confidence"],
+                "reasoning": v.get("reasoning", ""),
+            }
+            pipe.zadd(VARIANT_ALERTS_KEY, {json.dumps(entry): now})
+
+        cutoff = now - ALERTS_TTL_SECONDS
+        pipe.zremrangebyscore(VARIANT_ALERTS_KEY, "-inf", cutoff)
+        pipe.execute()
+
+    # ------------------------------------------------------------------
+    # Shared: message building, Telegram, Redis
+    # ------------------------------------------------------------------
+
     def _build_alert_message(
         self, findings, driver_sha, new_ver, old_ver, summary
     ) -> str:

services/karton-kernelsense/llm_client.py

@@ -5,12 +5,14 @@
 1. Prepends the disclosure boundary to every system prompt
 2. Handles retries with exponential backoff
 3. Tracks token usage to Redis for cost monitoring
+4. Enforces a daily token budget to prevent runaway API costs
 """
 
 import json
 import logging
 import os
 import time
+from datetime import datetime, timezone
 
 import anthropic
 import redis
@@ -39,6 +41,11 @@ def __init__(self):
 
         self.client = anthropic.Anthropic()
 
+        # Daily token budget (0 = unlimited)
+        self.daily_token_budget = int(
+            os.environ.get("KERNELSENSE_DAILY_TOKEN_BUDGET", "0")
+        )
+
         # Redis for usage tracking (optional)
         redis_host = os.environ.get("KARTON_REDIS_HOST", "localhost")
         try:
@@ -64,6 +71,21 @@ def analyze(self, prompt: str, task_context: str = "") -> dict:
         Returns:
             Parsed JSON response from the LLM, or error dict.
         """
+        # Check daily budget before making the API call
+        if self.daily_token_budget > 0:
+            over, used = self._check_daily_budget()
+            if over:
+                logger.warning(
+                    f"Daily token budget exceeded ({used}/{self.daily_token_budget}), "
+                    f"skipping LLM call for: {task_context}"
+                )
+                return {
+                    "error": "daily_token_budget_exceeded",
+                    "is_security_fix": False,
+                    "budget_used": used,
+                    "budget_limit": self.daily_token_budget,
+                }
+
         system = BOUNDARY_PROMPT
         if task_context:
             system += f"\nContext: {task_context}"
@@ -142,20 +164,67 @@ def _parse_response(self, text: str) -> dict:
         logger.warning(f"Could not parse LLM response as JSON: {text[:200]}...")
         return {"error": "unparseable response", "raw_text": text[:500]}
 
+    def _daily_key(self) -> str:
+        """Return the Redis key for today's usage."""
+        today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+        return f"kernelsense:usage:daily:{today}"
+
+    def _check_daily_budget(self) -> tuple[bool, int]:
+        """Check if today's token usage exceeds the daily budget.
+
+        Returns:
+            (over_budget, tokens_used) tuple.
+        """
+        if not self.redis:
+            return False, 0
+
+        try:
+            key = self._daily_key()
+            used_in = int(self.redis.hget(key, "input_tokens") or 0)
+            used_out = int(self.redis.hget(key, "output_tokens") or 0)
+            total_used = used_in + used_out
+            return total_used >= self.daily_token_budget, total_used
+        except redis.RedisError as e:
+            logger.debug(f"Failed to check daily budget: {e}")
+            return False, 0
+
     def _track_usage(self, usage, context: str = "") -> None:
         """Log token usage to Redis for cost monitoring."""
         if not self.redis:
             return
 
+        total_tokens = usage.input_tokens + usage.output_tokens
+
         try:
+            # Lifetime totals
             key = "kernelsense:usage:total"
             self.redis.hincrby(key, "input_tokens", usage.input_tokens)
             self.redis.hincrby(key, "output_tokens", usage.output_tokens)
             self.redis.hincrby(key, "calls", 1)
 
+            # Daily totals (auto-expire after 7 days for cleanup)
+            daily_key = self._daily_key()
+            pipe = self.redis.pipeline()
+            pipe.hincrby(daily_key, "input_tokens", usage.input_tokens)
+            pipe.hincrby(daily_key, "output_tokens", usage.output_tokens)
+            pipe.hincrby(daily_key, "calls", 1)
+            pipe.expire(daily_key, 7 * 24 * 3600)
+            pipe.execute()
+
             logger.debug(
                 f"LLM usage ({context}): "
                 f"{usage.input_tokens} in / {usage.output_tokens} out"
             )
+
+            # Warn when approaching budget
+            if self.daily_token_budget > 0:
+                daily_total = int(self.redis.hget(daily_key, "input_tokens") or 0) + \
+                              int(self.redis.hget(daily_key, "output_tokens") or 0)
+                pct = daily_total / self.daily_token_budget * 100
+                if pct >= 90:
+                    logger.warning(
+                        f"Daily token usage at {pct:.0f}% "
+                        f"({daily_total}/{self.daily_token_budget})"
+                    )
         except redis.RedisError as e:
             logger.debug(f"Failed to track usage: {e}")