feat(alerts): vertical slice — pure-Python domain + thin v1 route

app/domain/alerts/ (NO FastAPI imports, pure business logic):
models.py Pydantic v2: AlertType enum, AlertSubscription, CreateAlertRequest, AlertEvent
repository.py Async Redis (hset/hgetall) — uses core/redis
service.py CRUD + fire_event + severity mapping
broadcaster.py WebSocket push via core/websocket
__init__.py Public API: AlertService, AlertRepository, AlertBroadcaster, models

app/api/v1/auth/alerts.py (THIN route, <100 lines):
POST /api/v1/alerts/subscribe
GET /api/v1/alerts
DELETE /api/v1/alerts/{alert_id}
Uses get_optional_user from core/auth (delegates to legacy JWT verify).

app/api/v1/__init__.py — aggregator mounts the new alerts router alongside legacy.

app/core/auth.py — added get_optional_user + get_current_user JWT deps,
delegating to legacy app.auth (stranglerfig).

tests/unit/domain/alerts/test_service.py — 7 unit tests, all pass:
test_create_subscription_assigns_id_and_persists
test_create_subscription_with_explicit_types
test_list_filters_by_owner
test_cancel_unauthorized_returns_false
test_cancel_owner_succeeds
test_fire_event_broadcasts_to_matching_subs
test_severity_mapping

PROVES THE PATTERN:
- Domain layer is unit-testable WITHOUT spinning up FastAPI.
- Service uses core/redis (single source of truth) — no scattered get_redis().
- Service uses core/logging (structlog) — no scattered logging.basicConfig.
- Service uses core/auth (single source of truth) — no scattered JWT decode.
- Route is THIN: parse → call service → return. No business logic in HTTP.
- Pydantic v2 models in domain, no dict types crossing boundaries.
- <500 lines per file (largest is service.py at ~150 lines).

