Feature: Manual/Automated Switch to Burn Mode for Exploit Detection

src/crusades/api/server.py

@@ -28,14 +28,17 @@
 import logging
 import os
 from contextlib import asynccontextmanager
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
-from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException, Query, Request
 from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from crusades import COMPETITION_VERSION
+from crusades.chain.burn_mode import BurnMode
+from crusades.storage.database import Database
 from crusades.tui.client import DatabaseClient, MockClient
 
 logger = logging.getLogger(__name__)
@@ -65,8 +68,17 @@ async def lifespan(app: FastAPI):
     # Startup
     logger.info("Crusades API starting...")
     get_db_client()
+
+    # Initialize async database for burn mode operations
+    db_path = os.getenv("CRUSADES_DB_PATH", "crusades.db")
+    db = Database(url=f"sqlite+aiosqlite:///{db_path}")
+    await db.initialize()
+    app.state.db = db
+
     yield
+
     # Shutdown
+    await db.close()
     global _db_client
     if _db_client:
         _db_client.close()
@@ -82,6 +94,65 @@ def verify_api_key(request: Request, x_api_key: str | None = Header(None)):
             raise HTTPException(status_code=401, detail="Invalid or missing API key")
 
 
+# ============================================================
+# Burn Mode Router (defined before create_app so it can be included)
+# ============================================================
+
+
+class BurnModeActivateRequest(BaseModel):
+    """Request body for burn mode activation."""
+
+    burn_rate_override: float = Field(default=1.0, ge=0.0, le=1.0)
+    blocked_uids: list[int] = Field(default_factory=list)
+    reason: str = ""
+
+
+burn_mode_router = APIRouter(prefix="/burn-mode", tags=["burn-mode"])
+
+
+@burn_mode_router.post("/activate")
+async def activate_burn_mode(
+    request: Request,
+    body: BurnModeActivateRequest,
+) -> dict[str, Any]:
+    """Activate burn mode. Overrides hparams burn_rate and optionally blocks UIDs."""
+    db: Database = request.app.state.db
+    burn_mode = BurnMode(
+        enabled=True,
+        burn_rate_override=body.burn_rate_override,
+        blocked_uids=body.blocked_uids,
+        reason=body.reason,
+        activated_at=datetime.now(UTC),
+        activated_by="api",
+    )
+    await db.set_burn_mode(burn_mode)
+    logger.warning(
+        "Burn mode ACTIVATED: rate=%.0f%%, blocked=%s, reason=%s",
+        burn_mode.burn_rate_override * 100,
+        burn_mode.blocked_uids,
+        burn_mode.reason,
+    )
+    return burn_mode.model_dump(mode="json")
+
+
+@burn_mode_router.post("/deactivate")
+async def deactivate_burn_mode(request: Request) -> dict[str, Any]:
+    """Deactivate burn mode. Restores normal hparams-driven weight distribution."""
+    db: Database = request.app.state.db
+    burn_mode = BurnMode.inactive()
+    await db.set_burn_mode(burn_mode)
+    logger.warning("Burn mode DEACTIVATED")
+    return {"status": "deactivated"}
+
+
+@burn_mode_router.get("/status")
+async def get_burn_mode_status(request: Request) -> dict[str, Any]:
+    """Get current burn mode state."""
+    db: Database = request.app.state.db
+    burn_mode = await db.get_burn_mode()
+    return burn_mode.model_dump(mode="json")
+
+
 def create_app(api_key: str | None = None) -> FastAPI:
     """Create and configure the FastAPI app.
 
@@ -102,13 +173,16 @@ def create_app(api_key: str | None = None) -> FastAPI:
         CORSMiddleware,
         allow_origins=["*"],  # In production, restrict to your domain
         allow_credentials=False,
-        allow_methods=["GET"],
+        allow_methods=["GET", "POST"],
         allow_headers=["*"],
     )
 
     # Store API key for authentication
     new_app.state.api_key = api_key or os.getenv("DASHBOARD_API_KEY")
 
+    # Include burn mode router
+    new_app.include_router(burn_mode_router)
+
     return new_app
 
 

src/crusades/chain/burn_mode.py

@@ -0,0 +1,28 @@
+"""Burn mode state for exploit detection and emergency emission control."""
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class BurnMode(BaseModel):
+    """Runtime-toggleable burn mode that overrides normal weight distribution.
+
+    When enabled, overrides the hparams burn_rate and optionally blocks
+    specific UIDs from receiving emissions.
+    """
+
+    enabled: bool = False
+    burn_rate_override: float = Field(default=1.0, ge=0.0, le=1.0)
+    blocked_uids: list[int] = Field(default_factory=list)
+    reason: str = ""
+    activated_at: datetime | None = None
+    activated_by: str = ""
+
+    @classmethod
+    def inactive(cls) -> "BurnMode":
+        """Return a default inactive burn mode state."""
+        return cls(enabled=False)
+
+
+BURN_MODE_KEY = "burn_mode"

src/crusades/chain/weights.py

@@ -76,6 +76,16 @@ async def set_weights(self) -> tuple[bool, str]:
         Returns:
             Tuple of (success, message)
         """