Backend healthy, all 3 new routes in openapi, legacy /api/v1/alerts/* still
serving (stranglerfig — both work side-by-side, legacy wins on path conflict).

backend/app/api/v1/__init__.py

@@ -24,6 +24,20 @@ api_v1_router: list[APIRouter] = []
 router = APIRouter(prefix="/api/v1", tags=["v1"])
 
 
+# ── Migrated domains ───────────────────────────────────────────────────
+# Each migrated domain is imported here. The router exposes endpoints
+# at /api/v1/<domain>/* (path defined per-router).
+#
+# During strangelfig, the LEGACY /api/v1/alerts/* endpoints remain
+# mounted in main.py. The new v1 router is mounted at the same path
+# (FastAPI handles prefix-based routing) — first match wins, so the
+# legacy stays until we explicitly remove it.
+
+from app.api.v1.auth.alerts import router as alerts_router  # noqa: E402
+
+api_v1_router.append(alerts_router)
+
+
 def build_v1_router() -> APIRouter:
     """Construct the v1 aggregator with all migrated routes mounted."""
     aggregated = APIRouter(prefix="/api/v1")

backend/app/api/v1/auth/alerts.py

@@ -0,0 +1,70 @@
+"""V1 alerts route — thin HTTP layer over app.domain.alerts.
+
+Rules (per 2026 standards):
+  - Parse request → call service → return response. No business logic here.
+  - Pydantic models for request/response, defined in app.domain.alerts.models.
+  - No direct Redis or DB access. Goes through AlertService.
+  - All errors raised as AppError subclasses, handled by core.errors handlers.
+"""
+from __future__ import annotations
+
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends
+
+from app.core.auth import get_optional_user
+from app.core.errors import AuthError, NotFoundError
+from app.domain.alerts import (
+    AlertService,
+    AlertSubscription,
+    CreateAlertRequest,
+)
+from app.models import PaginatedResponse
+
+router = APIRouter(prefix="/api/v1/alerts", tags=["alerts"])
+
+
+def _service() -> AlertService:
+    """Dependency: a fresh AlertService per request (stateless)."""
+    return AlertService()
+
+
+@router.post("/subscribe", response_model=AlertSubscription, status_code=201)
+async def subscribe(
+    req: CreateAlertRequest,
+    user: Annotated[dict[str, Any] | None, Depends(get_optional_user)],
+    svc: Annotated[AlertService, Depends(_service)],
+) -> AlertSubscription:
+    """Create a new alert subscription for a token."""
+    owner_id = (user or {}).get("id")
+    return await svc.create_subscription(req, owner_id=owner_id)
+
+
+@router.get("", response_model=PaginatedResponse)
+async def list_alerts(
+    user: Annotated[dict[str, Any] | None, Depends(get_optional_user)],
+    svc: Annotated[AlertService, Depends(_service)],
+) -> PaginatedResponse:
+    """List alert subscriptions. Authenticated users see their own; anonymous sees all (legacy behavior)."""
+    owner_id = (user or {}).get("id")
+    items = await svc.list_subscriptions(owner_id=owner_id)
+    return PaginatedResponse(
+        items=[i.model_dump(mode="json") for i in items],
+        total=len(items),
+    )
+
+
+@router.delete("/{alert_id}", status_code=204)
+async def cancel(
+    alert_id: str,
+    user: Annotated[dict[str, Any] | None, Depends(get_optional_user)],
+    svc: Annotated[AlertService, Depends(_service)],
+) -> None:
+    """Cancel (delete) a subscription. Owner only."""
+    owner_id = (user or {}).get("id")
+    if owner_id is None:
+        raise AuthError("authentication required to cancel alerts")
+    ok = await svc.cancel_subscription(alert_id, owner_id=owner_id)
+    if not ok:
+        raise NotFoundError(f"alert {alert_id} not found or not owned by you")
+    return None

backend/app/core/auth.py

@@ -1,6 +1,13 @@
-"""RMI Backend — Auth middleware and API key verification."""
+"""RMI Backend — Auth middleware, API key verification, and JWT user identity.
+
+This module is the single source of truth for auth. Routes import
+`get_current_user` / `get_optional_user` from here (re-exported via
+app.api.deps for convenience).
+"""
+from __future__ import annotations
 
 import os
+from typing import Any
 
 from fastapi import Request
 from fastapi.responses import JSONResponse
@@ -27,7 +34,6 @@ PUBLIC_WRITE_PREFIXES = [
     "/api/v1/wallet-manager/",
 ]
 
-# Auth bypass paths
 PUBLIC_GET_PREFIXES = [
     "/api/v1/token/",
     "/api/v1/databus/",
@@ -63,3 +69,30 @@ class AuthMiddleware(BaseHTTPMiddleware):
                 content={"detail": "Unauthorized - valid X-API-Key header required for write operations"},
             )
         return await call_next(request)
+
+
+# ── JWT user identity (FastAPI dependencies) ─────────────────────────────
+# Delegates to the legacy app.auth JWT logic during strangelfig migration.
+# Once legacy auth.py is migrated to the new pattern, these become the
+# canonical implementation. Until then they reuse the working logic so
+# new routes (like app/api/v1/auth/alerts.py) can use modern Depends().
+
+async def get_optional_user(request: Request) -> dict[str, Any] | None:
+    """Return the authenticated user dict, or None if not authenticated.
+
+    Reads the Authorization: Bearer <jwt> header. Returns the user dict
+    (id, email, tier, role) on success, None if no/invalid token.
+
+    Use this for endpoints that work with OR without auth.
+    """
+    from app.auth import get_current_user as _legacy_get_user  # local import to avoid cycles
+    return await _legacy_get_user(request)
+
+
+async def get_current_user(request: Request) -> dict[str, Any]:
+    """Require an authenticated user. Raises 401 if missing.
+
+    Use this for endpoints that REQUIRE auth.
+    """
+    from app.auth import require_auth as _legacy_require  # local import to avoid cycles
+    return await _legacy_require(request)

backend/app/domain/__init__.py

@@ -0,0 +1,15 @@
+"""Pure business logic. NO FastAPI imports allowed.
+
+Per-domain packages (each follows: models.py + service.py + helpers):
+    - scanner:  token scanning, honeypot, rugcheck, holders, contract
+    - wallet:   wallet analysis, labels, behavior
+    - token:    token discovery, supply, metadata
+    - rag:      embeddings, search, ingest, firehose, feedback, agentic
+    - x402:     facilitator, settlement, enforcement
+    - intel:    feeds, narratives, graph
+    - scam:     classifier, patterns
+    - databus:  client, chain registry (96 chains)
+    - bulletin: user-generated content board
+
+Migration order: alerts (smoke test) → wallet → token → scanner → x402 → rag.
+"""

backend/app/domain/alerts/__init__.py

@@ -0,0 +1,32 @@
+"""Alerts domain — token alert subscriptions and event broadcasting.
+
+Public API:
+    from app.domain.alerts import (
+        AlertSubscription, AlertEvent, CreateAlertRequest, AlertType,
+        AlertService, AlertRepository, AlertBroadcaster,
+    )
+
+The domain layer is pure Python. NO FastAPI imports. NO HTTP concerns.
+This module is the only thing the api/ layer should import from.
+"""
+from __future__ import annotations
+
+from app.domain.alerts.broadcaster import AlertBroadcaster
+from app.domain.alerts.models import (
+    AlertEvent,
+    AlertSubscription,
+    AlertType,
+    CreateAlertRequest,
+)
+from app.domain.alerts.repository import AlertRepository
+from app.domain.alerts.service import AlertService
+
+__all__ = [
+    "AlertEvent",
+    "AlertSubscription",
+    "AlertType",
+    "CreateAlertRequest",
+    "AlertRepository",
+    "AlertService",
+    "AlertBroadcaster",
+]

backend/app/domain/alerts/broadcaster.py

@@ -0,0 +1,27 @@
+"""Broadcaster — pushes fired alert events to WebSocket subscribers.
+
+Uses core/websocket's broadcast helper. No FastAPI imports.
+"""
+from __future__ import annotations
+
+from app.core.logging import get_logger
+from app.core.websocket import broadcast_alert
+from app.domain.alerts.models import AlertEvent
+
+log = get_logger(__name__)
+
+
+class AlertBroadcaster:
+    """Thin wrapper that knows how to push an AlertEvent to live clients."""
+
+    async def broadcast_event(self, event: AlertEvent) -> None:
+        """Broadcast a single event to all connected WebSocket subscribers."""
+        try:
+            await broadcast_alert(event.model_dump(mode="json"))
+        except Exception as e:
+            # Broadcasting failure must not break the firing path.
+            log.warning(
+                "alert_broadcast_failed",
+                error=str(e),
+                alert_id=event.subscription_id,
+            )

backend/app/domain/alerts/models.py

@@ -0,0 +1,71 @@
+"""Pydantic v2 models for the alerts domain.
+
+No FastAPI imports. Pure data shapes.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class AlertType(str, Enum):
+    """Valid alert trigger types. Matches the legacy alert_types enum."""
+
+    LIQUIDITY_REMOVE = "liquidity_remove"
+    MINT = "mint"
+    BLACKLIST = "blacklist"
+    HONEYPOT = "honeypot"
+    RUG_PULL = "rug_pull"
+    WHALE_MOVEMENT = "whale_movement"
+    DEPLOYER_SELL = "deployer_sell"
+
+    @classmethod
+    def values(cls) -> list[str]:
+        return [m.value for m in cls]
+
+
+class AlertSubscription(BaseModel):
+    """A token alert subscription record (what we store in Redis)."""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    id: str = Field(..., description="Unique subscription id, e.g. alert:1700000000")
+    token_address: str = Field(..., min_length=1, max_length=256)
+    alert_types: list[AlertType] = Field(default_factory=lambda: [AlertType.LIQUIDITY_REMOVE, AlertType.MINT, AlertType.BLACKLIST])
+    webhook_url: str | None = Field(default=None, max_length=2048)
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    active: bool = True
+    owner_id: str | None = Field(default=None, description="User id of subscription owner, None = anonymous")
+
+    @field_validator("alert_types")
+    @classmethod
+    def _at_least_one_type(cls, v: list[AlertType]) -> list[AlertType]:
+        if not v:
+            raise ValueError("at least one alert_type required")
+        return v
+
+
+class CreateAlertRequest(BaseModel):
+    """What the API accepts to create a subscription."""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    token_address: str = Field(..., min_length=1, max_length=256)
+    alert_types: list[AlertType] | None = None
+    webhook_url: str | None = Field(default=None, max_length=2048)
+
+
+class AlertEvent(BaseModel):
+    """A fired alert, ready to broadcast to subscribers + WebSocket clients."""
+
+    model_config = ConfigDict(use_enum_values=True)
+
+    subscription_id: str
+    token_address: str
+    event_type: AlertType
+    payload: dict[str, Any] = Field(default_factory=dict)
+    fired_at: datetime = Field(default_factory=datetime.utcnow)
+    severity: str = Field(default="info", description="info | warning | critical")

backend/app/domain/alerts/repository.py

@@ -0,0 +1,60 @@
+"""Repository — async Redis storage for alert subscriptions.
+
+Storage layout (matches legacy for cutover):
+    Hash key: "rmi:alerts"
+    Field:     alert id (e.g. "alert:1700000000")
+    Value:     JSON-encoded AlertSubscription
+
+This is a thin async wrapper. No business logic. Pure data access.
+"""
+from __future__ import annotations
+
+from app.core.logging import get_logger
+from app.core.redis import get_redis_async
+from app.domain.alerts.models import AlertSubscription
+
+log = get_logger(__name__)
+
+HASH_KEY = "rmi:alerts"
+
+
+class AlertRepository:
+    """Async Redis-backed subscription storage."""
+
+    def __init__(self) -> None:
+        self._key = HASH_KEY
+
+    async def list_all(self) -> list[AlertSubscription]:
+        r = get_redis_async()
+        raw: dict[str, str] = await r.hgetall(self._key) or {}
+        out: list[AlertSubscription] = []
+        for value in raw.values():
+            try:
+                out.append(AlertSubscription.model_validate_json(value))
+            except Exception as e:
+                log.warning("alert_repo_corrupt_record", error=str(e))
+        return out
+
+    async def get(self, alert_id: str) -> AlertSubscription | None:
+        r = get_redis_async()
+        raw: str | None = await r.hget(self._key, alert_id)
+        if raw is None:
+            return None
+        try:
+            return AlertSubscription.model_validate_json(raw)
+        except Exception as e:
+            log.warning("alert_repo_corrupt_record", alert_id=alert_id, error=str(e))
+            return None
+
+    async def create(self, sub: AlertSubscription) -> None:
+        r = get_redis_async()
+        await r.hset(self._key, sub.id, sub.model_dump_json())
+
+    async def delete(self, alert_id: str) -> bool:
+        r = get_redis_async()
+        removed: int = await r.hdel(self._key, alert_id)
+        return removed > 0
+
+    async def list_by_token(self, token_address: str) -> list[AlertSubscription]:
+        all_subs = await self.list_all()
+        return [s for s in all_subs if s.token_address == token_address]

backend/app/domain/alerts/service.py

@@ -0,0 +1,133 @@
+"""Service — business logic for alert subscriptions and event firing.
+
+This is the only thing the api/ layer should call. It composes
+the repository (storage) and the broadcaster (WebSocket push).
+No FastAPI. No HTTP. Pure Python.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from app.core.logging import get_logger
+from app.domain.alerts.broadcaster import AlertBroadcaster
+from app.domain.alerts.models import (
+    AlertEvent,
+    AlertSubscription,
+    AlertType,
+    CreateAlertRequest,
+)
+from app.domain.alerts.repository import AlertRepository
+
+log = get_logger(__name__)
+
+
+class AlertService:
+    """Orchestrates subscription CRUD and event firing."""
+
+    def __init__(
+        self,
+        repo: AlertRepository | None = None,
+        broadcaster: AlertBroadcaster | None = None,
+    ) -> None:
+        self._repo = repo or AlertRepository()
+        self._broadcaster = broadcaster or AlertBroadcaster()
+
+    async def create_subscription(
+        self,
+        req: CreateAlertRequest,
+        owner_id: str | None = None,
+    ) -> AlertSubscription:
+        """Create a new subscription, persist it, return the saved record."""
+        # Use provided types, or the model default if not given.
+        types = req.alert_types or [
+            AlertType.LIQUIDITY_REMOVE,
+            AlertType.MINT,
+            AlertType.BLACKLIST,
+        ]
+        sub = AlertSubscription(
+            id=f"alert:{int(datetime.utcnow().timestamp())}",
+            token_address=req.token_address,
+            alert_types=types,
+            webhook_url=req.webhook_url,
+            owner_id=owner_id,
+        )
+        await self._repo.create(sub)
+        log.info(
+            "alert_subscription_created",
+            alert_id=sub.id,
+            token=sub.token_address,
+            types=[t.value for t in sub.alert_types],
+            owner=owner_id,
+        )
+        return sub
+
+    async def list_subscriptions(self, owner_id: str | None = None) -> list[AlertSubscription]:
+        """List subscriptions. If owner_id given, filter to that user."""
+        all_subs = await self._repo.list_all()
+        if owner_id is None:
+            return all_subs
+        return [s for s in all_subs if s.owner_id == owner_id]
+
+    async def cancel_subscription(self, alert_id: str, owner_id: str | None = None) -> bool:
+        """Cancel (delete) a subscription. Returns True if removed."""
+        sub = await self._repo.get(alert_id)
+        if sub is None:
+            return False
+        if owner_id is not None and sub.owner_id != owner_id:
+            log.warning(
+                "alert_cancel_unauthorized",
+                alert_id=alert_id,
+                owner=owner_id,
+                actual_owner=sub.owner_id,
+            )
+            return False
+        return await self._repo.delete(alert_id)
+
+    async def fire_event(
+        self,
+        token_address: str,
+        event_type: AlertType,
+        payload: dict | None = None,
+    ) -> list[AlertEvent]:
+        """Fire an event for a token. Returns the events that were broadcast.
+
+        Looks up all active subscriptions matching this token + event_type,
+        creates an AlertEvent for each, and broadcasts via the broadcaster.
+        """
+        subs = await self._repo.list_by_token(token_address)
+        matching = [
+            s for s in subs
+            if s.active and (event_type.value in [t.value for t in s.alert_types])
+        ]
+        if not matching:
+            return []
+
+        events: list[AlertEvent] = []
+        for sub in matching:
+            ev = AlertEvent(
+                subscription_id=sub.id,
+                token_address=token_address,
+                event_type=event_type,
+                payload=payload or {},
+                severity=_severity_for(event_type),
+            )
+            events.append(ev)
+            await self._broadcaster.broadcast_event(ev)
+        log.info(
+            "alert_events_fired",
+            token=token_address,
+            event_type=event_type.value,
+            count=len(events),
+        )
+        return events
+
+
+def _severity_for(event_type: AlertType) -> str:
+    """Map event type → severity (used by clients to choose alert UI)."""
+    critical = {AlertType.RUG_PULL, AlertType.HONEYPOT, AlertType.BLACKLIST}
+    warning = {AlertType.LIQUIDITY_REMOVE, AlertType.WHALE_MOVEMENT, AlertType.DEPLOYER_SELL}
+    if event_type in critical:
+        return "critical"
+    if event_type in warning:
+        return "warning"
+    return "info"

backend/tests/unit/domain/alerts/test_service.py

@@ -0,0 +1,129 @@
+"""Unit tests for the alerts domain.
+
+Pure-Python tests. No FastAPI. No HTTP. Imports the domain directly.
+This is the test that proves the architecture: domain code is
+unit-testable without spinning up the API.
+"""
+from __future__ import annotations
+
+import json
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.domain.alerts import (
+    AlertBroadcaster,
+    AlertEvent,
+    AlertRepository,
+    AlertService,
+    AlertSubscription,
+    AlertType,
+    CreateAlertRequest,
+)
+
+
+@pytest.fixture
+def fake_repo() -> AsyncMock:
+    repo = AsyncMock(spec=AlertRepository)
+    repo.list_all = AsyncMock(return_value=[])
+    repo.get = AsyncMock(return_value=None)
+    repo.create = AsyncMock(return_value=None)
+    repo.delete = AsyncMock(return_value=True)
+    repo.list_by_token = AsyncMock(return_value=[])
+    return repo
+
+
+@pytest.fixture
+def fake_broadcaster() -> AsyncMock:
+    bc = AsyncMock(spec=AlertBroadcaster)
+    bc.broadcast_event = AsyncMock(return_value=None)
+    return bc
+
+
+@pytest.fixture
+def service(fake_repo: AsyncMock, fake_broadcaster: AsyncMock) -> AlertService:
+    return AlertService(repo=fake_repo, broadcaster=fake_broadcaster)
+
+
+async def test_create_subscription_assigns_id_and_persists(service, fake_repo):
+    req = CreateAlertRequest(token_address="0xabc")
+    sub = await service.create_subscription(req, owner_id="user-1")
+    assert sub.id.startswith("alert:")
+    assert sub.token_address == "0xabc"
+    assert sub.owner_id == "user-1"
+    assert sub.active is True
+    # Default alert types applied when none provided
+    assert AlertType.LIQUIDITY_REMOVE in [AlertType(t) for t in sub.alert_types]
+    fake_repo.create.assert_awaited_once()
+
+
+async def test_create_subscription_with_explicit_types(service, fake_repo):
+    req = CreateAlertRequest(
+        token_address="0xdef",
+        alert_types=[AlertType.HONEYPOT, AlertType.RUG_PULL],
+    )
+    sub = await service.create_subscription(req)
+    assert [AlertType(t) for t in sub.alert_types] == [AlertType.HONEYPOT, AlertType.RUG_PULL]
+    fake_repo.create.assert_awaited_once()
+
+
+async def test_list_filters_by_owner(service, fake_repo):
+    fake_repo.list_all = AsyncMock(return_value=[
+        AlertSubscription(id="alert:1", token_address="0xa", owner_id="user-1"),
+        AlertSubscription(id="alert:2", token_address="0xb", owner_id="user-2"),
+        AlertSubscription(id="alert:3", token_address="0xc", owner_id=None),
+    ])
+    result = await service.list_subscriptions(owner_id="user-1")
+    assert len(result) == 1
+    assert result[0].id == "alert:1"
+
+
+async def test_cancel_unauthorized_returns_false(service, fake_repo):
+    fake_repo.get = AsyncMock(return_value=AlertSubscription(
+        id="alert:1", token_address="0xa", owner_id="user-1",
+    ))
+    ok = await service.cancel_subscription("alert:1", owner_id="user-2")
+    assert ok is False
+    fake_repo.delete.assert_not_awaited()
+
+
+async def test_cancel_owner_succeeds(service, fake_repo):
+    fake_repo.get = AsyncMock(return_value=AlertSubscription(
+        id="alert:1", token_address="0xa", owner_id="user-1",
+    ))
+    fake_repo.delete = AsyncMock(return_value=True)
+    ok = await service.cancel_subscription("alert:1", owner_id="user-1")
+    assert ok is True
+    fake_repo.delete.assert_awaited_once_with("alert:1")
+
+
+async def test_fire_event_broadcasts_to_matching_subs(service, fake_repo, fake_broadcaster):
+    fake_repo.list_by_token = AsyncMock(return_value=[
+        AlertSubscription(
+            id="alert:1", token_address="0xa",
+            alert_types=[AlertType.HONEYPOT, AlertType.RUG_PULL],
+            active=True,
+        ),
+        AlertSubscription(
+            id="alert:2", token_address="0xa",
+            alert_types=[AlertType.WHALE_MOVEMENT],  # doesn't match
+            active=True,
+        ),
+        AlertSubscription(
+            id="alert:3", token_address="0xa",
+            alert_types=[AlertType.HONEYPOT],
+            active=False,  # inactive
+        ),
+    ])
+    events = await service.fire_event("0xa", AlertType.HONEYPOT, {"score": 0.9})
+    assert len(events) == 1
+    assert events[0].subscription_id == "alert:1"
+    assert events[0].severity == "critical"  # HONEYPOT is critical
+    fake_broadcaster.broadcast_event.assert_awaited_once()
+
+
+async def test_severity_mapping():
+    from app.domain.alerts.service import _severity_for
+    assert _severity_for(AlertType.RUG_PULL) == "critical"
+    assert _severity_for(AlertType.LIQUIDITY_REMOVE) == "warning"
+    assert _severity_for(AlertType.MINT) == "info"