+        # Load burn mode state — takes precedence over hparams burn_rate
+        burn_mode = await self.db.get_burn_mode()
+        if burn_mode.enabled:
+            logger.warning(
+                "Burn mode ACTIVE: rate_override=%.0f%%, blocked_uids=%s, reason=%s",
+                burn_mode.burn_rate_override * 100,
+                burn_mode.blocked_uids,
+                burn_mode.reason,
+            )
+
         # Sync metagraph to get latest state
         await self.chain.sync_metagraph()
 
@@ -200,12 +210,25 @@ async def set_weights(self) -> tuple[bool, str]:
             self._previous_winner_score = winner_score
             await self._save_previous_winner(winner.submission_id, winner_score)
 
-        # Calculate weight distribution
-        winner_weight = 1.0 - self.burn_rate
-        burn_weight = self.burn_rate
+        # If burn mode is active and winner is blocked, redirect to burn
+        if burn_mode.enabled and winner_uid in burn_mode.blocked_uids:
+            logger.warning(
+                "Winner UID %d is blocked by burn mode — redirecting 100%% to burn_uid",
+                winner_uid,
+            )
+            return await self._set_burn_only_weights(
+                f"Winner UID {winner_uid} blocked by burn mode"
+            )
+
+        # Calculate weight distribution (burn mode overrides hparams burn_rate)
+        effective_burn_rate = (
+            burn_mode.burn_rate_override if burn_mode.enabled else self.burn_rate
+        )
+        winner_weight = 1.0 - effective_burn_rate
+        burn_weight = effective_burn_rate
 
         logger.info(
-            f"Setting weights with burn_rate={self.burn_rate:.0%}:\n"
+            f"Setting weights with burn_rate={effective_burn_rate:.0%}:\n"
             f"  - UID {self.burn_uid} (burn): {burn_weight:.2f}\n"
             f"  - UID {winner_uid} (winner, MFU={winner_score:.2f}%): {winner_weight:.2f}"
         )

src/crusades/storage/database.py

@@ -6,6 +6,7 @@
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
+from ..chain.burn_mode import BURN_MODE_KEY, BurnMode
 from ..config import get_hparams
 from ..core.protocols import SubmissionStatus
 from .models import (
@@ -487,6 +488,19 @@ async def get_payment_for_submission(self, submission_id: str) -> VerifiedPaymen
             )
             return result.scalar_one_or_none()
 
+    # Burn mode operations
+
+    async def get_burn_mode(self) -> BurnMode:
+        """Load burn mode state from validator_state KV store."""
+        raw = await self.get_validator_state(BURN_MODE_KEY)
+        if raw is None:
+            return BurnMode.inactive()
+        return BurnMode.model_validate_json(raw)
+
+    async def set_burn_mode(self, burn_mode: BurnMode) -> None:
+        """Persist burn mode state to validator_state KV store."""
+        await self.set_validator_state(BURN_MODE_KEY, burn_mode.model_dump_json())
+
     # Adaptive threshold operations
 
     async def get_adaptive_threshold(

tests/test_burn_mode.py

@@ -0,0 +1,100 @@
+"""Tests for burn mode model and database persistence."""
+
+from datetime import UTC, datetime
+
+import pytest
+
+from crusades.chain.burn_mode import BURN_MODE_KEY, BurnMode
+from crusades.storage.database import Database
+
+# -- BurnMode model tests --
+
+
+class TestBurnModeModel:
+    def test_inactive_factory(self):
+        bm = BurnMode.inactive()
+        assert bm.enabled is False
+        assert bm.burn_rate_override == 1.0
+        assert bm.blocked_uids == []
+        assert bm.reason == ""
+        assert bm.activated_at is None
+
+    def test_serialization_roundtrip(self):
+        bm = BurnMode(
+            enabled=True,
+            burn_rate_override=0.5,
+            blocked_uids=[15, 42],
+            reason="exploit detected",
+            activated_at=datetime(2026, 1, 1, tzinfo=UTC),
+            activated_by="operator",
+        )
+        raw = bm.model_dump_json()
+        restored = BurnMode.model_validate_json(raw)
+        assert restored.enabled is True
+        assert restored.burn_rate_override == 0.5
+        assert restored.blocked_uids == [15, 42]
+        assert restored.reason == "exploit detected"
+        assert restored.activated_by == "operator"
+
+    def test_burn_rate_bounds_low(self):
+        with pytest.raises(Exception):
+            BurnMode(burn_rate_override=-0.1)
+
+    def test_burn_rate_bounds_high(self):
+        with pytest.raises(Exception):
+            BurnMode(burn_rate_override=1.1)
+
+    def test_burn_rate_at_boundaries(self):
+        bm_zero = BurnMode(burn_rate_override=0.0)
+        assert bm_zero.burn_rate_override == 0.0
+        bm_one = BurnMode(burn_rate_override=1.0)
+        assert bm_one.burn_rate_override == 1.0
+
+
+# -- Database persistence tests --
+
+
+@pytest.fixture
+async def db():
+    """In-memory async database for testing."""
+    database = Database(url="sqlite+aiosqlite:///:memory:")
+    await database.initialize()
+    yield database
+    await database.close()
+
+
+class TestBurnModePersistence:
+    async def test_get_returns_inactive_when_unset(self, db: Database):
+        bm = await db.get_burn_mode()
+        assert bm.enabled is False
+
+    async def test_set_then_get_roundtrip(self, db: Database):
+        original = BurnMode(
+            enabled=True,
+            burn_rate_override=1.0,
+            blocked_uids=[15],
+            reason="skip backward exploit",
+            activated_at=datetime(2026, 4, 4, tzinfo=UTC),
+            activated_by="api",
+        )
+        await db.set_burn_mode(original)
+        loaded = await db.get_burn_mode()
+        assert loaded.enabled is True
+        assert loaded.burn_rate_override == 1.0
+        assert loaded.blocked_uids == [15]
+        assert loaded.reason == "skip backward exploit"
+
+    async def test_overwrite_existing(self, db: Database):
+        await db.set_burn_mode(BurnMode(enabled=True, blocked_uids=[1, 2]))
+        await db.set_burn_mode(BurnMode.inactive())
+        loaded = await db.get_burn_mode()
+        assert loaded.enabled is False
+        assert loaded.blocked_uids == []
+
+    async def test_persists_via_validator_state(self, db: Database):
+        """Burn mode is stored as a validator_state KV entry."""
+        bm = BurnMode(enabled=True, reason="test")
+        await db.set_burn_mode(bm)
+        raw = await db.get_validator_state(BURN_MODE_KEY)
+        assert raw is not None
+        assert '"enabled":true' in raw.lower() or '"enabled": true' in raw.lower()