.github/workflows/ci.yml

@@ -12,8 +12,8 @@ on:
 #
 # Phase 1 of AUDIT-2026-Q3.md item P1.6:
 # - CI was 6/8 decorative (|| true / continue-on-error: true on every job)
-# - Now: 3 GATING jobs (build + test + typecheck-gate) must pass for merge
-# - 6 INFORMATIONAL jobs (lint-info, typecheck-full-info, etc.) report but never gate
+# - Now: 2 GATING jobs (build + test) must pass for merge
+# - 6 INFORMATIONAL jobs (lint-info, typecheck-info, etc.) report but never gate
 # - Forgejo .forgejo/workflows/ci.yml remains the authoritative gate
 
 jobs:
@@ -61,21 +61,6 @@ jobs:
             2>&1 | tail -50
         # NO || true — failure GATES merge
 
-  typecheck-gate:
-    name: Typecheck scoped gate (gate)
-    needs: build
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: astral-sh/setup-uv@v2
-      - uses: actions/setup-python@v5
-        with:
-          python-version: "3.11"
-      - name: Install project editable
-        run: uv pip install --system -e ".[dev]"
-      - name: Run scoped mypy gate
-        run: make mypy-gate
-
   # ────────────────────────── INFORMATIONAL JOBS (fire-and-forget) ──────────────────────────
 
   lint-info:
@@ -94,8 +79,8 @@ jobs:
       - name: Run ruff lint (informational)
         run: ruff check . --statistics --output-format=concise 2>&1 | tail -30 || true
 
-  typecheck-full-info:
-    name: Typecheck full codebase (info only)
+  typecheck-info:
+    name: Typecheck (info only)
     if: always()
     runs-on: ubuntu-latest
     continue-on-error: true
@@ -108,7 +93,7 @@ jobs:
       - name: Install mypy
         run: uv pip install --system mypy
       - name: Run mypy typecheck (informational)
-        run: make typecheck || true
+        run: mypy app/ --config-file mypy.ini || true
 
   security-info:
     name: Security (info only)

ARCHITECTURE.md

@@ -1,11 +1,10 @@
 # RMI Backend — Architecture
 
 **Status:** canonical
-**Last updated:** 2026-07-07
+**Last updated:** 2026-07-03
 
 ## Overview
 FastAPI monolith serving crypto risk intelligence via HTTP, Telegram, and MCP.
-Domains are consolidated under `app/domains/` per AUDIT-2026-Q3 Phase 4.
 
 ## High-level flow
 ```
@@ -13,9 +12,9 @@ Client (Web / Telegram / AI Agent)
     ↓
 nginx → rmi-backend container (:8000)
     ↓
-FastAPI routers (mounted by app/mount.py)
+FastAPI routers
     ↓
-app/domains/<domain>/service.py  →  app/infra/ + app/domains/<domain>/repository.py
+Services / scanners / DataBus providers
     ↓
 Postgres / Redis / Qdrant / Neo4j / ClickHouse
 ```
@@ -23,42 +22,35 @@ Postgres / Redis / Qdrant / Neo4j / ClickHouse
 ## Components
 | Path | Purpose |
 |---|---|
-| `app/factory.py` | FastAPI app factory (lifespan, middleware, error handlers, routers) |
-| `app/mount.py` | Single source of truth for router mounting |
-| `app/core/` | Cross-cutting concerns (logging, errors, redis, http, auth, config) |
-| `app/api/v1/` | Thin HTTP transport layer (public, auth, admin, x402, mcp) |
-| `app/domains/` | Business logic domains: auth, billing, databus, news, reports, scanner, scanners, telegram, threat, token, tokens, wallet, x402 |
-| `app/domains/scanners/` | Token risk scanners (IP — will move to rmi-ip in Phase 6) |
-| `app/domains/databus/` | 30+ third-party data provider integrations |
-| `app/domains/telegram/rugmunchbot/` | Telegram bot command handlers |
+| `app/routers/` | HTTP route handlers (thin) |
+| `app/scanners/` | Token risk scanners |
+| `app/databus/` | 30+ third-party data providers |
+| `app/telegram_bot/` | Telegram bot command handlers |
 | `app/mcp/` | MCP manifest and tool manager |
-| `app/infra/` | External integrations (ollama, langfuse, vector stores, chain clients) |
-| `app/wallet_memory/` | Wallet label + clustering subsystem (IP — will move to rmi-ip in Phase 6) |
-| `app/routers/` | Legacy HTTP route handlers (being migrated to app/domains/ or app/api/v1/) |
+| `app/facilitators/` | x402 payment verification |
+| `app/wallet_memory/` | Wallet label + clustering subsystem |
 
 ## Storage
 | Store | Use | Container / Host |
 |---|---|---|
 | Postgres | Primary OLTP, Ponder indexer data | `rmi-postgres` |
 | Redis | Cache, Celery queue | `rmi-redis` |
-| Qdrant | Vector embeddings | `rmi-qdrant` |
+| Qdrant | Vector embeddings | `rmi-qdrant` (down) |
 | Neo4j | Wallet graph / labels | `rmi-neo4j` |
 | ClickHouse | Analytics, 39M address labels | `rmi-clickhouse` |
 
 ## Interfaces
 | Endpoint | Purpose |
 |---|---|
-| `GET /health` | Health check (degraded — ETH RPC endpoint missing) |
-| `GET /metrics` | Prometheus metrics |
+| `GET /health` | Health check (currently unhealthy due to Redis) |
 | `GET /docs` | Swagger/OpenAPI docs |
 | `POST /scanner/scan` | Live token scan |
 | `POST /mcp/...` | MCP tool endpoints |
 
 ## Current issues
-- `app/routers/x402_tools.py` (212 KB) and `app/routers/x402_enforcement.py` (108 KB) violate the 500-line limit and should become separate services or split by domain.
+- `x402_tools.py` (212 KB) and `x402_enforcement.py` (108 KB) violate the 500-line limit and should become separate services or split by domain.
 - `sqlite3` is used in 5 files for local state; should migrate to Postgres/aiosqlite.
 - F-string logs and `print()` statements prevent structured log aggregation.
-- `mypy` has a structural issue with `app/domains/auth/` that needs resolution.
 
 ## Deployment
 ```bash

Makefile

@@ -1,4 +1,4 @@
-.PHONY: help install dev build lint format check test test-all security ci clean mypy-gate typecheck-gate
+.PHONY: help install dev build lint format check test typecheck security ci clean
 
 help: ## Show this help
 	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-20s\033[0m %s\n", $$1, $$2}'
@@ -27,14 +27,8 @@ check: ## Full check: lint + format + typecheck + test
 	mypy app/ --ignore-missing-imports
 	pytest tests/ -x -q -m "not integration"
 
-typecheck: ## MyPy type check (full codebase, informational)
-	mypy app/ --config-file mypy.ini
-
-mypy-gate: ## MyPy gate for clean modules (authoritative)
-	mypy app/domains/auth/ app/core/redis.py --config-file mypy-gate.ini
-
-typecheck-gate: ## Alias for mypy-gate
-	$(MAKE) mypy-gate
+typecheck: ## MyPy type check
+	mypy app/ --ignore-missing-imports
 
 test: ## Run tests (non-integration)
 	pytest tests/ -x -q -m "not integration"
@@ -49,7 +43,7 @@ security: ## Security audit
 ci: ## Full CI pipeline
 	ruff check app/ tests/
 	ruff format --check app/ tests/
-	$(MAKE) mypy-gate
+	mypy app/ --ignore-missing-imports
 	pytest tests/ -x -q -m "not integration"
 
 clean: ## Remove build artifacts

STATUS.md

@@ -1,7 +1,7 @@
 # STATUS -- rmi-backend
 
 **Owner:** Rug Munch Media LLC Engineering
-**Last updated:** 2026-07-07 -- Phase 4 domain consolidation in progress; P4.1-P4.8 + P5.1 complete.
+**Last updated:** 2026-07-06 -- Phase 1 of [AUDIT-2026-Q3.md](AUDIT-2026-Q3.md) complete.
 
 ## What This Repo Is
 
@@ -13,7 +13,7 @@ Stack: Python 3.12 / FastAPI / SQLAlchemy / async / Playwright / Pydantic v2 / R
 
 ## Current Status
 
-**Phase 4 (domain consolidation) is active.** Foundation (Phase 1) is solid, dead-router wiring (Phase 2) is partial, god-file splitting (Phase 3) is deferred, and standardization (Phase 5) has begun with CI trustworthiness and mypy config cleanup.
+**Phase 1 (development hygiene) complete.** Phase 2 (delete dead code) starts next per [AUDIT-2026-Q3.md](AUDIT-2026-Q3.md).
 
 ### Phase 1 -- DONE
 
@@ -30,44 +30,16 @@ Committed to `main`:
 - `92a01ff` + `5294983` -- `AppError` hierarchy in `app/core/errors.py`, wired through `error_handlers.py`
 - `cd02714` + `2fb571e` -- author amend (canonical identity: `cryptorugmunch <admin@rugmunch.io>`)
 
-### Phase 2 -- IN PROGRESS (partial)
-
-Rather than deleting 148 routers, we validated and mounted the useful ones in `app/mount.py` Wave 1-3 (2026-07-07). Dead routers remain in tree and will be archived in a later pass.
-
-### Phase 3 -- NOT STARTED
-
-God-files (`x402_tools.py`, `x402_enforcement.py`, `auth.py`, `databus/providers.py`, `telegram_bot/bot.py`, `wallet_manager_v2.py`) still await splitting.
-
-### Phase 4 -- IN PROGRESS
-
-Domain consolidation (AUDIT-2026-Q3 P4.x) — committed to `main`:
-- `dca458e` -- P4.1 billing (`app/billing/` + `app/facilitators/` → `app/domains/billing/`)
-- `7109a16` -- P4.2 wallet (`app/wallet/` → `app/domains/wallet/`)
-- `948e41c` -- P4.3 auth (`app/auth/` → `app/domains/auth/`)
-- `cab9740` -- P4.4 tokens (`app/tokens/` → `app/domains/tokens/`)
-- `ed5b830` -- P4.5 databus (`providers/` + `_generated/` → `app/domains/databus/`)
-- `56075cf` -- P4.6 telegram (`rugmunchbot/` → `app/domains/telegram/rugmunchbot/`)
-- `3b7ef42` -- P4.7 rename `app/domain/` → `app/domains/` + consolidate
-- `7cced4e` + `4686cb3` -- P4.8 scanners (`app/scanners/` → `app/domains/scanners/`)
-
-Remaining Phase 4:
-- Finish databus root migration (`app/databus/` engine → `app/domains/databus/`)
-- Finish telegram migration (`app/telegram_bot/` → `app/domains/telegram/`)
-- Clean up deprecated shim dirs (`app/auth/`, `app/wallet/`, `app/billing/`, `app/facilitators/`, `app/scanners/`, `app/domain/`)
-- Consolidate bulletin, intelligence, markets, admin, referral, MCP domains
-
-### Phase 5 -- STARTED
-
-- `d666ad2` -- P5.1 `HEALTH_CHECK_DURATION` + `test_factory_has_minimum_routes`
-- `4686cb3` -- mypy config fixes (`mypy.ini` regex, `disallow_any_explicit` typo, `app.domains.*` module pattern)
-- Still open: alembic, real coverage gate, structured logging, Sentry wiring, Prometheus middleware
-
-### What Phase 1-4 DID NOT do (deferred)
+### What Phase 1 DID NOT do (deferred)
 
 - ~2K ruff warnings in legacy code (Phase 5)
-- Real coverage 8.17% -> 80% gate (Phase 5)
+- Real coverage 8.17% -> 80% gate (Phase 5, slowest part)
+- `mypy.ini` line 8 parse error + `disallow_any_express_imports` typo (Phase 5)
 - Alembic migrations (still MISSING; Phase 5)
+- 285 dead modules (Phase 2)
+- 148 unmounted routers (Phase 2)
 - God-files split (Phase 3)
+- Domain consolidation (Phase 4)
 
 ### Open Issues
 
@@ -96,11 +68,8 @@ e404e90 feat(rmi-backend,audit): add app/main.py entrypoint delegating to factor
 - `gitleaks detect` reports 28 findings -- all **public ERC-20/Solana token contract addresses** (USDC, WETH, MATIC, BONK, DAI, plus cryptoscam-address dataset). NOT secrets. Suppressed via `.gitleaksignore` + `.gitleaks.toml` `[allowlist]`. No rotation needed.
 - All secrets live in gopass under `rmi/contacts/admin_recovery` etc.
 
-## Next: Phase 4 completion
+## Next: Phase 2 (week 2)
 
-1. Finish `app/databus/` engine migration to `app/domains/databus/`.
-2. Finish `app/telegram_bot/` migration to `app/domains/telegram/rugmunchbot/`.
-3. Remove deprecated shim directories (`app/auth/`, `app/wallet/`, `app/billing/`, `app/facilitators/`, `app/scanners/`, `app/domain/`).
-4. Consolidate bulletin, intelligence, markets, admin, referral, MCP into `app/domains/`.
+Archive **148 unmounted routers** + **~140 dead `app/*.py` files** to `app/_archive/legacy_2026_07/`. Goal: ~30% LOC reduction without changing any current behavior.
 
 See [AUDIT-2026-Q3.md](AUDIT-2026-Q3.md) for the full 6-phase plan.

app/_archive/legacy_2026_07/mcp_server.py

@@ -46,7 +46,7 @@ def _get_tools() -> dict[str, Any]:
 
 def _get_facilitator_count() -> int:
     try:
-        from app.domains.billing.facilitators.base import get_registry
+        from app.facilitators.base import get_registry
 
         return len(get_registry().get_all())
     except Exception:
@@ -57,7 +57,7 @@ def _desc(tool_id: str, pricing: dict) -> str:
     d = pricing.get("description", "")
     if d and d != tool_id and len(d) > 10:
         return d
-    fallbacks = {
+    FALLBACKS = {
         "airdrop_check": "Verify airdrop legitimacy -- contract audit, distribution analysis, scam pattern detection. Know if an airdrop is real or a wallet drainer before connecting.",
         "airdrop_finder": "Discover active and upcoming airdrops across all major chains. Eligibility checks, value estimation, claim deadlines, and Sybil detection.",
         "all_in_one": "All-in-One Audit -- comprehensive security scan: rug pull, honeypot, clone detection, contract audit, and ownership analysis in a single call.",

app/ai_router.py

@@ -1,288 +1,72 @@
 #!/usr/bin/env python3
 """
-AI Router — service functions for chat completions.
-
-Primary: Ollama Cloud (deepseek-v4-flash, free).
-Fallback: OpenRouter (various models via OPENROUTER_API_KEY).
-
-Both endpoints are OpenAI-compatible /v1/chat/completions.
-Called by app/routers/chat.py and app/routers/wallet_clustering_router.py
-as ``ai_router.chat_completion(...)`` and
-``ai_router.stream_chat_completion(...)``.
+Stub AI Router - Intelligent Model-First Provider Swapping
+===========================================================
+Routes requests to optimal AI provider based on quota, latency, and cost.
+For now, a minimal stub that delegates to OpenRouter.
 """
 
 import base64
-import json
-import logging
 import os
-import time
-from collections.abc import AsyncGenerator
 from typing import Any
 
-import httpx
 from fastapi import APIRouter
 
-logger = logging.getLogger("rmi.ai_router")
-
 router = APIRouter(tags=["AI Router"])
 
 # Decode base64 LLM key if present, otherwise use plain LLM_API_KEY
+# (safety net: ensures key is decoded even when imported without main.py)
 if os.getenv("LLM_API_KEY_B64"):
     os.environ["LLM_API_KEY"] = base64.b64decode(os.getenv("LLM_API_KEY_B64")).decode()
 
-# ── Provider config ─────────────────────────────────────────────
-OLLAMA_KEY = os.getenv("OLLAMA_API_KEY", "")
-OLLAMA_URL = "https://ollama.com/v1/chat/completions"
-OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "deepseek-v4-flash")
-
-OPENROUTER_KEY = os.getenv("OPENROUTER_API_KEY", "")
-OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
-OPENROUTER_MODEL = os.getenv("OPENROUTER_MODEL", "deepseek/deepseek-chat")
-
-# Tier → model mapping (best effort; Ollama Cloud ignores non-model fields)
-TIER_MODELS: dict[str, str] = {
-    "T0": "deepseek-v4-pro",
-    "T1": "deepseek-v4-pro",
-    "T2": "deepseek-v4-flash",
-    "T3": "deepseek-v4-flash",
-    "T4": "deepseek-v4-flash",
-}
-
+# Model tiers (for reference, full config in ai_router.py)
 MODEL_TIERS = {
-    "T0": {"name": "Ultra", "models": ["deepseek-v4-pro"], "max_cost_per_1k": 0.0},
-    "T1": {"name": "Premium", "models": ["deepseek-v4-pro"], "max_cost_per_1k": 0.0},
-    "T2": {"name": "Standard", "models": ["deepseek-v4-flash"], "max_cost_per_1k": 0.0},
-    "T3": {"name": "Fast", "models": ["deepseek-v4-flash"], "max_cost_per_1k": 0.0},
-    "T4": {"name": "Free", "models": ["deepseek-v4-flash"], "max_cost_per_1k": 0.0},
+    "T0": {"name": "Ultra", "models": ["gpt-4o", "claude-3.5-sonnet"], "max_cost_per_1k": 0.05},
+    "T1": {"name": "Premium", "models": ["gpt-4-turbo", "claude-3-opus"], "max_cost_per_1k": 0.02},
+    "T2": {
+        "name": "Standard",
+        "models": ["gpt-3.5-turbo", "claude-3-haiku"],
+        "max_cost_per_1k": 0.005,
+    },
+    "T3": {"name": "Fast", "models": ["llama-3-8b", "mistral-tiny"], "max_cost_per_1k": 0.001},
+    "T4": {"name": "Free", "models": ["tiny-llama", "phi-2"], "max_cost_per_1k": 0.0},
 }
 
+# Providers (for reference)
 PROVIDERS = {
-    "ollama": {"url": OLLAMA_URL, "key_env": "OLLAMA_API_KEY", "model": OLLAMA_MODEL, "rpm": 100},
-    "openrouter": {"url": OPENROUTER_URL, "key_env": "OPENROUTER_API_KEY", "model": OPENROUTER_MODEL, "rpm": 100},
+    "deepseek": {
+        "url": os.getenv("LLM_BASE_URL", "https://api.deepseek.com/v1/chat/completions"),
+        "key_env": "LLM_API_KEY",
+        "model": os.getenv("LLM_MODEL", "deepseek-v4-flash"),
+        "rpm": 100,
+    },
+    "openrouter": {
+        "url": "https://openrouter.ai/api/v1/chat/completions",
+        "key_env": "OPENROUTER_API_KEY",
+        "rpm": 100,
+    },
 }
 
 
-def _resolve_model(model: str | None, tier: str | None) -> str:
-    """Pick the model: explicit model > tier mapping > default."""
-    if model:
-        return model
-    if tier and tier in TIER_MODELS:
-        return TIER_MODELS[tier]
-    return OLLAMA_MODEL
-
-
-async def chat_completion(
-    *,
-    messages: list[dict[str, str]],
-    model: str | None = None,
-    tier: str | None = None,
-    temperature: float = 0.3,
-    max_tokens: int = 500,
-    timeout: float = 30.0,
-) -> dict[str, Any]:
-    """Non-streaming chat completion.
-
-    Returns dict with keys: content, model, _provider, _latency_ms, usage, error (on failure).
-    Tries Ollama Cloud first, falls back to OpenRouter.
-    """
-    resolved = _resolve_model(model, tier)
-    start = time.time()
-
-    # Provider 1: Ollama Cloud
-    if OLLAMA_KEY:
-        try:
-            async with httpx.AsyncClient(timeout=timeout) as client:
-                resp = await client.post(
-                    OLLAMA_URL,
-                    headers={
-                        "Authorization": f"Bearer {OLLAMA_KEY}",
-                        "Content-Type": "application/json",
-                    },
-                    json={
-                        "model": resolved,
-                        "messages": messages,
-                        "temperature": temperature,
-                        "max_tokens": max_tokens,
-                        "stream": False,
-                    },
-                )
-                if resp.status_code == 200:
-                    data = resp.json()
-                    content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
-                    return {
-                        "content": content,
-                        "model": data.get("model", resolved),
-                        "_provider": "ollama",
-                        "_latency_ms": round((time.time() - start) * 1000, 1),
-                        "usage": data.get("usage", {}),
-                    }
-                logger.warning("ai_router ollama status=%s body=%s", resp.status_code, resp.text[:200])
-        except Exception as e:
-            logger.warning("ai_router ollama error: %s", e)
-
-    # Provider 2: OpenRouter fallback
-    if OPENROUTER_KEY:
-        try:
-            async with httpx.AsyncClient(timeout=timeout) as client:
-                resp = await client.post(
-                    OPENROUTER_URL,
-                    headers={
-                        "Authorization": f"Bearer {OPENROUTER_KEY}",
-                        "Content-Type": "application/json",
-                    },
-                    json={
-                        "model": resolved,
-                        "messages": messages,
-                        "temperature": temperature,
-                        "max_tokens": max_tokens,
-                        "stream": False,
-                    },
-                )
-                if resp.status_code == 200:
-                    data = resp.json()
-                    content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
-                    return {
-                        "content": content,
-                        "model": data.get("model", resolved),
-                        "_provider": "openrouter",
-                        "_latency_ms": round((time.time() - start) * 1000, 1),
-                        "usage": data.get("usage", {}),
-                    }
-                logger.warning("ai_router openrouter status=%s body=%s", resp.status_code, resp.text[:200])
-        except Exception as e:
-            logger.warning("ai_router openrouter error: %s", e)
-
-    return {
-        "content": "",
-        "model": resolved,
-        "_provider": "none",
-        "_latency_ms": round((time.time() - start) * 1000, 1),
-        "usage": {},
-        "error": "No AI provider available (set OLLAMA_API_KEY or OPENROUTER_API_KEY)",
-    }
-
-
-async def stream_chat_completion(
-    *,
-    messages: list[dict[str, str]],
-    model: str | None = None,
-    tier: str | None = None,
-    temperature: float = 0.3,
-    max_tokens: int = 500,
-    timeout: float = 30.0,
-) -> AsyncGenerator[str, None]:
-    """Streaming chat completion via SSE.
-
-    Yields content tokens as strings. On error, yields ``[ERROR: ...]``.
-    Tries Ollama Cloud first, falls back to OpenRouter.
-    """
-    resolved = _resolve_model(model, tier)
-
-    # Provider 1: Ollama Cloud (streaming)
-    if OLLAMA_KEY:
-        try:
-            async with httpx.AsyncClient(timeout=timeout) as client, client.stream(
-                "POST",
-                OLLAMA_URL,
-                headers={
-                    "Authorization": f"Bearer {OLLAMA_KEY}",
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "model": resolved,
-                    "messages": messages,
-                    "temperature": temperature,
-                    "max_tokens": max_tokens,
-                    "stream": True,
-                },
-            ) as resp:
-                if resp.status_code == 200:
-                    async for line in resp.aiter_lines():
-                        if line.startswith("data: ") and line != "data: [DONE]":
-                            try:
-                                chunk = json.loads(line[6:])
-                                delta = chunk.get("choices", [{}])[0].get("delta", {})
-                                content = delta.get("content", "")
-                                if content:
-                                    yield content
-                            except json.JSONDecodeError:
-                                pass
-                    return
-                logger.warning("ai_router ollama stream status=%s", resp.status_code)
-        except Exception as e:
-            logger.warning("ai_router ollama stream error: %s", e)
-
-    # Provider 2: OpenRouter streaming fallback
-    if OPENROUTER_KEY:
-        try:
-            async with httpx.AsyncClient(timeout=timeout) as client, client.stream(
-                "POST",
-                OPENROUTER_URL,
-                headers={
-                    "Authorization": f"Bearer {OPENROUTER_KEY}",
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "model": resolved,
-                    "messages": messages,
-                    "temperature": temperature,
-                    "max_tokens": max_tokens,
-                    "stream": True,
-                },
-            ) as resp:
-                if resp.status_code == 200:
-                    async for line in resp.aiter_lines():
-                        if line.startswith("data: ") and line != "data: [DONE]":
-                            try:
-                                chunk = json.loads(line[6:])
-                                delta = chunk.get("choices", [{}])[0].get("delta", {})
-                                content = delta.get("content", "")
-                                if content:
-                                    yield content
-                            except json.JSONDecodeError:
-                                pass
-                    return
-                logger.warning("ai_router openrouter stream status=%s", resp.status_code)
-        except Exception as e:
-            logger.warning("ai_router openrouter stream error: %s", e)
-
-    yield "[ERROR: No AI provider available (set OLLAMA_API_KEY or OPENROUTER_API_KEY)]"
-
-
-# ── HTTP endpoints (kept for compat, now backed by real service) ──
-
-
 @router.post("/ai/completions")
-async def ai_completions(request: dict[str, Any]) -> dict[str, Any]:
+async def ai_completions(request: dict[str, Any]):
     """AI completion via optimal provider routing."""
-    return await chat_completion(
-        messages=request.get("messages", []),
-        model=request.get("model"),
-        tier=request.get("tier"),
-        temperature=request.get("temperature", 0.3),
-        max_tokens=request.get("max_tokens", 500),
-    )
+    return {"error": "AI Router not fully configured", "provider": "openrouter"}
 
 
 @router.post("/ai/chat")
-async def ai_chat(request: dict[str, Any]) -> dict[str, Any]:
+async def ai_chat(request: dict[str, Any]):
     """AI chat endpoint with provider fallback."""
-    return await chat_completion(
-        messages=request.get("messages", []),
-        model=request.get("model"),
-        tier=request.get("tier"),
-        temperature=request.get("temperature", 0.3),
-        max_tokens=request.get("max_tokens", 500),
-    )
+    return {"error": "AI Router not fully configured", "provider": "openrouter"}
 
 
 @router.get("/ai/providers")
-async def list_providers() -> dict[str, Any]:
+async def list_providers():
     """List available AI providers."""
     return {"providers": list(PROVIDERS.keys())}
 
 
 @router.get("/ai/models")
-async def list_models() -> dict[str, Any]:
+async def list_models():
     """List available models by tier."""
     return {"tiers": MODEL_TIERS}

app/alert_pipeline.py

@@ -1,8 +1,370 @@
-"""Deprecated shim - re-exports app.domains.intelligence.alert_pipeline.
-
-Migrate imports from `app.alert_pipeline` to `app.domains.intelligence.alert_pipeline`.
-This shim will be removed in Phase 3 cleanup.
 """
-from __future__ import annotations
+RMI Alert Pipeline - Real-time threat intelligence from scanners.
+=================================================================
+Feeds the Live Intel panel, WebSocket streams, and alert endpoints.
 
-from app.domains.intelligence.alert_pipeline import *  # noqa: F403
+Data sources:
+  - SENTINEL scanner (risk scores, scam detection)
+  - GoPlus security API (honeypot, tax, proxy checks)
+  - DexScreener new pairs (fresh launches to scan)
+  - Our own RAG scam patterns
+  - Wallet label cross-references
+
+Alert flow:
+  Scanner → alert_pipeline.push_alert() → Redis sorted set + pub/sub
+  Homepage reads from /api/v1/alerts/recent
+  Sidebar reads from /api/v1/alerts/count
+  WebSocket streams to connected clients
+"""
+
+import json
+import logging
+import os
+import time
+from datetime import UTC, datetime
+
+logger = logging.getLogger("alert_pipeline")
+
+REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
+REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
+REDIS_PW = os.getenv("REDIS_PASSWORD", "")
+
+ALERT_KEY = "rmi:alerts:recent"
+ALERT_COUNT_KEY = "rmi:alerts:count:active"
+ALERT_MAX = 500  # Keep last 500 alerts
+
+
+async def _get_redis():
+    """Get async Redis connection."""
+    import redis.asyncio as aioredis
+
+    return aioredis.Redis(
+        host=REDIS_HOST,
+        port=REDIS_PORT,
+        password=REDIS_PW or None,
+        decode_responses=True,
+    )
+
+
+async def get_active_alert_count() -> int:
+    """Get count of active (unacknowledged) alerts."""
+    try:
+        r = await _get_redis()
+        count = await r.zcard(ALERT_KEY)
+        await r.close()
+        return count
+    except Exception as e:
+        logger.debug(f"Alert count error: {e}")
+        return 0
+
+
+async def push_alert(
+    alert_type: str,
+    title: str,
+    description: str = "",
+    severity: str = "high",
+    chain: str = "unknown",
+    token: str = "",
+    token_symbol: str = "",
+    wallet: str = "",
+    risk_score: int = 0,
+    metadata: dict | None = None,
+) -> str:
+    """
+    Push a new alert to the pipeline.
+    Returns alert_id.
+    """
+    alert_id = f"alt_{int(time.time())}_{os.urandom(4).hex()}"
+
+    alert = {
+        "id": alert_id,
+        "type": alert_type,
+        "title": title,
+        "description": description,
+        "severity": severity,
+        "chain": chain,
+        "token": token,
+        "token_symbol": token_symbol,
+        "wallet": wallet,
+        "risk_score": risk_score,
+        "acknowledged": False,
+        "created_at": datetime.now(UTC).isoformat(),
+        **(metadata or {}),
+    }
+
+    try:
+        r = await _get_redis()
+        score = time.time()
+        await r.zadd(ALERT_KEY, {json.dumps(alert): score})
+        # Trim old alerts
+        await r.zremrangebyrank(ALERT_KEY, 0, -(ALERT_MAX + 1))
+        # Pub/sub for WebSocket streaming
+        await r.publish("rmi:ws:alerts", json.dumps(alert))
+        await r.close()
+        logger.info(f"Alert pushed: {alert_type} | {title[:60]}")
+    except Exception as e:
+        logger.error(f"Failed to push alert: {e}")
+
+    return alert_id
+
+
+async def get_recent_alerts(limit: int = 20, severity: str = "") -> list[dict]:
+    """Get recent alerts with optional severity filter. Normalizes old/new formats."""
+    try:
+        r = await _get_redis()
+        raw = await r.zrevrange(ALERT_KEY, 0, limit * 2 - 1)
+        await r.close()
+
+        alerts = []
+        for entry in raw:
+            try:
+                alert = json.loads(entry)
+
+                # Normalize old alert format to new
+                if "title" not in alert:
+                    alert["title"] = alert.get("message", alert.get("event", "Unknown alert"))
+                if "description" not in alert:
+                    desc_parts = []
+                    if alert.get("symbol"):
+                        desc_parts.append(f"Token: {alert['symbol']}")
+                    if alert.get("risk_score"):
+                        desc_parts.append(f"Risk: {alert['risk_score']}/100")
+                    flags = alert.get("risk_flags", [])
+                    if flags:
+                        desc_parts.append("; ".join(str(f)[:80] for f in flags[:2]))
+                    alert["description"] = " | ".join(desc_parts)
+                if "severity" not in alert:
+                    score = alert.get("risk_score", 50)
+                    alert["severity"] = "critical" if score >= 85 else "high" if score >= 65 else "medium"
+                if "chain" not in alert:
+                    alert["chain"] = alert.get("chain", "unknown")
+
+                if severity and alert.get("severity") != severity:
+                    continue
+                alerts.append(alert)
+                if len(alerts) >= limit:
+                    break
+            except json.JSONDecodeError:
+                pass
+
+        return alerts
+    except Exception as e:
+        logger.error(f"get_recent_alerts error: {e}")
+        return []
+
+
+# ── Alert generators - called by cron jobs or on-demand ──────────
+
+
+async def scan_solana_new_pairs(limit: int = 5) -> int:
+    """
+    Scan latest Solana pairs from DexScreener for scam patterns.
+    Pushes alerts for high-risk tokens.
+    Returns number of alerts generated.
+    """
+    import httpx
+
+    pushed = 0
+
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            resp = await client.get("https://api.dexscreener.com/latest/dex/search", params={"q": "SOL USDC"})
+            if resp.status_code != 200:
+                return 0
+
+            data = resp.json()
+            pairs = data.get("pairs", [])[:limit]
+
+            for pair in pairs:
+                token_addr = pair.get("baseToken", {}).get("address", "")
+                token_name = pair.get("baseToken", {}).get("name", "Unknown")
+                token_symbol = pair.get("baseToken", {}).get("symbol", "???")
+                liquidity = float(pair.get("liquidity", {}).get("usd", 0) or 0)
+                volume = float(pair.get("volume", {}).get("h24", 0) or 0)
+                price_change = float(pair.get("priceChange", {}).get("h24", 0) or 0)
+                created_at = pair.get("pairCreatedAt", 0)
+                age_hours = (time.time() - created_at / 1000) / 3600 if created_at else 999
+
+                # Risk heuristics
+                risk_flags = []
+
+                if liquidity < 1000:
+                    risk_flags.append("low_liquidity")
+                if volume == 0:
+                    risk_flags.append("no_volume")
+                if price_change < -80:
+                    risk_flags.append(f"crashed_{abs(price_change):.0f}%")
+                if age_hours < 1 and liquidity < 5000:
+                    risk_flags.append("fresh_launch_low_liq")
+                if price_change > 500:
+                    risk_flags.append(f"pumped_{price_change:.0f}%")
+
+                if risk_flags:
+                    await push_alert(
+                        alert_type="new_pair_risk",
+                        title=f"{token_symbol}: {' | '.join(risk_flags[:2])}",
+                        description=f"New pair {token_name} ({token_symbol}) on Solana. "
+                        f"Liquidity: ${liquidity:,.0f}, Age: {age_hours:.1f}h, "
+                        f"24h change: {price_change:+.1f}%",
+                        severity="critical" if len(risk_flags) >= 3 else "high",
+                        chain="solana",
+                        token=token_addr,
+                        token_symbol=token_symbol,
+                        risk_score=min(90, len(risk_flags) * 25),
+                        metadata={
+                            "risk_flags": risk_flags,
+                            "liquidity_usd": liquidity,
+                            "age_hours": age_hours,
+                        },
+                    )
+                    pushed += 1
+
+        return pushed
+    except Exception as e:
+        logger.warning(f"Solana scan error: {e}")
+        return pushed
+
+
+async def scan_known_scams(limit: int = 3) -> int:
+    """
+    Check our RAG known_scams collection for recently added entries.
+    Pushes alerts for new scam patterns detected.
+    """
+    pushed = 0
+    try:
+        r = await _get_redis()
+        # Check for recent scam pattern additions
+        scam_ids = await r.smembers("rag:idx:known_scams")
+
+        recent_count = 0
+        for sid in list(scam_ids)[:20]:
+            doc = await r.get(f"rag:known_scams:{sid}")
+            if doc:
+                try:
+                    data = json.loads(doc)
+                    added = data.get("metadata", {}).get("added_at", "")
+                    if added:
+                        age_h = (time.time() - datetime.fromisoformat(added.replace("Z", "+00:00")).timestamp()) / 3600
+                        if age_h < 24:
+                            recent_count += 1
+                except Exception:
+                    pass
+
+        await r.close()
+
+        if recent_count > 0:
+            await push_alert(
+                alert_type="scam_pattern_update",
+                title=f"{recent_count} new scam patterns detected in last 24h",
+                description="New rug pull, honeypot, or phishing patterns added to the knowledge base.",
+                severity="high",
+                chain="multi",
+                risk_score=85,
+            )
+            pushed += 1
+
+        return pushed
+    except Exception as e:
+        logger.warning(f"Known scams scan error: {e}")
+        return pushed
+
+
+async def run_alert_scan() -> dict[str, int]:
+    """
+    Run a full alert scan across all sources.
+    Called by cron job every 15 minutes.
+    """
+    results = {}
+
+    # Scan Solana new pairs
+    results["solana_pairs"] = await scan_solana_new_pairs(limit=8)
+
+    # Scan known scams
+    results["known_scams"] = await scan_known_scams()
+
+    total = sum(results.values())
+    logger.info(f"Alert scan complete: {total} new alerts ({results})")
+    return results
+
+
+# ── Seed some initial alerts if Redis is empty ────────────────────
+
+
+async def seed_initial_alerts():
+    """Seed baseline alerts so the system isn't empty on first run."""
+    r = await _get_redis()
+    existing = await r.zcard(ALERT_KEY)
+    await r.close()
+
+    if existing > 0:
+        return  # Already has alerts
+
+    seeds = [
+        (
+            "honeypot",
+            "Honeypot detected on Base: 0xdead...",
+            "Token has sell restrictions and blacklist. Buyers cannot exit.",
+            "critical",
+            "base",
+        ),
+        (
+            "whale",
+            "Whale moved 5M USDC to Binance",
+            "Wallet 0xABCD... transferred $5M USDC to Binance hot wallet. Possible sell pressure.",
+            "high",
+            "ethereum",
+        ),
+        (
+            "rugpull",
+            "Liquidity removed from $SCAM token",
+            "100% of liquidity pool withdrawn by deployer. Token is now worthless.",
+            "critical",
+            "solana",
+        ),
+        (
+            "bundler",
+            "Sniper bundle detected: $NEWLAUNCH",
+            "Coordinated wallet cluster bought 60% of supply in first block.",
+            "high",
+            "solana",
+        ),
+        (
+            "contract",
+            "Unverified proxy contract found",
+            "Token uses upgradeable proxy with unverified implementation. Owner can change logic.",
+            "high",
+            "base",
+        ),
+        (
+            "concentration",
+            "90% supply held by 3 wallets on $DANGER",
+            "Extreme holder concentration. Classic rug pull setup.",
+            "critical",
+            "ethereum",
+        ),
+        (
+            "wash_trade",
+            "Wash trading detected on $FAKEVOL",
+            "95% of volume is self-trading between linked wallets.",
+            "high",
+            "bsc",
+        ),
+        (
+            "phishing",
+            "Fake airdrop targeting $BONK holders",
+            "Phishing site detected posing as official BONK airdrop. Users losing funds.",
+            "critical",
+            "solana",
+        ),
+    ]
+
+    for alert_type, title, desc, severity, chain in seeds:
+        await push_alert(
+            alert_type=alert_type,
+            title=title,
+            description=desc,
+            severity=severity,
+            chain=chain,
+        )
+
+    logger.info(f"Seeded {len(seeds)} initial alerts")

app/api/v1/mcp/__init__.py

@@ -1,10 +1,4 @@
-"""Deprecated shim - MCP router moved to app.domains.mcp.router.
-
-Migrate imports from `app.api.v1.mcp` to `app.domains.mcp`.
-This shim will be removed in Phase 3 cleanup.
-"""
-from __future__ import annotations
-
-from app.domains.mcp.router import router
+"""MCP v1 routes."""
+from .router import router
 
 __all__ = ["router"]

app/api/v1/mcp/router.py

@@ -18,7 +18,7 @@ from typing import Any
 from fastapi import APIRouter, Request
 from pydantic import BaseModel
 
-from app.domains.mcp.server import (
+from app.mcp.server import (
     MCP_PROTOCOL_VERSION,
     MCP_SERVER_VERSION,
     TOOL_CATALOG,

app/billing/__init__.py

@@ -0,0 +1,23 @@
+"""billing — DEPRECATED shim. Use app.domains.billing.
+
+Phase 4.1 of AUDIT-2026-Q3.md moved app/billing/ to app/domains/billing/.
+This shim re-exports the canonical surface and aliases submodules so that
+`from app.billing.x402.router import router` keeps working.
+
+MIGRATION: replace `from app.billing.x402 import ...` with
+`from app.domains.billing.x402 import ...`.
+"""
+import sys as _sys
+from app.domains.billing import x402 as _x402
+from app.domains.billing.x402 import router as _router
+
+# Public re-exports
+from app.domains.billing.x402 import (  # noqa: F401
+    X402Enforcer,
+    TOOL_PRICES,
+    CHAIN_USDC,
+)
+
+# Alias submodules so `from app.billing.x402.router import router` works
+_sys.modules["app.billing.x402"] = _x402
+_sys.modules["app.billing.x402.router"] = _router

app/bulletin_board.py

@@ -1,21 +1,903 @@
-"""Deprecated shim - re-exports app.domains.bulletin.
-
-Migrate imports from `app.bulletin_board` to `app.domains.bulletin`.
-This shim will be removed in Phase 3 cleanup.
 """
+RMI Bulletin Board Management System
+=====================================
+A full content management backend for announcements, news, alerts, platform
+communications, and community bulletin boards.
+
+Features:
+  • Posts - CRUD with rich text, attachments, scheduling, expiry
+  • Categories - organize by type (news, alert, update, promo, system)
+  • Targeting - audience segmentation (all, free, premium, admins, specific tiers)
+  • Moderation - draft/review/published/archived workflow, approval chains
+  • Pinning - sticky posts, priority ordering
+  • Analytics - views, clicks, engagement tracking per post
+  • Comments - threaded discussions on posts (optional)
+  • Notifications - push/email/Telegram alerts for critical posts
+  • Scheduling - publish at future date, auto-archive after expiry
+  • Versioning - track edit history, rollback capability
+  • Search - full-text search across all posts
+  • SEO - slug generation, meta tags, OpenGraph
+
+Security:
+  - All write operations require admin auth + content.write permission
+  - Audit log of every content change
+  - Rate limiting on publish operations
+  - Content sanitization (strip XSS, validate HTML)
+"""
+
 from __future__ import annotations
 
-from app.domains.bulletin import (  # noqa: F401
-    BulletinBoardManager,
-    Comment,
-    ContentSanitizer,
-    Post,
-    PostCategory,
-    PostStatus,
-    Priority,
-    TargetAudience,
-    award_badge,
-    get_user_badges,
-    get_user_reputation,
-    verify_x402_bot,
-)
+import html
+import json
+import logging
+import os
+import re
+import time
+from dataclasses import asdict, dataclass, field
+from datetime import datetime
+from enum import StrEnum
+from typing import Any, ClassVar
+
+logger = logging.getLogger("rmi_bulletin_board")
+
+
+# ── Enums ─────────────────────────────────────────────────────
+
+
+class PostStatus(StrEnum):
+    DRAFT = "draft"
+    REVIEW = "review"
+    PUBLISHED = "published"
+    ARCHIVED = "archived"
+    SCHEDULED = "scheduled"
+
+
+class PostCategory(StrEnum):
+    NEWS = "news"  # Platform news
+    ALERT = "alert"  # Security/urgent alerts
+    UPDATE = "update"  # Feature updates
+    PROMO = "promo"  # Promotions/offers
+    SYSTEM = "system"  # System maintenance
+    COMMUNITY = "community"  # Community posts
+    ANNOUNCEMENT = "announcement"  # General announcements
+    TUTORIAL = "tutorial"  # Guides/how-tos
+
+
+class TargetAudience(StrEnum):
+    ALL = "all"
+    FREE = "free"
+    PREMIUM = "premium"
+    PRO = "pro"
+    ENTERPRISE = "enterprise"
+    ADMINS = "admins"
+    MODERATORS = "moderators"
+
+
+class Priority(StrEnum):
+    LOW = "low"
+    NORMAL = "normal"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+# ── Data Models ─────────────────────────────────────────────
+
+
+@dataclass
+class Post:
+    """Bulletin board post."""
+
+    post_id: str
+    title: str
+    slug: str
+    content: str
+    summary: str
+    category: str
+    status: str
+    priority: str
+    target_audience: str
+    author_id: str
+    author_email: str
+    author_name: str
+    created_at: str
+    updated_at: str
+    published_at: str | None = None
+    scheduled_at: str | None = None
+    expires_at: str | None = None
+    archived_at: str | None = None
+    pinned: bool = False
+    pin_order: int = 0
+    featured_image: str = ""
+    attachments: list[dict] = field(default_factory=list)
+    tags: list[str] = field(default_factory=list)
+    meta_title: str = ""
+    meta_description: str = ""
+    og_image: str = ""
+    view_count: int = 0
+    click_count: int = 0
+    engagement_score: float = 0.0
+    version: int = 1
+    edit_history: list[dict] = field(default_factory=list)
+    approved_by: str = ""
+    approved_at: str | None = None
+    notification_sent: bool = False
+    allow_comments: bool = False
+    comments_count: int = 0
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    def to_public_dict(self) -> dict:
+        """Return public-safe version (no internal fields)."""
+        return {
+            "post_id": self.post_id,
+            "title": self.title,
+            "slug": self.slug,
+            "content": self.content,
+            "summary": self.summary,
+            "category": self.category,
+            "priority": self.priority,
+            "author_name": self.author_name,
+            "created_at": self.created_at,
+            "published_at": self.published_at,
+            "expires_at": self.expires_at,
+            "pinned": self.pinned,
+            "pin_order": self.pin_order,
+            "featured_image": self.featured_image,
+            "attachments": self.attachments,
+            "tags": self.tags,
+            "meta_title": self.meta_title,
+            "meta_description": self.meta_description,
+            "og_image": self.og_image,
+            "view_count": self.view_count,
+            "allow_comments": self.allow_comments,
+            "comments_count": self.comments_count,
+        }
+
+
+@dataclass
+class Comment:
+    """Comment on a post."""
+
+    comment_id: str
+    post_id: str
+    author_id: str
+    author_name: str
+    author_email: str
+    content: str
+    created_at: str
+    updated_at: str
+    parent_id: str | None = None
+    status: str = "approved"  # approved, pending, rejected
+    likes: int = 0
+    replies: list[dict] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+# ── Content Sanitizer ───────────────────────────────────────
+
+
+class ContentSanitizer:
+    """Sanitize user-generated content to prevent XSS."""
+
+    ALLOWED_TAGS: ClassVar[dict] ={
+        "p",
+        "br",
+        "strong",
+        "b",
+        "em",
+        "i",
+        "u",
+        "h1",
+        "h2",
+        "h3",
+        "h4",
+        "h5",
+        "h6",
+        "ul",
+        "ol",
+        "li",
+        "a",
+        "img",
+        "blockquote",
+        "code",
+        "pre",
+        "table",
+        "thead",
+        "tbody",
+        "tr",
+        "td",
+        "th",
+        "div",
+        "span",
+        "hr",
+        "sub",
+        "sup",
+        "del",
+        "ins",
+    }
+
+    ALLOWED_ATTRS: ClassVar[dict] ={
+        "a": ["href", "title", "target"],
+        "img": ["src", "alt", "title", "width", "height"],
+        "div": ["class"],
+        "span": ["class"],
+        "code": ["class"],
+        "pre": ["class"],
+    }
+
+    @staticmethod
+    def sanitize(text: str) -> str:
+        """Basic HTML sanitization."""
+        if not text:
+            return ""
+
+        # Escape HTML entities first
+        text = html.escape(text)
+
+        # Then selectively un-escape allowed tags
+        # This is a simplified approach - in production use bleach or similar
+        # For now, strip all HTML tags for safety
+        text = re.sub(r"<[^>]+>", "", text)
+
+        return text.strip()
+
+    @staticmethod
+    def generate_slug(title: str) -> str:
+        """Generate URL-friendly slug from title."""
+        slug = re.sub(r"[^\w\s-]", "", title.lower())
+        slug = re.sub(r"[-\s]+", "-", slug)
+        return slug[:80]
+
+    @staticmethod
+    def generate_summary(content: str, max_length: int = 200) -> str:
+        """Generate summary from content."""
+        # Strip HTML
+        text = re.sub(r"<[^>]+>", "", content)
+        text = text.replace("\n", " ").strip()
+        if len(text) > max_length:
+            text = text[:max_length].rsplit(" ", 1)[0] + "..."
+        return text
+
+
+# ── Bulletin Board Manager ────────────────────────────────────
+
+
+class BulletinBoardManager:
+    """
+    Core manager for bulletin board operations.
+    Uses Redis as primary store + Supabase backup.
+    """
+
+    @staticmethod
+    async def _get_redis():
+        import redis.asyncio as redis_lib
+
+        return redis_lib.Redis(
+            host=os.getenv("REDIS_HOST", "localhost"),
+            port=int(os.getenv("REDIS_PORT", "6379")),
+            password=os.getenv("REDIS_PASSWORD", ""),
+            decode_responses=True,
+        )
+
+    @staticmethod
+    async def create_post(
+        title: str,
+        content: str,
+        category: str,
+        author_id: str,
+        author_email: str,
+        author_name: str = "",
+        priority: str = "normal",
+        target_audience: str = "all",
+        status: str = "draft",
+        featured_image: str = "",
+        attachments: list[dict] | None = None,
+        tags: list[str] | None = None,
+        scheduled_at: str | None = None,
+        expires_at: str | None = None,
+        pinned: bool = False,
+        allow_comments: bool = False,
+        meta_title: str = "",
+        meta_description: str = "",
+    ) -> Post:
+        """Create a new post."""
+        post_id = f"post_{int(time.time())}_{os.urandom(4).hex()}"
+        slug = ContentSanitizer.generate_slug(title)
+        summary = ContentSanitizer.generate_summary(content)
+        now = datetime.utcnow().isoformat()
+
+        # Sanitize content
+        content = ContentSanitizer.sanitize(content)
+
+        post = Post(
+            post_id=post_id,
+            title=title[:200],
+            slug=slug,
+            content=content,
+            summary=summary,
+            category=category,
+            status=status,
+            priority=priority,
+            target_audience=target_audience,
+            author_id=author_id,
+            author_email=author_email,
+            author_name=author_name or author_email.split("@")[0],
+            created_at=now,
+            updated_at=now,
+            scheduled_at=scheduled_at,
+            expires_at=expires_at,
+            pinned=pinned,
+            featured_image=featured_image,
+            attachments=attachments or [],
+            tags=tags or [],
+            meta_title=meta_title or title[:70],
+            meta_description=meta_description or summary[:160],
+            allow_comments=allow_comments,
+        )
+
+        r = await BulletinBoardManager._get_redis()
+
+        # Save post
+        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post.to_dict()))
+
+        # Add to category index
+        await r.sadd(f"bulletin:category:{category}", post_id)
+
+        # Add to status index
+        await r.sadd(f"bulletin:status:{status}", post_id)
+
+        # Add to author index
+        await r.sadd(f"bulletin:author:{author_id}", post_id)
+
+        # Add to audience index
+        await r.sadd(f"bulletin:audience:{target_audience}", post_id)
+
+        # Add to pinned index if pinned
+        if pinned:
+            await r.sadd("bulletin:pinned", post_id)
+            await r.zadd("bulletin:pinned_order", {post_id: post.pin_order})
+
+        # Add to scheduled index if scheduled
+        if scheduled_at and status == "scheduled":
+            ts = int(datetime.fromisoformat(scheduled_at).timestamp())
+            await r.zadd("bulletin:scheduled", {post_id: ts})
+
+        # Add to search index (simple word index)
+        words = set(re.findall(r"\w+", title.lower() + " " + content.lower()))
+        for word in words:
+            if len(word) > 2:
+                await r.sadd(f"bulletin:search:{word}", post_id)
+
+        # Save to Supabase
+        try:
+            from supabase import create_client
+
+            supabase_url = os.getenv("SUPABASE_URL")
+            supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
+            if supabase_url and supabase_key:
+                client = create_client(supabase_url, supabase_key)
+                client.table("bulletin_posts").insert(post.to_dict()).execute()
+        except Exception as e:
+            logger.error(f"Supabase bulletin post save failed: {e}")
+
+        return post
+
+    @staticmethod
+    async def get_post(post_id: str, increment_views: bool = False) -> Post | None:
+        """Get a post by ID."""
+        r = await BulletinBoardManager._get_redis()
+        data = await r.hget("rmi:bulletin_posts", post_id)
+        if not data:
+            return None
+
+        post_dict = json.loads(data)
+
+        if increment_views and post_dict.get("status") == "published":
+            post_dict["view_count"] = post_dict.get("view_count", 0) + 1
+            await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+
+        return Post(**post_dict)
+
+    @staticmethod
+    async def get_post_by_slug(slug: str) -> Post | None:
+        """Get a post by slug."""
+        r = await BulletinBoardManager._get_redis()
+        # Search all posts for matching slug
+        all_posts = await r.hgetall("rmi:bulletin_posts")
+        for _post_id, data in all_posts.items():
+            post_dict = json.loads(data)
+            if post_dict.get("slug") == slug:
+                return Post(**post_dict)
+        return None
+
+    @staticmethod
+    async def update_post(
+        post_id: str,
+        updates: dict[str, Any],
+        editor_id: str = "",
+        editor_email: str = "",
+    ) -> Post | None:
+        """Update a post with versioning."""
+        r = await BulletinBoardManager._get_redis()
+        data = await r.hget("rmi:bulletin_posts", post_id)
+        if not data:
+            return None
+
+        post_dict = json.loads(data)
+        before_state = {k: post_dict.get(k) for k in updates if k in post_dict}
+
+        # Record edit history
+        edit_entry = {
+            "edited_at": datetime.utcnow().isoformat(),
+            "edited_by": editor_id,
+            "editor_email": editor_email,
+            "changes": before_state,
+            "version": post_dict.get("version", 1),
+        }
+
+        history = post_dict.get("edit_history", [])
+        history.append(edit_entry)
+        post_dict["edit_history"] = history
+        post_dict["version"] = post_dict.get("version", 1) + 1
+        post_dict["updated_at"] = datetime.utcnow().isoformat()
+
+        # Apply updates
+        for key, value in updates.items():
+            if key == "content":
+                value = ContentSanitizer.sanitize(value)
+                post_dict["summary"] = ContentSanitizer.generate_summary(value)
+            if key == "title":
+                post_dict["slug"] = ContentSanitizer.generate_slug(value)
+            post_dict[key] = value
+
+        # Handle status transitions
+        old_status = before_state.get("status")
+        new_status = post_dict.get("status")
+        if old_status != new_status:
+            # Update status indexes
+            if old_status:
+                await r.srem(f"bulletin:status:{old_status}", post_id)
+            await r.sadd(f"bulletin:status:{new_status}", post_id)
+
+            if new_status == "published":
+                post_dict["published_at"] = datetime.utcnow().isoformat()
+            elif new_status == "archived":
+                post_dict["archived_at"] = datetime.utcnow().isoformat()
+
+        # Handle pinning changes
+        if "pinned" in updates:
+            if updates["pinned"]:
+                await r.sadd("bulletin:pinned", post_id)
+                await r.zadd("bulletin:pinned_order", {post_id: post_dict.get("pin_order", 0)})
+            else:
+                await r.srem("bulletin:pinned", post_id)
+                await r.zrem("bulletin:pinned_order", post_id)
+
+        # Save updated post
+        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+
+        # Update Supabase
+        try:
+            from supabase import create_client
+
+            supabase_url = os.getenv("SUPABASE_URL")
+            supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
+            if supabase_url and supabase_key:
+                client = create_client(supabase_url, supabase_key)
+                client.table("bulletin_posts").upsert(post_dict).execute()
+        except Exception as e:
+            logger.error(f"Supabase bulletin post update failed: {e}")
+
+        return Post(**post_dict)
+
+    @staticmethod
+    async def delete_post(post_id: str) -> bool:
+        """Delete a post (soft delete - move to archive)."""
+        r = await BulletinBoardManager._get_redis()
+        data = await r.hget("rmi:bulletin_posts", post_id)
+        if not data:
+            return False
+
+        post_dict = json.loads(data)
+        post_dict["status"] = "archived"
+        post_dict["archived_at"] = datetime.utcnow().isoformat()
+
+        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+        await r.srem("bulletin:status:published", post_id)
+        await r.srem("bulletin:status:draft", post_id)
+        await r.srem("bulletin:status:review", post_id)
+        await r.sadd("bulletin:status:archived", post_id)
+        await r.srem("bulletin:pinned", post_id)
+
+        return True
+
+    @staticmethod
+    async def list_posts(
+        category: str | None = None,
+        status: str | None = None,
+        target_audience: str | None = None,
+        author_id: str | None = None,
+        pinned_only: bool = False,
+        search_query: str | None = None,
+        tags: list[str] | None = None,
+        limit: int = 50,
+        offset: int = 0,
+        sort_by: str = "created_at",
+        sort_order: str = "desc",
+    ) -> dict[str, Any]:
+        """List posts with filtering."""
+        r = await BulletinBoardManager._get_redis()
+
+        # Start with all posts or filtered set
+        if pinned_only:
+            post_ids = await r.smembers("bulletin:pinned")
+        elif category:
+            post_ids = await r.smembers(f"bulletin:category:{category}")
+        elif status:
+            post_ids = await r.smembers(f"bulletin:status:{status}")
+        elif author_id:
+            post_ids = await r.smembers(f"bulletin:author:{author_id}")
+        elif target_audience:
+            post_ids = await r.smembers(f"bulletin:audience:{target_audience}")
+        elif search_query:
+            # Search by words
+            words = re.findall(r"\w+", search_query.lower())
+            if words:
+                sets = [f"bulletin:search:{w}" for w in words if len(w) > 2]
+                if sets:
+                    post_ids = await r.sinter(sets)
+                else:
+                    post_ids = set()
+            else:
+                post_ids = set()
+        else:
+            all_posts = await r.hgetall("rmi:bulletin_posts")
+            post_ids = set(all_posts.keys())
+
+        # Apply additional filters
+        if tags:
+            tagged_posts = set()
+            all_posts = await r.hgetall("rmi:bulletin_posts")
+            for pid, data in all_posts.items():
+                post_dict = json.loads(data)
+                if any(tag in post_dict.get("tags", []) for tag in tags):
+                    tagged_posts.add(pid)
+            post_ids = post_ids.intersection(tagged_posts)
+
+        # Fetch and sort posts
+        posts = []
+        all_posts = await r.hgetall("rmi:bulletin_posts")
+        for pid in post_ids:
+            data = all_posts.get(pid)
+            if data:
+                post_dict = json.loads(data)
+                posts.append(post_dict)
+
+        # Sort
+        reverse = sort_order == "desc"
+        posts.sort(key=lambda x: x.get(sort_by, ""), reverse=reverse)
+
+        # Pinned posts first if not pinned_only
+        if not pinned_only:
+            pinned_ids = await r.smembers("bulletin:pinned")
+            posts.sort(
+                key=lambda x: (x["post_id"] not in pinned_ids, x.get(sort_by, "")),
+                reverse=not reverse,
+            )
+
+        total = len(posts)
+        posts = posts[offset : offset + limit]
+
+        return {
+            "posts": posts,
+            "total": total,
+            "limit": limit,
+            "offset": offset,
+        }
+
+    @staticmethod
+    async def publish_scheduled() -> list[str]:
+        """Publish posts that are scheduled for now."""
+        r = await BulletinBoardManager._get_redis()
+        now = int(time.time())
+
+        # Get scheduled posts that are due
+        due = await r.zrangebyscore("bulletin:scheduled", 0, now)
+        published = []
+
+        for post_id in due:
+            data = await r.hget("rmi:bulletin_posts", post_id)
+            if data:
+                post_dict = json.loads(data)
+                post_dict["status"] = "published"
+                post_dict["published_at"] = datetime.utcnow().isoformat()
+                post_dict["scheduled_at"] = None
+
+                await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+                await r.srem("bulletin:status:scheduled", post_id)
+                await r.sadd("bulletin:status:published", post_id)
+                await r.zrem("bulletin:scheduled", post_id)
+
+                published.append(post_id)
+
+        return published
+
+    @staticmethod
+    async def archive_expired() -> list[str]:
+        """Archive posts that have expired."""
+        r = await BulletinBoardManager._get_redis()
+        now = datetime.utcnow().isoformat()
+
+        all_posts = await r.hgetall("rmi:bulletin_posts")
+        archived = []
+
+        for post_id, data in all_posts.items():
+            post_dict = json.loads(data)
+            if post_dict.get("status") == "published" and post_dict.get("expires_at"):
+                if post_dict["expires_at"] < now:
+                    post_dict["status"] = "archived"
+                    post_dict["archived_at"] = now
+
+                    await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+                    await r.srem("bulletin:status:published", post_id)
+                    await r.sadd("bulletin:status:archived", post_id)
+                    await r.srem("bulletin:pinned", post_id)
+
+                    archived.append(post_id)
+
+        return archived
+
+    @staticmethod
+    async def add_comment(
+        post_id: str,
+        author_id: str,
+        author_name: str,
+        author_email: str,
+        content: str,
+        parent_id: str | None = None,
+    ) -> Comment | None:
+        """Add a comment to a post."""
+        r = await BulletinBoardManager._get_redis()
+
+        # Check if post exists and allows comments
+        post_data = await r.hget("rmi:bulletin_posts", post_id)
+        if not post_data:
+            return None
+
+        post_dict = json.loads(post_data)
+        if not post_dict.get("allow_comments", False):
+            return None
+
+        comment_id = f"comment_{int(time.time())}_{os.urandom(4).hex()}"
+        now = datetime.utcnow().isoformat()
+
+        comment = Comment(
+            comment_id=comment_id,
+            post_id=post_id,
+            author_id=author_id,
+            author_name=author_name,
+            author_email=author_email,
+            content=ContentSanitizer.sanitize(content)[:2000],
+            created_at=now,
+            updated_at=now,
+            parent_id=parent_id,
+        )
+
+        await r.hset("rmi:bulletin_comments", comment_id, json.dumps(comment.to_dict()))
+        await r.sadd(f"bulletin:post_comments:{post_id}", comment_id)
+
+        # Update post comment count
+        post_dict["comments_count"] = post_dict.get("comments_count", 0) + 1
+        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+
+        return comment
+
+    @staticmethod
+    async def get_comments(post_id: str, limit: int = 100) -> list[dict]:
+        """Get comments for a post."""
+        r = await BulletinBoardManager._get_redis()
+        comment_ids = await r.smembers(f"bulletin:post_comments:{post_id}")
+
+        comments = []
+        for cid in comment_ids:
+            data = await r.hget("rmi:bulletin_comments", cid)
+            if data:
+                comments.append(json.loads(data))
+
+        comments.sort(key=lambda x: x.get("created_at", ""), reverse=True)
+        return comments[:limit]
+
+    @staticmethod
+    async def get_stats() -> dict[str, Any]:
+        """Get bulletin board statistics."""
+        r = await BulletinBoardManager._get_redis()
+
+        stats = {
+            "total_posts": await r.hlen("rmi:bulletin_posts") or 0,
+            "published": await r.scard("bulletin:status:published") or 0,
+            "drafts": await r.scard("bulletin:status:draft") or 0,
+            "review": await r.scard("bulletin:status:review") or 0,
+            "archived": await r.scard("bulletin:status:archived") or 0,
+            "scheduled": await r.scard("bulletin:status:scheduled") or 0,
+            "pinned": await r.scard("bulletin:pinned") or 0,
+            "total_comments": await r.hlen("rmi:bulletin_comments") or 0,
+            "categories": {},
+        }
+
+        # Count by category
+        for cat in PostCategory:
+            count = await r.scard(f"bulletin:category:{cat.value}") or 0
+            stats["categories"][cat.value] = count
+
+        return stats
+
+    @staticmethod
+    async def track_engagement(post_id: str, action: str) -> bool:
+        """Track engagement (view, click, etc.)."""
+        r = await BulletinBoardManager._get_redis()
+        data = await r.hget("rmi:bulletin_posts", post_id)
+        if not data:
+            return False
+
+        post_dict = json.loads(data)
+
+        if action == "view":
+            post_dict["view_count"] = post_dict.get("view_count", 0) + 1
+        elif action == "click":
+            post_dict["click_count"] = post_dict.get("click_count", 0) + 1
+
+        # Calculate engagement score
+        views = post_dict.get("view_count", 0)
+        clicks = post_dict.get("click_count", 0)
+        post_dict["engagement_score"] = round((clicks / max(views, 1)) * 100, 2)
+
+        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
+        return True
+
+
+# ═══════════════════════════════════════════════════════════
+# BADGES & X402 BOT PAYMENTS
+# ═══════════════════════════════════════════════════════════
+
+BADGES = {
+    "first_post": {
+        "name": "First Words",
+        "icon": "💬",
+        "desc": "Made your first post",
+        "tier": "bronze",
+    },
+    "10_posts": {"name": "Chatterbox", "icon": "📢", "desc": "10 posts", "tier": "bronze"},
+    "50_posts": {"name": "Board Regular", "icon": "🎙️", "desc": "50 posts", "tier": "silver"},
+    "100_posts": {
+        "name": "Terminally Online",
+        "icon": "🖥️",
+        "desc": "100 posts - touch grass",
+        "tier": "gold",
+    },
+    "10_upvotes": {
+        "name": "Approved",
+        "icon": "👍",
+        "desc": "10 upvotes on a post",
+        "tier": "bronze",
+    },
+    "50_upvotes": {"name": "Crowd Favorite", "icon": "⭐", "desc": "50 upvotes", "tier": "silver"},
+    "100_upvotes": {"name": "Legendary", "icon": "👑", "desc": "100 upvotes", "tier": "gold"},
+    "rug_reporter": {
+        "name": "Rug Detective",
+        "icon": "🔍",
+        "desc": "Reported 3 verified rugs",
+        "tier": "silver",
+    },
+    "scam_buster": {
+        "name": "Scam Buster",
+        "icon": "🛡️",
+        "desc": "10 scam alerts verified",
+        "tier": "gold",
+    },
+    "honeypot_hunter": {
+        "name": "Honeypot Hunter",
+        "icon": "🍯",
+        "desc": "Found 5 honeypots",
+        "tier": "silver",
+    },
+    "whale_watcher": {
+        "name": "Whale Watcher",
+        "icon": "🐋",
+        "desc": "Tracked 10 whale moves",
+        "tier": "silver",
+    },
+    "alpha_caller": {
+        "name": "Alpha Caller",
+        "icon": "📈",
+        "desc": "Called 5 pumps",
+        "tier": "gold",
+    },
+    "degen": {
+        "name": "Certified Degen",
+        "icon": "🎰",
+        "desc": "Posted in every category",
+        "tier": "gold",
+    },
+    "rug_survivor": {
+        "name": "Rug Survivor",
+        "icon": "💀",
+        "desc": "Posted about getting rugged",
+        "tier": "bronze",
+    },
+    "diamond_hands": {
+        "name": "Diamond Hands",
+        "icon": "💎",
+        "desc": "Held through -90%",
+        "tier": "diamond",
+    },
+    "based": {
+        "name": "Based",
+        "icon": "🧠",
+        "desc": "Called a 10x before it happened",
+        "tier": "diamond",
+    },
+    "bot_verified": {
+        "name": "Verified Bot",
+        "icon": "🤖",
+        "desc": "Registered x402 bot",
+        "tier": "silver",
+    },
+    "bot_pro": {"name": "Bot Pro", "icon": "⚡", "desc": "100+ API calls", "tier": "gold"},
+}
+BADGE_TIERS = {"bronze": "#CD7F32", "silver": "#C0C0C0", "gold": "#FFD700", "diamond": "#B9F2FF"}
+
+
+async def get_user_badges(user_id: str) -> list:
+    try:
+        r = await BulletinBoardManager._get_redis()
+        ids = await r.smembers(f"bb:badges:{user_id}")
+        await r.close()
+        return [{"id": bid, **BADGES[bid]} for bid in ids if bid in BADGES]
+    except Exception:
+        return []
+
+
+async def award_badge(user_id: str, badge_id: str) -> bool:
+    if badge_id not in BADGES:
+        return False
+    try:
+        r = await BulletinBoardManager._get_redis()
+        ok = await r.sadd(f"bb:badges:{user_id}", badge_id)
+        await r.close()
+        return ok > 0
+    except Exception:
+        return False
+
+
+async def get_user_reputation(user_id: str) -> dict:
+    try:
+        r = await BulletinBoardManager._get_redis()
+        p = r.pipeline()
+        p.get(f"bb:rep:{user_id}")
+        p.scard(f"bb:badges:{user_id}")
+        p.get(f"bb:posts:{user_id}")
+        rep, bc, pc = await p.execute()
+        await r.close()
+        return {"reputation": int(rep or 0), "badge_count": bc or 0, "post_count": int(pc or 0)}
+    except Exception:
+        return {"reputation": 0, "badge_count": 0, "post_count": 0}
+
+
+X402_BB_POST_PRICE = "$1.00"
+
+
+async def verify_x402_bot(tx_hash: str, bot_addr: str) -> bool:
+    try:
+        r = await BulletinBoardManager._get_redis()
+        if await r.get(f"bb:x402:tx:{tx_hash}"):
+            await r.close()
+            return False
+        await r.setex(f"bb:x402:tx:{tx_hash}", 86400, bot_addr)
+        await r.setex(f"bb:x402:bot:{bot_addr}", 2592000, "1")  # 30 day auth
+        await r.close()
+        return True
+    except Exception:
+        return False

app/contract_deepscan.py

@@ -14,7 +14,7 @@ import tempfile
 from dataclasses import asdict, dataclass
 from typing import Any
 
-import httpx
+from app.telegram_bot.requirements import httpx
 
 logger = logging.getLogger(__name__)
 

app/core/agent_memory.py

@@ -4,9 +4,10 @@ Enables agents to remember past interactions across sessions."""
 import os
 from datetime import UTC, datetime
 
-import httpx
 from fastapi import APIRouter
 
+from app.telegram_bot.requirements import httpx
+
 MEMGRAPH_URI = os.getenv("MEMGRAPH_URI", "bolt://localhost:7687")
 MEMGRAPH_USER = os.getenv("MEMGRAPH_USER", "")
 MEMGRAPH_PASS = os.getenv("MEMGRAPH_PASSWORD", "")

app/core/langfuse.py

@@ -1,49 +0,0 @@
-"""Langfuse LLM tracing - opt-in via env vars.
-
-Safe no-op when LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY are unset
-or the `langfuse` package is not installed.
-"""
-
-import os
-
-LANGFUSE_ENABLED = bool(
-    os.getenv("LANGFUSE_PUBLIC_KEY", "") and os.getenv("LANGFUSE_SECRET_KEY", "")
-)
-LANGFUSE_HOST = os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
-
-_initialized = False
-
-
-def init_langfuse() -> bool:
-    """Initialize the Langfuse client. Returns True on success."""
-    global _initialized
-    if not LANGFUSE_ENABLED:
-        return False
-    try:
-        from langfuse import Langfuse  # noqa: F401
-
-        Langfuse(
-            public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
-            secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
-            host=LANGFUSE_HOST,
-        )
-        _initialized = True
-        return True
-    except ImportError:
-        return False
-    except Exception:
-        return False
-
-
-def flush_langfuse() -> None:
-    """Flush any pending Langfuse events. Safe to call on shutdown."""
-    if not _initialized:
-        return
-    try:
-        from langfuse import Langfuse
-
-        Langfuse().flush()
-    except ImportError:
-        pass
-    except Exception:
-        pass

app/core/lifespan.py

@@ -59,7 +59,7 @@ async def _start_background_tasks(app: FastAPI) -> None:
 
     # Cache warmer (passes app instance)
     try:
-        from app.domains.databus.core import cache_warm_loop
+        from app.databus.core import cache_warm_loop
 
         asyncio.create_task(cache_warm_loop(app))
         logger.info("background_task_started", task="databus_cache_warmer", interval="")

app/core/redis.py

@@ -73,7 +73,7 @@ def get_redis_async() -> "aioredis.Redis | None":
         return None
 
 
-async def close_redis() -> None:
+def close_redis():
     """Close Redis connection pool."""
     global _sync_pool, _async_pool
     if _sync_pool:
@@ -85,7 +85,7 @@ async def close_redis() -> None:
         _sync_pool = None
     if _async_pool:
         try:
-            await _async_pool.disconnect()
+            _async_pool.disconnect()
             logger.info("Async Redis connection pool closed")
         except Exception as e:
             logger.error(f"Async Redis pool close failed: {e}")

app/core/tracing.py

@@ -7,7 +7,6 @@ import uuid
 from fastapi import FastAPI, Request
 
 TRACING_ENABLED = os.getenv("OTEL_ENABLED", "false").lower() == "true"
-OTEL_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")
 
 
 def setup_tracing(app: FastAPI) -> list:
@@ -34,49 +33,6 @@ def setup_tracing(app: FastAPI) -> list:
         return {"traces": [], "note": "OpenTelemetry export to Grafana Tempo configured. Traces available at :3000."}
 
 
-def setup_otel() -> bool:
-    """Initialize OpenTelemetry tracing. Returns True on success.
-
-    No-op if OTEL_ENABLED != "true". If OTEL_ENABLED is true but the SDK
-    is not installed, logs and returns False.
-    """
-    if not TRACING_ENABLED:
-        return False
-    try:
-        # Lazy import — opentelemetry is an optional dependency.
-        from opentelemetry import trace  # noqa: F401
-        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter  # noqa: F401
-        from opentelemetry.sdk.resources import Resource  # noqa: F401
-        from opentelemetry.sdk.trace import TracerProvider  # noqa: F401
-        from opentelemetry.sdk.trace.export import BatchSpanProcessor  # noqa: F401
-
-        resource = Resource.create({"service.name": "rmi-backend"})
-        provider = TracerProvider(resource=resource)
-        if OTEL_ENDPOINT:
-            exporter = OTLPSpanExporter(endpoint=OTEL_ENDPOINT)
-            provider.add_span_processor(BatchSpanProcessor(exporter))
-        trace.set_tracer_provider(provider)
-        return True
-    except ImportError:
-        return False
-    except Exception:
-        return False
-
-
-def shutdown_otel() -> None:
-    """Flush and shut down OpenTelemetry tracer provider."""
-    try:
-        from opentelemetry import trace
-
-        provider = trace.get_tracer_provider()
-        if hasattr(provider, "shutdown"):
-            provider.shutdown()
-    except ImportError:
-        pass
-    except Exception:
-        pass
-
-
 def start_span(name: str, attributes: dict | None = None) -> dict:
     return {"name": name, "start": time.perf_counter(), "attrs": attributes or {}}
 

app/databus/__init__.py

@@ -1,23 +1,69 @@
-"""databus - DEPRECATED shim. Use app.domains.databus.
-
-Phase 4.5 of AUDIT-2026-Q3.md moved app/databus/ to app.domains.databus/.
-This shim re-exports the canonical public surface so legacy imports like
-`from app.databus import databus` keep working.
-
-MIGRATION: replace `from app.databus import ...` with
-`from app.domains.databus import ...`.
 """
-from app.domains.databus import *  # noqa: F403
-from app.domains.databus import (  # noqa: F401
-    AccessController,
-    ConsumerType,
-    DataBus,
-    KeyAffinitySelector,
-    SchemaValidator,
-    SocialDataAggregator,
-    XTwitterProvider,
-    access_controller,
-    databus,
-    key_affinity,
-    schema_validator,
-)
+RMI DataBus - The Single Source of Truth for ALL Data
+=====================================================
+
+Every API call, MCP tool, x402 tool, scanner, and frontend hook routes through here.
+No raw HTTP calls to external APIs anywhere else. Period.
+
+Architecture (request flow):
+  Request → SecurityGate → CreditGate → CacheLayer → ProviderChain → Result
+        ↓                      ↓             ↓              ↓            ↓
+   admin check         free-first logic   L1→L2→L3      local-first!   RAG index
+   vault keys          auto-rotate      Redis+R2      OUR data 1st    WS broadcast
+
+What it REPLACES (do NOT use these anymore):
+  - app/cache_manager.py (RMICache)          → databus.cache
+  - app/caching_shield/unified_layer.py       → databus.fetch()
+  - app/caching_shield/data_fallback.py       → databus.fetch()
+  - app/caching_shield/api_registry.py        → databus.key_pool
+  - app/caching_shield/rate_limiter.py         → databus.rate_limiter
+  - app/arkham_connector.py                    → databus.providers.arkham
+  - app/coingecko_connector.py                → databus.providers.coingecko
+  - Direct .env reads for API keys            → databus.vault (encrypted in-memory)
+
+OWN DATA FIRST:
+  Our crown jewels - Wallet Memory Bank, RAG (17K docs), SENTINEL scanner,
+  Consensus RPC, Funding Tracer, ClickHouse, Price Consensus, News Network,
+  Bundle Detection, Label Import (169K+) - these are FAST, FREE, and OURS.
+  They go FIRST in every fallback chain. External APIs only augment or fill gaps.
+
+Usage:
+    from app.databus import databus
+
+    # Simple fetch - auto-selects best provider chain
+    result = await databus.fetch("token_price", mint="So11111111111111111111111111111111111111111")
+
+    # Explicit chain override
+    result = await databus.fetch("wallet_labels", address="7EcD...",
+                                  chain="local_first")
+
+    # Admin-only data (requires admin key in request)
+    result = await databus.fetch("arkham_entity", address="0x...",
+                                  admin_key="...")
+
+    # Check system health
+    health = await databus.health()
+
+    # Get capacity report (credits, rate limits, recommendations)
+    report = databus.capacity_report()
+"""
+
+from app.databus.access_control import AccessController, ConsumerType, access_controller
+from app.databus.core import DataBus, databus
+from app.databus.key_affinity import KeyAffinitySelector, key_affinity
+from app.databus.response_schema import SchemaValidator, schema_validator
+from app.databus.social import SocialDataAggregator, XTwitterProvider
+
+__all__ = [
+    "AccessController",
+    "ConsumerType",
+    "DataBus",
+    "KeyAffinitySelector",
+    "SchemaValidator",
+    "SocialDataAggregator",
+    "XTwitterProvider",
+    "access_controller",
+    "databus",
+    "key_affinity",
+    "schema_validator",
+]

app/databus/arkham_ws.py

@@ -97,7 +97,7 @@ async def broadcast_ws_update(address: str, update: dict):
 
     # Push to DataBus WebSocket for frontend subscribers
     try:
-        from app.domains.databus.ws_stream import ws_manager
+        from app.databus.ws_stream import ws_manager
 
         await ws_manager.broadcast(
             "arkham_realtime",

app/databus/bitquery_provider.py

@@ -36,7 +36,7 @@ import time
 
 import httpx
 
-from app.domains.databus.cache import CacheLayer, get_cache
+from app.databus.cache import CacheLayer, get_cache
 
 logger = logging.getLogger("databus.bitquery")
 

app/databus/core.py

@@ -32,11 +32,11 @@ import time
 from collections import defaultdict
 from typing import Any
 
-from app.domains.databus.access_control import access_controller
-from app.domains.databus.cache import CacheLayer, get_cache
-from app.domains.databus.providers import ProviderChain, ProviderTier, build_provider_chains
-from app.domains.databus.security import security
-from app.domains.databus.vault import DataBusVault, get_vault
+from app.databus.access_control import access_controller
+from app.databus.cache import CacheLayer, get_cache
+from app.databus.providers import ProviderChain, ProviderTier, build_provider_chains
+from app.databus.security import security
+from app.databus.vault import DataBusVault, get_vault
 
 logger = logging.getLogger("databus.core")
 
@@ -223,7 +223,7 @@ class DataBus:
             self.chains = build_provider_chains()
             # Register Bitquery provider for blockchain data
             try:
-                from app.domains.databus.bitquery_provider import BitqueryProvider
+                from app.databus.bitquery_provider import BitqueryProvider
 
                 self.bitquery = BitqueryProvider(self.cache)
                 logger.info("Bitquery provider registered")
@@ -485,7 +485,7 @@ class DataBus:
             "market_movers",
         ):
             try:
-                from app.domains.databus.ws_stream import databus_ws_broadcast
+                from app.databus.ws_stream import databus_ws_broadcast
 
                 await databus_ws_broadcast(data_type, result)
             except Exception:

app/databus/daily_intel.py

@@ -89,7 +89,7 @@ async def _gather_all_data() -> dict:
 
     # Market data
     try:
-        from app.domains.databus.news_provider import get_market_brief
+        from app.databus.news_provider import get_market_brief
 
         data["market"] = await get_market_brief()
     except Exception:
@@ -97,7 +97,7 @@ async def _gather_all_data() -> dict:
 
     # News intel
     try:
-        from app.domains.databus.news_intel import aggregate_all_news
+        from app.databus.news_intel import aggregate_all_news
 
         data["news"] = await aggregate_all_news(limit=20)
     except Exception:
@@ -105,7 +105,7 @@ async def _gather_all_data() -> dict:
 
     # CT Rundown
     try:
-        from app.domains.databus.x_intel import fetch_ct_rundown
+        from app.databus.x_intel import fetch_ct_rundown
 
         data["ct"] = await fetch_ct_rundown(limit=15)
     except Exception:
@@ -113,7 +113,7 @@ async def _gather_all_data() -> dict:
 
     # Social metrics
     try:
-        from app.domains.databus.social_intel import get_social_metrics
+        from app.databus.social_intel import get_social_metrics
 
         data["social"] = await get_social_metrics()
     except Exception:
@@ -121,7 +121,7 @@ async def _gather_all_data() -> dict:
 
     # Fear & Greed
     try:
-        from app.domains.databus.news_provider import get_fear_greed
+        from app.databus.news_provider import get_fear_greed
 
         data["fear_greed"] = await get_fear_greed()
     except Exception:
@@ -129,7 +129,7 @@ async def _gather_all_data() -> dict:
 
     # Prediction markets
     try:
-        from app.domains.databus.news_provider import get_prediction_markets
+        from app.databus.news_provider import get_prediction_markets
 
         data["prediction_markets"] = await get_prediction_markets(limit=5)
     except Exception:
@@ -234,7 +234,7 @@ async def generate_daily_intel(publish: bool = False, **kw) -> dict | None:
     Args:
         publish: If True, publish to Ghost (primary) + X/Telegram (syndication)
     """
-    from app.domains.databus.model_registry import ai_call, review_content
+    from app.databus.model_registry import ai_call, review_content
 
     # ── PHASE 0: Gather all data ──
     logger.info("Daily Intel: gathering data...")

app/databus/data_quality.py

@@ -570,7 +570,7 @@ async def enhanced_token_report(address: str, chain: str = "solana", **kw) -> di
 
     # Get the base token report
     try:
-        from app.domains.databus.rugcharts_intel import instant_token_report
+        from app.databus.rugcharts_intel import instant_token_report
 
         base_report = await instant_token_report(address, chain, **kw)
     except Exception as e:

app/databus/ohlcv_engine.py

@@ -340,7 +340,7 @@ async def fetch_ohlcv(
     # Also compute authenticity if we have volume data
     authenticity = None
     try:
-        from app.domains.databus.volume_authenticity import quick_authenticity_score
+        from app.databus.volume_authenticity import quick_authenticity_score
 
         auth = quick_authenticity_score(
             volume_24h=summary.get("volume_24h", 0),

app/databus/provider_implementations.py

@@ -183,10 +183,7 @@ async def _passthrough_news(**kwargs) -> dict | None:
 async def _passthrough_alerts(**kwargs) -> dict | None:
     """Alerts from our real alert pipeline."""
     try:
-        from app.domains.intelligence.alert_pipeline import (
-            get_active_alert_count,
-            get_recent_alerts,
-        )
+        from app.alert_pipeline import get_active_alert_count, get_recent_alerts
 
         count = await get_active_alert_count()
         recent = await get_recent_alerts(limit=kwargs.get("limit", 20))

app/databus/rag_ingestion.py

@@ -26,7 +26,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 1. Index recent news articles ──
-        from app.domains.databus.news_intel import aggregate_all_news
+        from app.databus.news_intel import aggregate_all_news
 
         news = await aggregate_all_news(limit=100)
         news_docs = []
@@ -55,7 +55,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 2. Index CT Rundown ──
-        from app.domains.databus.x_intel import fetch_ct_rundown
+        from app.databus.x_intel import fetch_ct_rundown
 
         ct = await fetch_ct_rundown(limit=30)
         ct_docs = []
@@ -84,7 +84,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 3. Index market data snapshot ──
-        from app.domains.databus.news_provider import get_fear_greed, get_market_brief
+        from app.databus.news_provider import get_fear_greed, get_market_brief
 
         market = await get_market_brief()
         fear = await get_fear_greed()
@@ -107,7 +107,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 4. Index social metrics ──
-        from app.domains.databus.social_intel import get_social_metrics
+        from app.databus.social_intel import get_social_metrics
 
         social = await get_social_metrics()
         social_doc = {

app/databus/rag_provider.py

@@ -30,7 +30,7 @@ async def _rag_search_provider(**kwargs) -> dict | None:
         }
 
     try:
-        from app.domains.databus.rag_engine import ai_enrich, hybrid_search
+        from app.databus.rag_engine import ai_enrich, hybrid_search
 
         if isinstance(collections, str):
             collections = [c.strip() for c in collections.split(",") if c.strip()]

app/databus/router.py

@@ -20,8 +20,8 @@ import os
 
 from fastapi import APIRouter, HTTPException, Query, Request
 
-from app.domains.databus.core import databus
-from app.domains.databus.social import (
+from app.databus.core import databus
+from app.databus.social import (
     X_DAILY_READ_BUDGET,
     X_FREE_MONTHLY_READ_LIMIT,
 )
@@ -133,11 +133,11 @@ async def databus_chains():
 async def databus_access_matrix(request: Request):
     """View the full access control matrix. Admin key required."""
     if not _verify_admin(request):
-        from app.domains.databus.access_control import access_controller
+        from app.databus.access_control import access_controller
 
         # Non-admin sees a limited view - only their tier's access
         return {"note": "Full matrix requires admin key. Your access depends on your consumer type."}
-    from app.domains.databus.access_control import access_controller
+    from app.databus.access_control import access_controller
 
     return access_controller.list_access_matrix()
 
@@ -145,7 +145,7 @@ async def databus_access_matrix(request: Request):
 @router.get("/mcp-scope/{tool_id}")
 async def databus_mcp_scope(tool_id: str):
     """Get the data types an MCP tool is authorized to access."""
-    from app.domains.databus.access_control import access_controller
+    from app.databus.access_control import access_controller
 
     allowed = access_controller.get_mcp_allowed_types(tool_id)
     if not allowed:
@@ -156,7 +156,7 @@ async def databus_mcp_scope(tool_id: str):
 @router.get("/x402-scope/{tier}")
 async def databus_x402_scope(tier: str):
     """Get the data types an x402 pricing tier can access."""
-    from app.domains.databus.access_control import access_controller
+    from app.databus.access_control import access_controller
 
     allowed = access_controller.get_x402_allowed_types(tier)
     return {"tier": tier, "allowed_types": allowed}
@@ -465,7 +465,7 @@ async def databus_schema_validate(request: Request):
     body = await request.json()
     data_type = body.get("data_type", "")
     data = body.get("data", {})
-    from app.domains.databus.response_schema import schema_validator
+    from app.databus.response_schema import schema_validator
 
     is_valid, missing = schema_validator.validate(data_type, data)
     return {"data_type": data_type, "valid": is_valid, "missing_fields": missing}
@@ -477,7 +477,7 @@ async def databus_schema_validate(request: Request):
 @router.get("/social/x/profile/{username}")
 async def social_x_profile(username: str):
     """Get X/Twitter user profile - cached 24h."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.x.get_user(username)
@@ -489,7 +489,7 @@ async def social_x_profile(username: str):
 @router.get("/social/x/tweets/{username}")
 async def social_x_tweets(username: str, count: int = Query(20, ge=1, le=100)):
     """Get recent tweets from user - cached 15min."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.get_our_tweets(count=count) if username.lower() == "cryptorugmunch" else None
@@ -506,7 +506,7 @@ async def social_x_tweets(username: str, count: int = Query(20, ge=1, le=100)):
 @router.get("/social/x/mentions")
 async def social_x_mentions(count: int = Query(20, ge=1, le=100)):
     """Get mentions of @CryptoRugMunch - cached 15min."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.get_our_mentions(count=count)
@@ -518,7 +518,7 @@ async def social_x_mentions(count: int = Query(20, ge=1, le=100)):
 @router.get("/social/x/engagement")
 async def social_x_engagement(tweet_ids: str = Query(..., description="Comma-separated tweet IDs")):
     """Get engagement metrics for tweets - cached 1h."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     ids = [tid.strip() for tid in tweet_ids.split(",")[:100]]
@@ -529,7 +529,7 @@ async def social_x_engagement(tweet_ids: str = Query(..., description="Comma-sep
 @router.get("/social/x/search")
 async def social_x_search(q: str = Query(..., description="Search query"), count: int = Query(10, ge=1, le=100)):
     """Search X/Twitter - VERY expensive, cached 24h. Use sparingly."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.search_mentions(q, count=count)
@@ -541,7 +541,7 @@ async def social_x_search(q: str = Query(..., description="Search query"), count
 @router.get("/social/kol/{username}")
 async def social_kol_reputation(username: str):
     """KOL reputation score - cached 24h."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     return await agg.get_kol_reputation(username)
@@ -550,7 +550,7 @@ async def social_kol_reputation(username: str):
 @router.get("/social/sentiment")
 async def social_sentiment(username: str = Query("CryptoRugMunch")):
     """Brand sentiment analysis - cached 1h."""
-    from app.domains.databus.social import SocialDataAggregator
+    from app.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     return await agg.get_sentiment(username)
@@ -559,7 +559,7 @@ async def social_sentiment(username: str = Query("CryptoRugMunch")):
 @router.get("/social/x/budget")
 async def social_x_budget():
     """Current X API read budget status."""
-    from app.domains.databus.social import XTwitterProvider
+    from app.databus.social import XTwitterProvider
 
     provider = XTwitterProvider(databus.cache)
     return {
@@ -577,7 +577,7 @@ X_HANDLE = "CryptoRugMunch"
 @router.get("/social/x/discover")
 async def social_x_discover(handle: str = Query(X_HANDLE), limit: int = Query(50, ge=1, le=100)):
     """Discover tweets via web search - no API needed, works for free."""
-    from app.domains.databus.social_scraper import XWebScraper
+    from app.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     tweets = await scraper.discover_tweets(handle=handle, limit=limit)
@@ -587,7 +587,7 @@ async def social_x_discover(handle: str = Query(X_HANDLE), limit: int = Query(50
 @router.get("/social/x/engagement-report")
 async def social_x_engagement_report(handle: str = Query(X_HANDLE)):
     """Get engagement report for a handle - avg likes, best tweets, posting frequency."""
-    from app.domains.databus.social_scraper import XWebScraper
+    from app.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     report = await scraper.get_engagement_report(handle=handle)
@@ -597,7 +597,7 @@ async def social_x_engagement_report(handle: str = Query(X_HANDLE)):
 @router.get("/social/x/mentions-discover")
 async def social_x_mentions_discover(handle: str = Query(X_HANDLE), limit: int = Query(20, ge=1, le=50)):
     """Find tweets mentioning a handle - via web search."""
-    from app.domains.databus.social_scraper import XWebScraper
+    from app.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     mentions = await scraper.get_mentions(handle=handle, limit=limit)
@@ -607,7 +607,7 @@ async def social_x_mentions_discover(handle: str = Query(X_HANDLE), limit: int =
 @router.get("/social/x/trending")
 async def social_x_trending():
     """Get current crypto trending topics from web search."""
-    from app.domains.databus.social_scraper import XWebScraper
+    from app.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     topics = await scraper.get_trending_topics()
@@ -620,7 +620,7 @@ async def social_x_trending():
 @router.get("/bitquery/health")
 async def bitquery_health():
     """Check Bitquery provider health and billing status."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     return await provider.health()
@@ -629,7 +629,7 @@ async def bitquery_health():
 @router.get("/bitquery/token-price/{network}/{token_address}")
 async def bitquery_token_price(network: str, token_address: str):
     """Get DEX token price from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_token_price(network, token_address)
@@ -641,7 +641,7 @@ async def bitquery_token_price(network: str, token_address: str):
 @router.get("/bitquery/holders/{network}/{token_address}")
 async def bitquery_holders(network: str, token_address: str, limit: int = Query(100, ge=1, le=500)):
     """Get token holder distribution from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_holder_distribution(network, token_address, limit)
@@ -653,7 +653,7 @@ async def bitquery_holders(network: str, token_address: str, limit: int = Query(
 @router.get("/bitquery/tx-trace/{network}/{tx_hash}")
 async def bitquery_tx_trace(network: str, tx_hash: str):
     """Get full transaction trace from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_transaction_trace(network, tx_hash)
@@ -665,7 +665,7 @@ async def bitquery_tx_trace(network: str, tx_hash: str):
 @router.get("/bitquery/dex-volume/{network}")
 async def bitquery_dex_volume(network: str, pool: str = Query(None), timeframe: str = Query("24h")):
     """Get DEX trading volume from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_dex_volume(network, pool, timeframe)
@@ -683,7 +683,7 @@ async def bitquery_dex_volume(network: str, pool: str = Query(None), timeframe:
 @router.get("/bitquery/balance/{network}/{address}")
 async def bitquery_balance(network: str, address: str):
     """Get address token balances from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_address_balance(network, address)
@@ -695,7 +695,7 @@ async def bitquery_balance(network: str, address: str):
 @router.get("/bitquery/cross-chain/{address}")
 async def bitquery_cross_chain(address: str, networks: str = Query("ethereum,bsc,solana")):
     """Track token transfers across chains from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     net_list = [n.strip() for n in networks.split(",")]
@@ -710,7 +710,7 @@ async def bitquery_contract_events(
     network: str, contract: str, event: str = Query(None), limit: int = Query(50, ge=1, le=200)
 ):
     """Get smart contract events from Bitquery."""
-    from app.domains.databus.bitquery_provider import BitqueryProvider
+    from app.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_smart_contract_events(network, contract, event, limit)
@@ -734,7 +734,7 @@ async def receive_webhook(service: str, request: Request):
     indexes in RAG, triggers premium scanner, pushes alerts.
     """
     try:
-        from app.domains.databus.webhooks import handle_webhook
+        from app.databus.webhooks import handle_webhook
     except ImportError:
         raise HTTPException(501, "Webhook system not available") from None
 
@@ -758,7 +758,7 @@ async def receive_webhook(service: str, request: Request):
 async def list_webhooks_endpoint():
     """List all recent webhook events."""
     try:
-        from app.domains.databus.webhooks import list_webhooks
+        from app.databus.webhooks import list_webhooks
 
         return await list_webhooks()
     except ImportError:
@@ -777,7 +777,7 @@ async def setup_webhook_endpoint(service: str, request: Request):
     }
     """
     try:
-        from app.domains.databus.webhooks import setup_webhook
+        from app.databus.webhooks import setup_webhook
     except ImportError:
         raise HTTPException(501, "Webhook system not available") from None
 

app/databus/rugcharts_intel.py

@@ -1198,7 +1198,7 @@ async def instant_token_report(address: str = "", chain: str = "solana", **kw) -
     try:
         # ── Section 1: Security Scan ──
         try:
-            from app.domains.databus.token_security import run_full_scan
+            from app.databus.token_security import run_full_scan
 
             security = await run_full_scan(address, chain, api_key=api_key)
             if security:
@@ -1288,7 +1288,7 @@ async def instant_token_report(address: str = "", chain: str = "solana", **kw) -
             liq = float(kw.get("liquidity_usd", 0))
             uw = int(kw.get("unique_wallets", 0))
             if vol_24h > 0 or liq > 0:
-                from app.domains.databus.volume_authenticity import quick_authenticity_score
+                from app.databus.volume_authenticity import quick_authenticity_score
 
                 auth = quick_authenticity_score(vol_24h, liq, uw, 0)
                 report["sections"]["volume"] = {

app/databus/social.py

@@ -20,7 +20,7 @@ from datetime import UTC, datetime
 
 import httpx
 
-from app.domains.databus.cache import CacheLayer
+from app.databus.cache import CacheLayer
 
 logger = logging.getLogger("databus.social")
 

app/databus/social_intel.py

@@ -295,21 +295,21 @@ async def generate_daily_intel(**kw) -> dict:
     """
     # Gather all data sources
     try:
-        from app.domains.databus.news_provider import get_market_brief
+        from app.databus.news_provider import get_market_brief
 
         brief = await get_market_brief()
     except Exception:
         brief = {}
 
     try:
-        from app.domains.databus.news_intel import aggregate_all_news
+        from app.databus.news_intel import aggregate_all_news
 
         news = await aggregate_all_news(limit=15)
     except Exception:
         news = {"articles": []}
 
     try:
-        from app.domains.databus.x_intel import fetch_ct_rundown
+        from app.databus.x_intel import fetch_ct_rundown
 
         ct = await fetch_ct_rundown(limit=10)
     except Exception:
@@ -411,7 +411,7 @@ async def get_social_metrics(**kw) -> dict:
     """
     # Gather data
     try:
-        from app.domains.databus.news_intel import aggregate_all_news
+        from app.databus.news_intel import aggregate_all_news
 
         news = await aggregate_all_news(limit=50)
     except Exception:

app/databus/social_scraper.py

@@ -16,7 +16,7 @@ Cron: Every 6 hours, discover new tweets, extract content, update metrics.
 import logging
 from datetime import UTC, datetime
 
-from app.domains.databus.cache import CacheLayer, get_cache
+from app.databus.cache import CacheLayer, get_cache
 
 logger = logging.getLogger("databus.social_scraper")
 

app/databus/spl_metadata_decoder.py

@@ -257,6 +257,7 @@ async def fetch_metaplex_metadata(mint: str, rpc_url: str) -> dict[str, Any] | N
         # Check if it's off the ed25519 curve (valid PDA)
         # For simplicity, we'll just use the first one that doesn't have the high bit set
         # Actually, proper PDA derivation is complex. Let's use a known endpoint or skip if too complex.
+        pass
 
     # Simpler approach: use getProgramAccounts with filters
     payload = {

app/databus/token_security.py

@@ -613,7 +613,7 @@ async def run_quick_scan(address: str, chain: str = "ethereum", **kw) -> dict[st
         tx_count = int(kw.get("tx_count", 0))
 
         if volume_24h > 0 or liquidity > 0:
-            from app.domains.databus.volume_authenticity import quick_authenticity_score
+            from app.databus.volume_authenticity import quick_authenticity_score
 
             auth = quick_authenticity_score(volume_24h, liquidity, unique_wallets, tx_count)
             fake_pct = auth.get("fake_volume_pct", 0)

app/databus/webhooks.py

@@ -152,7 +152,7 @@ async def handle_webhook(service: str, payload: dict, headers: dict, raw_body: b
     risk = _assess_webhook_risk(service, event_type, payload)
     if risk["level"] in ("HIGH", "CRITICAL"):
         try:
-            from app.domains.intelligence.alert_pipeline import push_alert
+            from app.alert_pipeline import push_alert
 
             await push_alert(
                 title=f"[{service.upper()}] {risk['title']}",

app/degen_security_scanner.py

@@ -1,4 +1,4 @@
-import httpx
+from app.telegram_bot.requirements import httpx
 
 #!/usr/bin/env python3
 """

app/domain/__init__.py

@@ -0,0 +1,25 @@
+"""domain - DEPRECATED shim. Use app.domains.
+
+Phase 4.7 of AUDIT-2026-Q3.md renamed app/domain/ to app/domains/.
+This shim re-exports for legacy callers.
+"""
+import sys as _sys
+import importlib as _importlib
+
+import app.domains as _new
+_sys.modules['app.domain'] = _new
+
+_sys.modules['app.domain.alerts'] = _importlib.import_module('app.domains.alerts')
+_sys.modules['app.domain.auth'] = _importlib.import_module('app.domains.auth')
+_sys.modules['app.domain.billing'] = _importlib.import_module('app.domains.billing')
+_sys.modules['app.domain.databus'] = _importlib.import_module('app.domains.databus')
+_sys.modules['app.domain.labels'] = _importlib.import_module('app.domains.labels')
+_sys.modules['app.domain.news'] = _importlib.import_module('app.domains.news')
+_sys.modules['app.domain.reports'] = _importlib.import_module('app.domains.reports')
+_sys.modules['app.domain.scanner'] = _importlib.import_module('app.domains.scanner')
+_sys.modules['app.domain.telegram'] = _importlib.import_module('app.domains.telegram')
+_sys.modules['app.domain.threat'] = _importlib.import_module('app.domains.threat')
+_sys.modules['app.domain.token'] = _importlib.import_module('app.domains.token')
+_sys.modules['app.domain.tokens'] = _importlib.import_module('app.domains.tokens')
+_sys.modules['app.domain.wallet'] = _importlib.import_module('app.domains.wallet')
+_sys.modules['app.domain.x402'] = _importlib.import_module('app.domains.x402')

app/domains/admin/__init__.py

@@ -1,31 +0,0 @@
-"""Admin domain - public API.
-
-Phase 4 domain consolidation: app.admin_backend -> app.domains.admin.
-"""
-from __future__ import annotations
-
-from app.domains.admin.core import (
-    PERMISSIONS,
-    AdminRole,
-    AdminUserStore,
-    AuditLogEntry,
-    AuditLogger,
-    SecurityManager,
-    SessionManager,
-    SystemHealthMonitor,
-    has_permission,
-    require_admin,
-)
-
-__all__ = [
-    "PERMISSIONS",
-    "AdminRole",
-    "AdminUserStore",
-    "AuditLogEntry",
-    "AuditLogger",
-    "SecurityManager",
-    "SessionManager",
-    "SystemHealthMonitor",
-    "has_permission",
-    "require_admin",
-]

app/domains/auth/deps.py

@@ -1,61 +1,12 @@
-"""FastAPI dependencies for auth domain.
+"""FastAPI dependencies - re-exports from app.domains.auth.
 
 Phase 3B of AUDIT-2026-Q3.md.
 
 Public API:
-- get_current_user
-- require_auth
-- require_public_profile
+- get_current_user, require_auth, require_public_profile
 """
-from __future__ import annotations
-
-from typing import Any
-
-from fastapi import HTTPException, Request
-
-from app.domains.auth.jwt import _verify_jwt
-from app.domains.auth.store import _get_user
-
-
-async def get_current_user(request: Request) -> dict[str, Any] | None:
-    """Verify JWT from Authorization header (wallet or email)."""
-    auth_header = request.headers.get("Authorization", "")
-    if not auth_header.startswith("Bearer "):
-        return None
-    token = auth_header[7:]
-    payload = _verify_jwt(token)
-    if not payload:
-        return None
-    user = _get_user(payload["id"])
-    if not user:
-        return None
-    return user
-
-
-async def require_auth(request: Request) -> dict[str, Any]:
-    """Require valid JWT. Raises 401 if missing/invalid."""
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-    return user
-
-
-async def require_public_profile(request: Request) -> dict[str, Any]:
-    """
-    Dependency to ensure the user has a public profile.
-    Required for actions that interact with the community (posting, commenting).
-    """
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-
-    privacy_settings = user.get("privacy_settings", {})
-    visibility = privacy_settings.get("profile_visibility", "public")
-
-    if visibility != "public":
-        raise HTTPException(
-            status_code=403,
-            detail="A public profile is required to post or comment. Please update your privacy settings in your profile.",
-        )
-
-    return user
+from app.auth import (  # noqa: F401
+    get_current_user,
+    require_auth,
+    require_public_profile,
+)

app/domains/auth/jwt.py

@@ -1,66 +1,18 @@
-"""JWT helpers.
+"""JWT helpers - re-exports from app.auth for cleaner import paths.
 
 Phase 3B of AUDIT-2026-Q3.md.
 
 Public API:
 - JWT_SECRET, JWT_ALGORITHM, JWT_EXPIRY_DAYS
 - _create_jwt, _verify_jwt
-- generate_nonce, _derive_user_id
+- generate_nonce
 """
-
-from __future__ import annotations
-
-import hashlib
-import os
-import secrets
-from datetime import datetime, timedelta
-from typing import Any
-
-from jose import JWTError, jwt
-
-JWT_SECRET = os.getenv("JWT_SECRET", "rmi-jwt-secret-change-me")
-JWT_ALGORITHM = "HS256"
-JWT_EXPIRY_DAYS = 7
-
-
-def _create_jwt(
-    user_id: str,
-    email: str,
-    tier: str = "FREE",
-    role: str = "USER",
-    wallet: str | None = None,
-) -> str:
-    now = datetime.utcnow()
-    payload = {
-        "sub": user_id,
-        "email": email,
-        "tier": tier,
-        "role": role,
-        "iat": now,
-        "exp": now + timedelta(days=JWT_EXPIRY_DAYS),
-    }
-    if wallet:
-        payload["wallet"] = wallet
-    return str(jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM))
-
-
-def _verify_jwt(token: str) -> dict[str, Any] | None:
-    try:
-        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
-        return {
-            "id": payload["sub"],
-            "email": payload["email"],
-            "tier": payload.get("tier", "FREE"),
-            "role": payload.get("role", "USER"),
-            "wallet": payload.get("wallet"),
-        }
-    except JWTError:
-        return None
-
-
-def _derive_user_id(email: str) -> str:
-    return hashlib.sha256(email.lower().encode()).hexdigest()[:32]
-
-
-def generate_nonce() -> str:
-    return secrets.token_urlsafe(16)
+from app.auth import (  # noqa: F401
+    JWT_ALGORITHM,
+    JWT_EXPIRY_DAYS,
+    JWT_SECRET,
+    _create_jwt,
+    _derive_user_id,
+    _verify_jwt,
+    generate_nonce,
+)

app/domains/auth/oauth.py

@@ -1,420 +1,20 @@
-"""OAuth + Telegram auth endpoints.
+"""OAuth flows - re-exports from app.domains.auth.
 
 Phase 3B of AUDIT-2026-Q3.md.
+
+Public API:
+- google_auth_url, google_callback, google_callback_post
+- github_auth_url, github_callback
+- x_auth_url, x_callback
+- telegram_auth
 """
-from __future__ import annotations
-
-import json
-import os
-from datetime import datetime
-from typing import cast
-
-from fastapi import APIRouter, HTTPException, Request
-from fastapi.responses import RedirectResponse
-
-from app.core.redis import get_redis
-from app.domains.auth.jwt import _create_jwt, _derive_user_id
-from app.domains.auth.schemas import (
-    GoogleAuthResponse,
-    TelegramAuthRequest,
-    WalletAuthResponse,
+from app.auth import (  # noqa: F401
+    github_auth_url,
+    github_callback,
+    google_auth_url,
+    google_callback,
+    google_callback_post,
+    telegram_auth,
+    x_auth_url,
+    x_callback,
 )
-from app.domains.auth.store import _get_user, _get_user_by_email
-
-router = APIRouter(tags=["auth"])
-
-FRONTEND_URL = os.getenv("FRONTEND_URL", "https://rugmunch.io")
-
-
-@router.get("/google/url", response_model=GoogleAuthResponse)
-async def google_auth_url() -> GoogleAuthResponse:
-    """Get Google OAuth URL."""
-    client_id = os.getenv("GOOGLE_CLIENT_ID")
-    redirect_uri = os.getenv(
-        "GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
-    )
-
-    if not client_id:
-        return cast("GoogleAuthResponse", {"url": "/auth/callback?error=google_not_configured"})
-
-    from urllib.parse import urlencode
-
-    params = {
-        "client_id": client_id,
-        "redirect_uri": redirect_uri,
-        "response_type": "code",
-        "scope": "openid email profile",
-        "access_type": "offline",
-        "prompt": "consent",
-    }
-    url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(params)}"
-    return cast("GoogleAuthResponse", {"url": url})
-
-
-@router.get("/google/callback")
-async def google_callback(request: Request) -> RedirectResponse:
-    """Handle Google OAuth redirect - exchanges code, creates user, redirects to frontend with token."""
-    code = request.query_params.get("code")
-    error = request.query_params.get("error")
-
-    if error:
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error={error}")
-
-    if not code:
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=missing_code")
-
-    import httpx
-
-    client_id = os.getenv("GOOGLE_CLIENT_ID")
-    client_secret = os.getenv("GOOGLE_CLIENT_SECRET")
-    redirect_uri = os.getenv(
-        "GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
-    )
-
-    if not client_id or not client_secret:
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=not_configured")
-
-    async with httpx.AsyncClient() as client:
-        resp = await client.post(
-            "https://oauth2.googleapis.com/token",
-            data={
-                "code": code,
-                "client_id": client_id,
-                "client_secret": client_secret,
-                "redirect_uri": redirect_uri,
-                "grant_type": "authorization_code",
-            },
-        )
-        if resp.status_code != 200:
-            return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=token_exchange_failed")
-
-        tokens = resp.json()
-        access_token = tokens.get("access_token")
-
-        resp = await client.get(
-            "https://www.googleapis.com/oauth2/v2/userinfo",
-            headers={"Authorization": f"Bearer {access_token}"},
-        )
-        if resp.status_code != 200:
-            return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=userinfo_failed")
-
-        google_user = resp.json()
-        email = google_user.get("email")
-
-        user = _get_user_by_email(email)
-        if not user:
-            user_id = _derive_user_id(email)
-            user = {
-                "id": user_id,
-                "email": email,
-                "display_name": google_user.get("name", email),
-                "tier": "FREE",
-                "role": "USER",
-                "created_at": datetime.utcnow().isoformat(),
-                "xp": 0,
-                "level": 1,
-                "badges": [],
-                "scans_remaining": 5,
-                "scans_used": 0,
-            }
-            r = get_redis()
-            assert r is not None
-            r.hset("rmi:users", user_id, json.dumps(user))
-            r.hset("rmi:users:email", email.lower(), user_id)
-
-        jwt_token = _create_jwt(
-            user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
-        )
-
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=google")
-
-
-@router.post("/google/callback", response_model=WalletAuthResponse)
-async def google_callback_post(request: Request) -> WalletAuthResponse:
-    """Legacy POST endpoint for manual code exchange."""
-    body = await request.json()
-    code = body.get("code")
-
-    if not code:
-        raise HTTPException(status_code=400, detail="Missing authorization code")
-
-    import httpx
-
-    client_id = os.getenv("GOOGLE_CLIENT_ID")
-    client_secret = os.getenv("GOOGLE_CLIENT_SECRET")
-    redirect_uri = os.getenv(
-        "GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
-    )
-
-    if not client_id or not client_secret:
-        raise HTTPException(status_code=500, detail="Google OAuth not configured")
-
-    async with httpx.AsyncClient() as client:
-        resp = await client.post(
-            "https://oauth2.googleapis.com/token",
-            data={
-                "code": code,
-                "client_id": client_id,
-                "client_secret": client_secret,
-                "redirect_uri": redirect_uri,
-                "grant_type": "authorization_code",
-            },
-        )
-        if resp.status_code != 200:
-            raise HTTPException(status_code=400, detail="Failed to exchange code")
-
-        tokens = resp.json()
-        access_token = tokens.get("access_token")
-
-        resp = await client.get(
-            "https://www.googleapis.com/oauth2/v2/userinfo",
-            headers={"Authorization": f"Bearer {access_token}"},
-        )
-        if resp.status_code != 200:
-            raise HTTPException(status_code=400, detail="Failed to get user info")
-
-        google_user = resp.json()
-        email = google_user.get("email")
-
-        user = _get_user_by_email(email)
-        if not user:
-            user_id = _derive_user_id(email)
-            user = {
-                "id": user_id,
-                "email": email,
-                "display_name": google_user.get("name", email),
-                "tier": "FREE",
-                "role": "USER",
-                "created_at": datetime.utcnow().isoformat(),
-                "xp": 0,
-                "level": 1,
-                "badges": [],
-                "scans_remaining": 5,
-                "scans_used": 0,
-            }
-            r = get_redis()
-            assert r is not None
-            r.hset("rmi:users", user_id, json.dumps(user))
-            r.hset("rmi:users:email", email.lower(), user_id)
-
-        jwt_token = _create_jwt(
-            user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
-        )
-
-        return cast(
-            "WalletAuthResponse",
-            {
-                "access_token": jwt_token,
-                "refresh_token": jwt_token,
-                "user": {
-                    "id": user["id"],
-                    "email": user["email"],
-                    "display_name": user.get("display_name", email),
-                    "wallet_address": None,
-                    "wallet_chain": None,
-                    "tier": user.get("tier", "FREE"),
-                    "role": user.get("role", "USER"),
-                    "created_at": user.get("created_at"),
-                },
-            },
-        )
-
-
-@router.post("/telegram", response_model=WalletAuthResponse)
-async def telegram_auth(req: TelegramAuthRequest) -> WalletAuthResponse:
-    """Authenticate via Telegram Web App data."""
-    bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
-    if not bot_token:
-        raise HTTPException(status_code=500, detail="Telegram auth not configured")
-
-    data_check = {
-        "id": str(req.id),
-        "auth_date": str(req.auth_date),
-    }
-    if req.first_name:
-        data_check["first_name"] = req.first_name
-    if req.last_name:
-        data_check["last_name"] = req.last_name
-    if req.username:
-        data_check["username"] = req.username
-    if req.photo_url:
-        data_check["photo_url"] = req.photo_url
-
-    import hashlib
-    import hmac
-
-    sorted_data = "\n".join(f"{k}={v}" for k, v in sorted(data_check.items()))
-    secret = hashlib.sha256(bot_token.encode()).digest()
-    expected_hash = hmac.new(secret, sorted_data.encode(), hashlib.sha256).hexdigest()
-
-    if not hmac.compare_digest(req.hash, expected_hash):
-        raise HTTPException(status_code=401, detail="Invalid Telegram auth data")
-
-    telegram_user_id = f"tg:{req.id}"
-    user = _get_user(telegram_user_id)
-
-    if not user:
-        display_name = (
-            f"{req.first_name or ''} {req.last_name or ''}".strip()
-            or req.username
-            or f"User{req.id}"
-        )
-        email = f"{req.id}@telegram.rmi"
-
-        user = {
-            "id": telegram_user_id,
-            "email": email,
-            "display_name": display_name,
-            "telegram_id": req.id,
-            "telegram_username": req.username,
-            "tier": "FREE",
-            "role": "USER",
-            "created_at": datetime.utcnow().isoformat(),
-            "xp": 0,
-            "level": 1,
-            "badges": [],
-            "scans_remaining": 5,
-            "scans_used": 0,
-        }
-        r = get_redis()
-        assert r is not None
-        r.hset("rmi:users", telegram_user_id, json.dumps(user))
-        r.hset("rmi:users:telegram", str(req.id), telegram_user_id)
-
-    jwt_token = _create_jwt(
-        user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
-    )
-
-    return cast(
-        "WalletAuthResponse",
-        {
-            "access_token": jwt_token,
-            "refresh_token": jwt_token,
-            "user": {
-                "id": user["id"],
-                "email": user["email"],
-                "display_name": user.get("display_name"),
-                "wallet_address": None,
-                "wallet_chain": None,
-                "tier": user.get("tier", "FREE"),
-                "role": user.get("role", "USER"),
-                "created_at": user.get("created_at"),
-            },
-        },
-    )
-
-
-@router.get("/github/url", response_model=GoogleAuthResponse)
-async def github_auth_url() -> GoogleAuthResponse:
-    """Get GitHub OAuth URL."""
-    client_id = os.getenv("GITHUB_CLIENT_ID")
-    redirect_uri = os.getenv(
-        "GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback"
-    )
-
-    if not client_id:
-        return cast("GoogleAuthResponse", {"url": "/auth/callback?error=github_not_configured"})
-
-    from urllib.parse import urlencode
-
-    params = {
-        "client_id": client_id,
-        "redirect_uri": redirect_uri,
-        "scope": "user:email",
-    }
-    url = f"https://github.com/login/oauth/authorize?{urlencode(params)}"
-    return cast("GoogleAuthResponse", {"url": url})
-
-
-@router.get("/github/callback")
-async def github_callback(request: Request) -> RedirectResponse:
-    """Handle GitHub OAuth redirect - exchanges code, creates user, redirects to frontend with token."""
-    code = request.query_params.get("code")
-    error = request.query_params.get("error")
-
-    if error:
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error={error}")
-
-    if not code:
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=missing_code")
-
-    import httpx
-
-    client_id = os.getenv("GITHUB_CLIENT_ID")
-    client_secret = os.getenv("GITHUB_CLIENT_SECRET")
-    redirect_uri = os.getenv(
-        "GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback"
-    )
-
-    if not client_id or not client_secret:
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=not_configured")
-
-    async with httpx.AsyncClient() as client:
-        resp = await client.post(
-            "https://github.com/login/oauth/access_token",
-            data={
-                "client_id": client_id,
-                "client_secret": client_secret,
-                "code": code,
-                "redirect_uri": redirect_uri,
-            },
-            headers={"Accept": "application/json"},
-        )
-        if resp.status_code != 200:
-            return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=token_exchange_failed")
-
-        tokens = resp.json()
-        access_token = tokens.get("access_token")
-
-        resp = await client.get(
-            "https://api.github.com/user",
-            headers={"Authorization": f"token {access_token}", "Accept": "application/json"},
-        )
-        if resp.status_code != 200:
-            return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=userinfo_failed")
-
-        github_user = resp.json()
-        email = github_user.get("email")
-
-        if not email:
-            resp = await client.get(
-                "https://api.github.com/user/emails",
-                headers={"Authorization": f"token {access_token}", "Accept": "application/json"},
-            )
-            if resp.status_code == 200:
-                emails = resp.json()
-                primary = next((e for e in emails if e.get("primary")), None)
-                email = (
-                    primary.get("email")
-                    if primary
-                    else (emails[0].get("email") if emails else None)
-                )
-
-        if not email:
-            email = f"{github_user.get('login', 'github_user')}@github.rmi"
-
-        user = _get_user_by_email(email)
-        if not user:
-            user_id = _derive_user_id(email)
-            user = {
-                "id": user_id,
-                "email": email,
-                "display_name": github_user.get("name", github_user.get("login", email)),
-                "tier": "FREE",
-                "role": "USER",
-                "created_at": datetime.utcnow().isoformat(),
-                "xp": 0,
-                "level": 1,
-                "badges": [],
-                "scans_remaining": 5,
-                "scans_used": 0,
-            }
-            r = get_redis()
-            assert r is not None
-            r.hset("rmi:users", user_id, json.dumps(user))
-            r.hset("rmi:users:email", email.lower(), user_id)
-
-        jwt_token = _create_jwt(
-            user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
-        )
-        return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=github")

app/domains/auth/passwords.py

@@ -1,24 +1,11 @@
-"""Password helpers.
+"""Password helpers - re-exports from app.domains.auth.
 
 Phase 3B of AUDIT-2026-Q3.md.
 
 Public API:
 - hash_password, verify_password
 """
-
-from __future__ import annotations
-
-import bcrypt
-
-
-def hash_password(password: str) -> str:
-    """Hash password using bcrypt directly."""
-    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
-
-
-def verify_password(password: str, hashed: str) -> bool:
-    """Verify password against hash."""
-    try:
-        return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))
-    except Exception:
-        return False
+from app.auth import (  # noqa: F401
+    hash_password,
+    verify_password,
+)

app/domains/auth/router.py

@@ -1,509 +0,0 @@
-"""
-Auth Router - Complete authentication system (email, wallet, OAuth, Telegram)
-
-This is the FastAPI router for the auth domain. It is mounted by
-``app/mount.py`` via the re-export in ``app.domains.auth.router``.
-
-Phase 2.1: extracted from ``app/domains/auth/__init__.py`` so that the
-package marker is a clean re-export shim and mypy does not see the
-router file as both module and package.
-"""
-
-import json
-import logging
-from datetime import datetime, timedelta
-from typing import Any, cast
-
-from fastapi import APIRouter, HTTPException, Request
-
-from app.core.redis import get_redis
-from app.domains.auth.deps import (
-    get_current_user,
-)
-from app.domains.auth.jwt import (
-    JWT_ALGORITHM,
-    JWT_SECRET,
-    _create_jwt,
-    _derive_user_id,
-    _verify_jwt,
-    generate_nonce,
-)
-from app.domains.auth.oauth import router as oauth_router
-from app.domains.auth.passwords import hash_password, verify_password
-from app.domains.auth.schemas import (
-    EmailLoginRequest,
-    EmailRegisterRequest,
-    NonceResponse,
-    TwoFAEnableRequest,
-    TwoFALoginRequest,
-    TwoFASetupResponse,
-    TwoFAStatusResponse,
-    TwoFAVerifyRequest,
-    UserResponse,
-    WalletAuthResponse,
-    WalletNonceRequest,
-    WalletVerifyRequest,
-)
-from app.domains.auth.store import (
-    _get_user,
-    _get_user_by_email,
-    _is_valid_email,
-    _save_user,
-)
-from app.domains.auth.totp import (
-    PYOTP_AVAILABLE,
-    _generate_backup_codes,
-    _generate_qr_base64,
-    _generate_totp_secret,
-    _get_totp_uri,
-    _hash_backup_code,
-    _verify_backup_code,
-    _verify_totp,
-)
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter(tags=["auth"])
-
-# Include OAuth/Telegram routes from dedicated submodule
-router.include_router(oauth_router, tags=["auth"])
-
-# ── Config ──
-
-
-# ── Storage (Redis-backed user store) ──
-# ── Email Auth ──
-@router.post("/register", response_model=WalletAuthResponse)
-def register_email(req: EmailRegisterRequest) -> WalletAuthResponse:
-    """Register a new user with email/password."""
-    if not _is_valid_email(req.email):
-        raise HTTPException(status_code=400, detail="Invalid email format")
-
-    if len(req.password) < 8:
-        raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
-
-    # Check if user exists
-    existing = _get_user_by_email(req.email)
-    if existing:
-        raise HTTPException(status_code=400, detail="Email already registered")
-
-    user_id = _derive_user_id(req.email)
-    hashed = hash_password(req.password)
-
-    user: dict[str, Any] = {
-        "id": user_id,
-        "email": req.email,
-        "display_name": req.display_name,
-        "password_hash": hashed,
-        "wallet_address": req.wallet_address,
-        "wallet_chain": req.wallet_chain,
-        "tier": "FREE",
-        "role": "USER",
-        "created_at": datetime.utcnow().isoformat(),
-        "xp": 0,
-        "level": 1,
-        "badges": [],
-        "scans_remaining": 5,
-        "scans_used": 0,
-    }
-
-    # Save user + email index
-    r = get_redis()
-    assert r is not None
-    r.hset("rmi:users", user_id, json.dumps(user))
-    r.hset("rmi:users:email", req.email.lower(), user_id)
-
-    token = _create_jwt(user_id, req.email, "FREE", "USER", req.wallet_address)
-
-    return cast(
-        "WalletAuthResponse",
-        {
-            "access_token": token,
-            "refresh_token": token,
-            "user": {
-                "id": user_id,
-                "email": req.email,
-                "display_name": req.display_name,
-                "wallet_address": req.wallet_address,
-                "wallet_chain": req.wallet_chain,
-                "tier": "FREE",
-                "role": "USER",
-                "created_at": user["created_at"],
-            },
-        },
-    )
-
-
-@router.post("/login", response_model=WalletAuthResponse)
-def login_email(req: EmailLoginRequest) -> WalletAuthResponse:
-    """Login with email/password. If 2FA enabled, returns partial token requiring /2fa/login."""
-    user = _get_user_by_email(req.email)
-    if not user or not user.get("password_hash"):
-        raise HTTPException(status_code=401, detail="Invalid credentials")
-
-    if not verify_password(req.password, user["password_hash"]):
-        raise HTTPException(status_code=401, detail="Invalid credentials")
-
-    # If 2FA enabled, return a pre-auth token that requires TOTP verification
-    if user.get("totp_enabled") and user.get("totp_secret"):
-        # Generate a short-lived pre-auth token (5 min)
-        pre_token = _create_jwt(
-            user["id"],
-            user["email"],
-            user.get("tier", "FREE"),
-            user.get("role", "USER"),
-            user.get("wallet_address"),
-        )
-        # Override expiry to 5 minutes
-        import jose.jwt as jose_jwt
-
-        payload = jose_jwt.decode(pre_token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
-        payload["exp"] = datetime.utcnow() + timedelta(minutes=5)
-        payload["2fa_required"] = True
-        payload["pre_auth"] = True
-        pre_token = jose_jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
-
-        return cast(
-            "WalletAuthResponse",
-            {
-                "access_token": pre_token,
-                "refresh_token": pre_token,
-                "user": {
-                    "id": user["id"],
-                    "email": user["email"],
-                    "display_name": user.get("display_name", user["email"]),
-                    "wallet_address": user.get("wallet_address"),
-                    "wallet_chain": user.get("wallet_chain"),
-                    "tier": user.get("tier", "FREE"),
-                    "role": user.get("role", "USER"),
-                    "created_at": user.get("created_at"),
-                    "_2fa_required": True,
-                },
-            },
-        )
-
-    token = _create_jwt(
-        user["id"],
-        user["email"],
-        user.get("tier", "FREE"),
-        user.get("role", "USER"),
-        user.get("wallet_address"),
-    )
-
-    return cast(
-        "WalletAuthResponse",
-        {
-            "access_token": token,
-            "refresh_token": token,
-            "user": {
-                "id": user["id"],
-                "email": user["email"],
-                "display_name": user.get("display_name", user["email"]),
-                "wallet_address": user.get("wallet_address"),
-                "wallet_chain": user.get("wallet_chain"),
-                "tier": user.get("tier", "FREE"),
-                "role": user.get("role", "USER"),
-                "created_at": user.get("created_at"),
-            },
-        },
-    )
-
-
-# ── Wallet Auth ──
-@router.post("/wallet/nonce", response_model=NonceResponse)
-async def wallet_nonce(req: WalletNonceRequest) -> NonceResponse:
-    """Get a nonce for wallet signature."""
-    nonce = generate_nonce()
-    timestamp = datetime.utcnow().isoformat()
-    message = f"RugMunch Intelligence wants you to sign in with your {req.chain.title()} account.\n\nWallet: {req.address}\nNonce: {nonce}\nTimestamp: {timestamp}"
-    return cast("NonceResponse", {"nonce": nonce, "timestamp": timestamp, "message": message})
-
-
-@router.post("/wallet/verify", response_model=WalletAuthResponse)
-async def wallet_verify(req: WalletVerifyRequest) -> WalletAuthResponse:
-    """Verify wallet signature and create session."""
-    from app.domains.auth.wallet import get_or_create_wallet_user, verify_wallet_signature
-
-    if not verify_wallet_signature(req.message, req.signature, req.address):
-        raise HTTPException(status_code=401, detail="Invalid signature")
-
-    user_data = await get_or_create_wallet_user(req.address)
-
-    return cast(
-        "WalletAuthResponse",
-        {
-            "access_token": user_data["access_token"],
-            "refresh_token": user_data["refresh_token"],
-            "user": {
-                "id": user_data["id"],
-                "email": user_data["email"],
-                "display_name": user_data.get("display_name", f"Agent {req.address[2:8].upper()}"),
-                "wallet_address": req.address,
-                "wallet_chain": req.chain,
-                "tier": user_data["tier"],
-                "role": user_data["role"],
-                "created_at": user_data["created_at"],
-            },
-        },
-    )
-
-
-# ── User Profile ──
-@router.get("/user/me", response_model=UserResponse)
-async def user_me(request: Request) -> UserResponse:
-    """Get current authenticated user profile."""
-    auth_header = request.headers.get("Authorization", "")
-    if not auth_header.startswith("Bearer "):
-        raise HTTPException(status_code=401, detail="Missing auth token")
-
-    token = auth_header[7:]
-    payload = _verify_jwt(token)
-    if not payload:
-        raise HTTPException(status_code=401, detail="Invalid token")
-
-    user = _get_user(payload["id"])
-    if not user:
-        raise HTTPException(status_code=404, detail="User not found")
-
-    return cast(
-        "UserResponse",
-        {
-            "id": user["id"],
-            "email": user["email"],
-            "display_name": user.get("display_name", user["email"]),
-            "wallet_address": user.get("wallet_address"),
-            "wallet_chain": user.get("wallet_chain"),
-            "tier": user.get("tier", "FREE"),
-            "role": user.get("role", "USER"),
-            "created_at": user.get("created_at"),
-            "xp": user.get("xp", 0),
-            "level": user.get("level", 1),
-            "badges": user.get("badges", []),
-        },
-    )
-
-
-# ── 2FA / TOTP Endpoints ──
-
-
-@router.post("/2fa/setup")
-async def twofa_setup(request: Request) -> TwoFASetupResponse:
-    """Begin 2FA setup - generate secret + QR code. Returns plaintext secret + backup codes (shown once)."""
-    if not PYOTP_AVAILABLE:
-        raise HTTPException(status_code=503, detail="2FA service unavailable")
-
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-
-    # Check if already enabled
-    if user.get("totp_secret"):
-        raise HTTPException(
-            status_code=400, detail="2FA already enabled. Disable first to reconfigure."
-        )
-
-    secret = _generate_totp_secret()
-    uri = _get_totp_uri(secret, user["email"])
-    qr_b64 = _generate_qr_base64(uri)
-    backup_codes = _generate_backup_codes(8)
-
-    # Store pending secret (not enabled until verified)
-    r = get_redis()
-    assert r is not None
-    pending = {
-        "secret": secret,
-        "backup_codes": [_hash_backup_code(c) for c in backup_codes],
-        "created_at": datetime.utcnow().isoformat(),
-    }
-    r.setex(f"rmi:2fa_pending:{user['id']}", timedelta(minutes=10), json.dumps(pending))
-
-    return cast(
-        "TwoFASetupResponse",
-        {
-            "secret": secret,
-            "qr_code": qr_b64,
-            "uri": uri,
-            "backup_codes": backup_codes,  # plaintext - shown once
-        },
-    )
-
-
-@router.post("/2fa/enable")
-async def twofa_enable(req: TwoFAEnableRequest, request: Request) -> dict[str, Any]:
-    """Enable 2FA - verify the first TOTP code, then persist."""
-    if not PYOTP_AVAILABLE:
-        raise HTTPException(status_code=503, detail="2FA service unavailable")
-
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-
-    r = get_redis()
-    assert r is not None
-    pending_raw = r.get(f"rmi:2fa_pending:{user['id']}")
-    if not pending_raw:
-        raise HTTPException(status_code=400, detail="No pending 2FA setup. Call /2fa/setup first.")
-
-    pending = json.loads(pending_raw)
-    secret = pending["secret"]
-
-    # Verify the code
-    if not _verify_totp(secret, req.code):
-        raise HTTPException(
-            status_code=400, detail="Invalid TOTP code. Check your authenticator app time sync."
-        )
-
-    # Enable 2FA on user record
-    user["totp_secret"] = secret
-    user["totp_enabled"] = True
-    user["totp_created_at"] = pending["created_at"]
-    user["totp_backup_codes"] = pending["backup_codes"]  # already hashed
-    user["totp_last_used"] = None
-    _save_user(user)
-
-    # Clean up pending
-    r.delete(f"rmi:2fa_pending:{user['id']}")
-
-    # Audit log
-    logger.info(f"[2FA ENABLED] user={user['id']} email={user['email']}")
-
-    return {"status": "ok", "message": "2FA enabled successfully", "method": "totp"}
-
-
-@router.post("/2fa/disable")
-async def twofa_disable(request: Request) -> dict[str, Any]:
-    """Disable 2FA - requires current password for security."""
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-
-    if not user.get("totp_enabled"):
-        raise HTTPException(status_code=400, detail="2FA is not enabled")
-
-    # Remove 2FA fields
-    for key in [
-        "totp_secret",
-        "totp_enabled",
-        "totp_created_at",
-        "totp_backup_codes",
-        "totp_last_used",
-    ]:
-        user.pop(key, None)
-
-    _save_user(user)
-    logger.info(f"[2FA DISABLED] user={user['id']} email={user['email']}")
-
-    return cast("dict[str, Any]", {"status": "ok", "message": "2FA disabled successfully"})
-
-
-@router.get("/2fa/status", response_model=TwoFAStatusResponse)
-async def twofa_status(request: Request) -> TwoFAStatusResponse:
-    """Check 2FA status for current user."""
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-
-    enabled = user.get("totp_enabled", False)
-    return cast(
-        "TwoFAStatusResponse",
-        {
-            "enabled": enabled,
-            "method": "totp" if enabled else "none",
-            "created_at": user.get("totp_created_at"),
-            "last_used": user.get("totp_last_used"),
-        },
-    )
-
-
-@router.post("/2fa/verify")
-async def twofa_verify(req: TwoFAVerifyRequest, request: Request) -> dict[str, Any]:
-    """Verify a TOTP code (for re-auth during sensitive operations)."""
-    if not PYOTP_AVAILABLE:
-        raise HTTPException(status_code=503, detail="2FA service unavailable")
-
-    user = await get_current_user(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required")
-
-    if not user.get("totp_enabled") or not user.get("totp_secret"):
-        raise HTTPException(status_code=400, detail="2FA not enabled")
-
-    if not _verify_totp(user["totp_secret"], req.code):
-        raise HTTPException(status_code=400, detail="Invalid TOTP code")
-
-    # Update last used
-    user["totp_last_used"] = datetime.utcnow().isoformat()
-    _save_user(user)
-
-    return cast("dict[str, Any]", {"status": "ok", "verified": True})
-
-
-@router.post("/2fa/login")
-async def twofa_login(req: TwoFALoginRequest) -> WalletAuthResponse:
-    """Login with email + password + 2FA code. Returns full JWT on success."""
-    if not PYOTP_AVAILABLE:
-        raise HTTPException(status_code=503, detail="2FA service unavailable")
-
-    user = _get_user_by_email(req.email)
-    if not user or not user.get("password_hash"):
-        raise HTTPException(status_code=401, detail="Invalid credentials")
-
-    if not verify_password(req.password, user["password_hash"]):
-        raise HTTPException(status_code=401, detail="Invalid credentials")
-
-    # Check if 2FA is enabled
-    if not user.get("totp_enabled") or not user.get("totp_secret"):
-        raise HTTPException(
-            status_code=400, detail="2FA not enabled for this account. Use /login instead."
-        )
-
-    code = req.code.strip().replace("-", "")
-
-    # Try TOTP first
-    if _verify_totp(user["totp_secret"], code):
-        pass  # valid TOTP
-    else:
-        # Try backup codes
-        backup_codes = user.get("totp_backup_codes", [])
-        matched = False
-        for i, hashed in enumerate(backup_codes):
-            if _verify_backup_code(req.code, hashed):
-                matched = True
-                # Remove used backup code
-                backup_codes.pop(i)
-                user["totp_backup_codes"] = backup_codes
-                break
-        if not matched:
-            raise HTTPException(status_code=401, detail="Invalid 2FA code or backup code")
-
-    # Update last used
-    user["totp_last_used"] = datetime.utcnow().isoformat()
-    _save_user(user)
-
-    token = _create_jwt(
-        user["id"],
-        user["email"],
-        user.get("tier", "FREE"),
-        user.get("role", "USER"),
-        user.get("wallet_address"),
-    )
-
-    logger.info(f"[2FA LOGIN] user={user['id']} email={user['email']}")
-
-    return cast(
-        "WalletAuthResponse",
-        {
-            "access_token": token,
-            "refresh_token": token,
-            "user": {
-                "id": user["id"],
-                "email": user["email"],
-                "display_name": user.get("display_name", user["email"]),
-                "wallet_address": user.get("wallet_address"),
-                "wallet_chain": user.get("wallet_chain"),
-                "tier": user.get("tier", "FREE"),
-                "role": user.get("role", "USER"),
-                "created_at": user.get("created_at"),
-            },
-        },
-    )

app/domains/auth/schemas.py

@@ -1,106 +1,27 @@
-"""Pydantic schemas for auth domain.
+"""Pydantic schemas - re-exports from app.domains.auth.
 
 Phase 3B of AUDIT-2026-Q3.md.
+
+Public API:
+- EmailLoginRequest, EmailRegisterRequest
+- WalletNonceRequest, WalletVerifyRequest, WalletAuthResponse
+- NonceResponse, UserResponse, GoogleAuthResponse
+- TelegramAuthRequest, TwoFAEnableRequest, TwoFALoginRequest,
+  TwoFASetupResponse, TwoFAStatusResponse, TwoFAVerifyRequest
 """
-from __future__ import annotations
-
-from typing import Any
-
-from pydantic import BaseModel, Field
-
-
-class EmailLoginRequest(BaseModel):
-    email: str
-    password: str
-
-
-class EmailRegisterRequest(BaseModel):
-    email: str
-    password: str
-    display_name: str
-    wallet_address: str | None = None
-    wallet_chain: str | None = None
-
-
-class WalletNonceRequest(BaseModel):
-    address: str
-    chain: str
-
-
-class WalletVerifyRequest(BaseModel):
-    address: str
-    chain: str
-    nonce: str
-    signature: str
-    message: str
-
-
-class NonceResponse(BaseModel):
-    nonce: str
-    timestamp: str
-    message: str | None = None
-
-
-class WalletAuthResponse(BaseModel):
-    access_token: str
-    refresh_token: str
-    user: dict[str, Any]
-
-
-class UserResponse(BaseModel):
-    id: str
-    email: str
-    display_name: str
-    wallet_address: str | None
-    wallet_chain: str | None
-    tier: str
-    role: str
-    created_at: str
-    xp: int = 0
-    level: int = 1
-    badges: list[str] = []
-
-
-class GoogleAuthResponse(BaseModel):
-    url: str
-
-
-class TelegramAuthRequest(BaseModel):
-    id: int
-    first_name: str | None = None
-    last_name: str | None = None
-    username: str | None = None
-    photo_url: str | None = None
-    auth_date: int
-    hash: str
-
-
-# ── 2FA Models ──
-class TwoFASetupResponse(BaseModel):
-    secret: str
-    qr_code: str  # base64 data URI
-    uri: str  # otpauth:// URI
-    backup_codes: list[str]  # plaintext - shown once
-
-
-class TwoFAEnableRequest(BaseModel):
-    code: str = Field(..., min_length=6, max_length=6, pattern=r"^\d{6}$")
-
-
-class TwoFAVerifyRequest(BaseModel):
-    code: str = Field(..., min_length=6, max_length=6, pattern=r"^\d{6}$")
-
-
-class TwoFALoginRequest(BaseModel):
-    email: str
-    password: str
-    code: str = Field(..., min_length=6, max_length=10)  # 6 digits or backup code XXXX-XXXX
-
-
-class TwoFAStatusResponse(BaseModel):
-    enabled: bool
-    method: str = "totp"  # future: u2f, webauthn
-    created_at: str | None = None
-    last_used: str | None = None
-
-
+from app.auth import (  # noqa: F401
+    EmailLoginRequest,
+    EmailRegisterRequest,
+    GoogleAuthResponse,
+    NonceResponse,
+    TelegramAuthRequest,
+    TwoFAEnableRequest,
+    TwoFALoginRequest,
+    TwoFASetupResponse,
+    TwoFAStatusResponse,
+    TwoFAVerifyRequest,
+    UserResponse,
+    WalletAuthResponse,
+    WalletNonceRequest,
+    WalletVerifyRequest,
+)

app/domains/auth/store.py

@@ -1,4 +1,4 @@
-"""User storage helpers.
+"""User storage helpers - re-exports from app.domains.auth.
 
 Phase 3B of AUDIT-2026-Q3.md.
 
@@ -6,65 +6,10 @@ Public API:
 - _get_user, _get_user_by_email, _save_user, _delete_user
 - _is_valid_email
 """
-from __future__ import annotations
-
-import json
-import re
-from typing import Any, cast
-
-from app.core.redis import get_redis
-
-
-def _get_user(user_id: str) -> dict[str, Any] | None:
-    """Fetch user record by id from Redis hash rmi:users."""
-    r = get_redis()
-    if not r:
-        return None
-    raw = r.hget("rmi:users", user_id)
-    if not raw:
-        return None
-    try:
-        return cast("dict[str, Any]", json.loads(raw))
-    except (json.JSONDecodeError, TypeError):
-        return None
-
-
-def _get_user_by_email(email: str) -> dict[str, Any] | None:
-    """Fetch user record by email via rmi:users:email index -> rmi:users hash."""
-    r = get_redis()
-    if not r:
-        return None
-    user_id = r.hget("rmi:users:email", email.lower())
-    if not user_id:
-        return None
-    if isinstance(user_id, bytes):
-        user_id = user_id.decode("utf-8")
-    return _get_user(user_id)
-
-
-def _save_user(user: dict[str, Any]) -> None:
-    """Persist user record to Redis hash rmi:users, refresh email index if present."""
-    r = get_redis()
-    if not r:
-        return
-    user_id = user["id"]
-    r.hset("rmi:users", user_id, json.dumps(user))
-    if user.get("email"):
-        r.hset("rmi:users:email", user["email"].lower(), user_id)
-
-
-def _delete_user(user_id: str) -> None:
-    """Remove user record from Redis hash rmi:users and email index."""
-    r = get_redis()
-    if not r:
-        return
-    user = _get_user(user_id)
-    if user and user.get("email"):
-        r.hdel("rmi:users:email", user["email"].lower())
-    r.hdel("rmi:users", user_id)
-
-
-def _is_valid_email(email: str) -> bool:
-    """Basic email validation."""
-    pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
-    return re.match(pattern, email) is not None
+from app.auth import (  # noqa: F401
+    _delete_user,
+    _get_user,
+    _get_user_by_email,
+    _is_valid_email,
+    _save_user,
+)

app/domains/auth/totp.py

@@ -1,92 +1,24 @@
-"""TOTP / 2FA helpers.
+"""TOTP / 2FA helpers - re-exports from app.domains.auth.
 
 Phase 3B of AUDIT-2026-Q3.md.
 
 Public API:
-- PYOTP_AVAILABLE
-- TOTP_ISSUER, TOTP_DIGITS, TOTP_INTERVAL, TOTP_WINDOW
-- _generate_totp_secret, _build_totp, _verify_totp
-- _get_totp_uri, _generate_qr_base64
-- _generate_backup_codes, _hash_backup_code, _verify_backup_code
+- _generate_totp_secret, _build_totp, _verify_totp, _get_totp_uri
+- _generate_qr_base64, _generate_backup_codes, _hash_backup_code, _verify_backup_code
+- TOTP_ISSUER, PYOTP_AVAILABLE
 """
-from __future__ import annotations
-
-import base64
-import io
-import logging
-import os
-import secrets
-from typing import Any
-
-from app.domains.auth.passwords import hash_password, verify_password
-
-logger = logging.getLogger(__name__)
-
-try:
-    import pyotp
-    import qrcode
-
-    PYOTP_AVAILABLE = True
-except ImportError:
-    PYOTP_AVAILABLE = False
-    logger.warning("pyotp not installed - 2FA endpoints will return 503")
-
-TOTP_ISSUER = os.getenv("TOTP_ISSUER", "RugMunch Intelligence")
-TOTP_DIGITS = 6
-TOTP_INTERVAL = 30
-TOTP_WINDOW = 1  # Allow ±1 interval clock drift
-
-
-def _generate_totp_secret() -> str:
-    """Generate a new base32 TOTP secret."""
-    if not PYOTP_AVAILABLE:
-        raise RuntimeError("pyotp not installed")
-    return str(pyotp.random_base32())
-
-
-def _build_totp(secret: str) -> Any:
-    """Build a TOTP object from secret."""
-    return pyotp.TOTP(secret, digits=TOTP_DIGITS, interval=TOTP_INTERVAL)
-
-
-def _verify_totp(secret: str, code: str) -> bool:
-    """Verify a TOTP code with clock-drift tolerance."""
-    if not PYOTP_AVAILABLE or not secret or not code:
-        return False
-    totp = _build_totp(secret)
-    return bool(totp.verify(code, valid_window=TOTP_WINDOW))
-
-
-def _get_totp_uri(secret: str, email: str) -> str:
-    """Get otpauth:// URI for QR code generation."""
-    totp = _build_totp(secret)
-    return str(totp.provisioning_uri(name=email, issuer_name=TOTP_ISSUER))
-
-
-def _generate_qr_base64(uri: str) -> str:
-    """Generate QR code as base64 PNG data URI."""
-    img = qrcode.make(uri)
-    buf = io.BytesIO()
-    img.save(buf, format="PNG")
-    return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode()
-
-
-def _generate_backup_codes(count: int = 8) -> list[str]:
-    """Generate single-use backup codes."""
-    codes = []
-    for _ in range(count):
-        # 8-digit numeric codes, hyphenated for readability
-        raw = secrets.randbelow(10_000_000_000)
-        code = f"{raw:010d}"[:8]
-        codes.append(f"{code[:4]}-{code[4:]}")
-    return codes
-
-
-def _hash_backup_code(code: str) -> str:
-    """Hash a backup code for storage (same as password hashing)."""
-    return hash_password(code)
-
-
-def _verify_backup_code(code: str, hashed: str) -> bool:
-    """Verify a backup code against its hash."""
-    return verify_password(code, hashed)
+from app.auth import (  # noqa: F401
+    PYOTP_AVAILABLE,
+    TOTP_DIGITS,
+    TOTP_INTERVAL,
+    TOTP_ISSUER,
+    TOTP_WINDOW,
+    _build_totp,
+    _generate_backup_codes,
+    _generate_qr_base64,
+    _generate_totp_secret,
+    _get_totp_uri,
+    _hash_backup_code,
+    _verify_backup_code,
+    _verify_totp,
+)

app/domains/auth/wallet.py

@@ -33,7 +33,7 @@ def verify_wallet_signature(message: str, signature: str, address: str) -> bool:
     return len(signature) >= 60
 
 
-def decode_signature(signature: str) -> tuple[str, str, int]:
+def decode_signature(signature: str) -> tuple:
     """
     Decode a wallet signature into r, s, v components.
 
@@ -118,7 +118,7 @@ async def get_or_create_wallet_user(address: str, chain: str = "base") -> dict[s
             r.hset("rmi:users", user_id, json.dumps(user))
 
     # Generate JWT token
-    from app.domains.auth.jwt import _create_jwt
+    from app.auth import _create_jwt
 
     # Use email if available, otherwise derive from address
     email = user.get("email") or f"{address.lower()}@wallet.rmi"
@@ -145,7 +145,7 @@ async def verify_auth_token(token: str) -> dict[str, Any] | None:
     Returns:
         Dict with user info if valid, None if invalid/expired
     """
-    from app.domains.auth.jwt import _verify_jwt
+    from app.auth import _verify_jwt
 
     try:
         user = _verify_jwt(token)

app/domains/billing/x402/tools/integration.py

@@ -388,7 +388,7 @@ async def framework_discovery():
                 "free": True,
             },
             "mcp": {
-                "endpoint": "python -m app.domains.mcp.x402_mcp_server",
+                "endpoint": "python -m app.mcp.x402_mcp_server",
                 "format": "Model Context Protocol (stdio)",
                 "usage": "Claude Desktop, Claude Code, any MCP client",
                 "free": True,

app/domains/billing/x402/tools/token_tools.py

@@ -51,7 +51,7 @@ async def deep_contract_audit(req: TokenRequest):
 
         # If primary returned no data, use full fallback chain
         if not result.get("sources_used"):
-            from app.core.redis import get_redis as _get_redis
+            from app.auth import get_redis as _get_redis
             from app.fallback_engine import try_all_fallbacks
 
             r = await _get_redis()
@@ -72,7 +72,7 @@ async def deep_contract_audit(req: TokenRequest):
     except Exception as e:
         # Last resort: try fallback before raising error
         try:
-            from app.core.redis import get_redis as _get_redis
+            from app.auth import get_redis as _get_redis
             from app.fallback_engine import try_all_fallbacks
 
             r = await _get_redis()

app/domains/bulletin/__init__.py

@@ -1,35 +0,0 @@
-"""Bulletin domain - public API.
-
-Phase 4 domain consolidation: app.bulletin_board -> app.domains.bulletin.
-"""
-from __future__ import annotations
-
-from app.domains.bulletin.core import (
-    BulletinBoardManager,
-    Comment,
-    ContentSanitizer,
-    Post,
-    PostCategory,
-    PostStatus,
-    Priority,
-    TargetAudience,
-    award_badge,
-    get_user_badges,
-    get_user_reputation,
-    verify_x402_bot,
-)
-
-__all__ = [
-    "BulletinBoardManager",
-    "Comment",
-    "ContentSanitizer",
-    "Post",
-    "PostCategory",
-    "PostStatus",
-    "Priority",
-    "TargetAudience",
-    "award_badge",
-    "get_user_badges",
-    "get_user_reputation",
-    "verify_x402_bot",
-]

app/domains/bulletin/core.py

@@ -1,902 +0,0 @@
-"""
-RMI Bulletin Board Management System
-=====================================
-A full content management backend for announcements, news, alerts, platform
-communications, and community bulletin boards.
-
-Features:
-  • Posts - CRUD with rich text, attachments, scheduling, expiry
-  • Categories - organize by type (news, alert, update, promo, system)
-  • Targeting - audience segmentation (all, free, premium, admins, specific tiers)
-  • Moderation - draft/review/published/archived workflow, approval chains
-  • Pinning - sticky posts, priority ordering
-  • Analytics - views, clicks, engagement tracking per post
-  • Comments - threaded discussions on posts (optional)
-  • Notifications - push/email/Telegram alerts for critical posts
-  • Scheduling - publish at future date, auto-archive after expiry
-  • Versioning - track edit history, rollback capability
-  • Search - full-text search across all posts
-  • SEO - slug generation, meta tags, OpenGraph
-
-Security:
-  - All write operations require admin auth + content.write permission
-  - Audit log of every content change
-  - Rate limiting on publish operations
-  - Content sanitization (strip XSS, validate HTML)
-"""
-
-from __future__ import annotations
-
-import html
-import json
-import logging
-import os
-import re
-import time
-from dataclasses import asdict, dataclass, field
-from datetime import datetime
-from enum import StrEnum
-from typing import Any, ClassVar
-
-logger = logging.getLogger("rmi_bulletin_board")
-
-
-# ── Enums ─────────────────────────────────────────────────────
-
-
-class PostStatus(StrEnum):
-    DRAFT = "draft"
-    REVIEW = "review"
-    PUBLISHED = "published"
-    ARCHIVED = "archived"
-    SCHEDULED = "scheduled"
-
-
-class PostCategory(StrEnum):
-    NEWS = "news"  # Platform news
-    ALERT = "alert"  # Security/urgent alerts
-    UPDATE = "update"  # Feature updates
-    PROMO = "promo"  # Promotions/offers
-    SYSTEM = "system"  # System maintenance
-    COMMUNITY = "community"  # Community posts
-    ANNOUNCEMENT = "announcement"  # General announcements
-    TUTORIAL = "tutorial"  # Guides/how-tos
-
-
-class TargetAudience(StrEnum):
-    ALL = "all"
-    FREE = "free"
-    PREMIUM = "premium"
-    PRO = "pro"
-    ENTERPRISE = "enterprise"
-    ADMINS = "admins"
-    MODERATORS = "moderators"
-
-
-class Priority(StrEnum):
-    LOW = "low"
-    NORMAL = "normal"
-    HIGH = "high"
-    CRITICAL = "critical"
-
-
-# ── Data Models ─────────────────────────────────────────────
-
-
-@dataclass
-class Post:
-    """Bulletin board post."""
-
-    post_id: str
-    title: str
-    slug: str
-    content: str
-    summary: str
-    category: str
-    status: str
-    priority: str
-    target_audience: str
-    author_id: str
-    author_email: str
-    author_name: str
-    created_at: str
-    updated_at: str
-    published_at: str | None = None
-    scheduled_at: str | None = None
-    expires_at: str | None = None
-    archived_at: str | None = None
-    pinned: bool = False
-    pin_order: int = 0
-    featured_image: str = ""
-    attachments: list[dict] = field(default_factory=list)
-    tags: list[str] = field(default_factory=list)
-    meta_title: str = ""
-    meta_description: str = ""
-    og_image: str = ""
-    view_count: int = 0
-    click_count: int = 0
-    engagement_score: float = 0.0
-    version: int = 1
-    edit_history: list[dict] = field(default_factory=list)
-    approved_by: str = ""
-    approved_at: str | None = None
-    notification_sent: bool = False
-    allow_comments: bool = False
-    comments_count: int = 0
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-    def to_public_dict(self) -> dict:
-        """Return public-safe version (no internal fields)."""
-        return {
-            "post_id": self.post_id,
-            "title": self.title,
-            "slug": self.slug,
-            "content": self.content,
-            "summary": self.summary,
-            "category": self.category,
-            "priority": self.priority,
-            "author_name": self.author_name,
-            "created_at": self.created_at,
-            "published_at": self.published_at,
-            "expires_at": self.expires_at,
-            "pinned": self.pinned,
-            "pin_order": self.pin_order,
-            "featured_image": self.featured_image,
-            "attachments": self.attachments,
-            "tags": self.tags,
-            "meta_title": self.meta_title,
-            "meta_description": self.meta_description,
-            "og_image": self.og_image,
-            "view_count": self.view_count,
-            "allow_comments": self.allow_comments,
-            "comments_count": self.comments_count,
-        }
-
-
-@dataclass
-class Comment:
-    """Comment on a post."""
-
-    comment_id: str
-    post_id: str
-    author_id: str
-    author_name: str
-    author_email: str
-    content: str
-    created_at: str
-    updated_at: str
-    parent_id: str | None = None
-    status: str = "approved"  # approved, pending, rejected
-    likes: int = 0
-    replies: list[dict] = field(default_factory=list)
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-
-# ── Content Sanitizer ───────────────────────────────────────
-
-
-class ContentSanitizer:
-    """Sanitize user-generated content to prevent XSS."""
-
-    ALLOWED_TAGS: ClassVar[dict] ={
-        "p",
-        "br",
-        "strong",
-        "b",
-        "em",
-        "i",
-        "u",
-        "h1",
-        "h2",
-        "h3",
-        "h4",
-        "h5",
-        "h6",
-        "ul",
-        "ol",
-        "li",
-        "a",
-        "img",
-        "blockquote",
-        "code",
-        "pre",
-        "table",
-        "thead",
-        "tbody",
-        "tr",
-        "td",
-        "th",
-        "div",
-        "span",
-        "hr",
-        "sub",
-        "sup",
-        "del",
-        "ins",
-    }
-
-    ALLOWED_ATTRS: ClassVar[dict] ={
-        "a": ["href", "title", "target"],
-        "img": ["src", "alt", "title", "width", "height"],
-        "div": ["class"],
-        "span": ["class"],
-        "code": ["class"],
-        "pre": ["class"],
-    }
-
-    @staticmethod
-    def sanitize(text: str) -> str:
-        """Basic HTML sanitization."""
-        if not text:
-            return ""
-
-        # Escape HTML entities first
-        text = html.escape(text)
-
-        # Then selectively un-escape allowed tags
-        # This is a simplified approach - in production use bleach or similar
-        # For now, strip all HTML tags for safety
-        text = re.sub(r"<[^>]+>", "", text)
-
-        return text.strip()
-
-    @staticmethod
-    def generate_slug(title: str) -> str:
-        """Generate URL-friendly slug from title."""
-        slug = re.sub(r"[^\w\s-]", "", title.lower())
-        slug = re.sub(r"[-\s]+", "-", slug)
-        return slug[:80]
-
-    @staticmethod
-    def generate_summary(content: str, max_length: int = 200) -> str:
-        """Generate summary from content."""
-        # Strip HTML
-        text = re.sub(r"<[^>]+>", "", content)
-        text = text.replace("\n", " ").strip()
-        if len(text) > max_length:
-            text = text[:max_length].rsplit(" ", 1)[0] + "..."
-        return text
-
-
-# ── Bulletin Board Manager ────────────────────────────────────
-
-
-class BulletinBoardManager:
-    """
-    Core manager for bulletin board operations.
-    Uses Redis as primary store + Supabase backup.
-    """
-
-    @staticmethod
-    async def _get_redis():
-        import redis.asyncio as redis_lib
-
-        return redis_lib.Redis(
-            host=os.getenv("REDIS_HOST", "localhost"),
-            port=int(os.getenv("REDIS_PORT", "6379")),
-            password=os.getenv("REDIS_PASSWORD", ""),
-            decode_responses=True,
-        )
-
-    @staticmethod
-    async def create_post(
-        title: str,
-        content: str,
-        category: str,
-        author_id: str,
-        author_email: str,
-        author_name: str = "",
-        priority: str = "normal",
-        target_audience: str = "all",
-        status: str = "draft",
-        featured_image: str = "",
-        attachments: list[dict] | None = None,
-        tags: list[str] | None = None,
-        scheduled_at: str | None = None,
-        expires_at: str | None = None,
-        pinned: bool = False,
-        allow_comments: bool = False,
-        meta_title: str = "",
-        meta_description: str = "",
-    ) -> Post:
-        """Create a new post."""
-        post_id = f"post_{int(time.time())}_{os.urandom(4).hex()}"
-        slug = ContentSanitizer.generate_slug(title)
-        summary = ContentSanitizer.generate_summary(content)
-        now = datetime.utcnow().isoformat()
-
-        # Sanitize content
-        content = ContentSanitizer.sanitize(content)
-
-        post = Post(
-            post_id=post_id,
-            title=title[:200],
-            slug=slug,
-            content=content,
-            summary=summary,
-            category=category,
-            status=status,
-            priority=priority,
-            target_audience=target_audience,
-            author_id=author_id,
-            author_email=author_email,
-            author_name=author_name or author_email.split("@")[0],
-            created_at=now,
-            updated_at=now,
-            scheduled_at=scheduled_at,
-            expires_at=expires_at,
-            pinned=pinned,
-            featured_image=featured_image,
-            attachments=attachments or [],
-            tags=tags or [],
-            meta_title=meta_title or title[:70],
-            meta_description=meta_description or summary[:160],
-            allow_comments=allow_comments,
-        )
-
-        r = await BulletinBoardManager._get_redis()
-
-        # Save post
-        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post.to_dict()))
-
-        # Add to category index
-        await r.sadd(f"bulletin:category:{category}", post_id)
-
-        # Add to status index
-        await r.sadd(f"bulletin:status:{status}", post_id)
-
-        # Add to author index
-        await r.sadd(f"bulletin:author:{author_id}", post_id)
-
-        # Add to audience index
-        await r.sadd(f"bulletin:audience:{target_audience}", post_id)
-
-        # Add to pinned index if pinned
-        if pinned:
-            await r.sadd("bulletin:pinned", post_id)
-            await r.zadd("bulletin:pinned_order", {post_id: post.pin_order})
-
-        # Add to scheduled index if scheduled
-        if scheduled_at and status == "scheduled":
-            ts = int(datetime.fromisoformat(scheduled_at).timestamp())
-            await r.zadd("bulletin:scheduled", {post_id: ts})
-
-        # Add to search index (simple word index)
-        words = set(re.findall(r"\w+", title.lower() + " " + content.lower()))
-        for word in words:
-            if len(word) > 2:
-                await r.sadd(f"bulletin:search:{word}", post_id)
-
-        # Save to Supabase
-        try:
-            from supabase import create_client
-
-            supabase_url = os.getenv("SUPABASE_URL")
-            supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
-            if supabase_url and supabase_key:
-                client = create_client(supabase_url, supabase_key)
-                client.table("bulletin_posts").insert(post.to_dict()).execute()
-        except Exception as e:
-            logger.error(f"Supabase bulletin post save failed: {e}")
-
-        return post
-
-    @staticmethod
-    async def get_post(post_id: str, increment_views: bool = False) -> Post | None:
-        """Get a post by ID."""
-        r = await BulletinBoardManager._get_redis()
-        data = await r.hget("rmi:bulletin_posts", post_id)
-        if not data:
-            return None
-
-        post_dict = json.loads(data)
-
-        if increment_views and post_dict.get("status") == "published":
-            post_dict["view_count"] = post_dict.get("view_count", 0) + 1
-            await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-
-        return Post(**post_dict)
-
-    @staticmethod
-    async def get_post_by_slug(slug: str) -> Post | None:
-        """Get a post by slug."""
-        r = await BulletinBoardManager._get_redis()
-        # Search all posts for matching slug
-        all_posts = await r.hgetall("rmi:bulletin_posts")
-        for _post_id, data in all_posts.items():
-            post_dict = json.loads(data)
-            if post_dict.get("slug") == slug:
-                return Post(**post_dict)
-        return None
-
-    @staticmethod
-    async def update_post(
-        post_id: str,
-        updates: dict[str, Any],
-        editor_id: str = "",
-        editor_email: str = "",
-    ) -> Post | None:
-        """Update a post with versioning."""
-        r = await BulletinBoardManager._get_redis()
-        data = await r.hget("rmi:bulletin_posts", post_id)
-        if not data:
-            return None
-
-        post_dict = json.loads(data)
-        before_state = {k: post_dict.get(k) for k in updates if k in post_dict}
-
-        # Record edit history
-        edit_entry = {
-            "edited_at": datetime.utcnow().isoformat(),
-            "edited_by": editor_id,
-            "editor_email": editor_email,
-            "changes": before_state,
-            "version": post_dict.get("version", 1),
-        }
-
-        history = post_dict.get("edit_history", [])
-        history.append(edit_entry)
-        post_dict["edit_history"] = history
-        post_dict["version"] = post_dict.get("version", 1) + 1
-        post_dict["updated_at"] = datetime.utcnow().isoformat()
-
-        # Apply updates
-        for key, value in updates.items():
-            if key == "content":
-                value = ContentSanitizer.sanitize(value)
-                post_dict["summary"] = ContentSanitizer.generate_summary(value)
-            if key == "title":
-                post_dict["slug"] = ContentSanitizer.generate_slug(value)
-            post_dict[key] = value
-
-        # Handle status transitions
-        old_status = before_state.get("status")
-        new_status = post_dict.get("status")
-        if old_status != new_status:
-            # Update status indexes
-            if old_status:
-                await r.srem(f"bulletin:status:{old_status}", post_id)
-            await r.sadd(f"bulletin:status:{new_status}", post_id)
-
-            if new_status == "published":
-                post_dict["published_at"] = datetime.utcnow().isoformat()
-            elif new_status == "archived":
-                post_dict["archived_at"] = datetime.utcnow().isoformat()
-
-        # Handle pinning changes
-        if "pinned" in updates:
-            if updates["pinned"]:
-                await r.sadd("bulletin:pinned", post_id)
-                await r.zadd("bulletin:pinned_order", {post_id: post_dict.get("pin_order", 0)})
-            else:
-                await r.srem("bulletin:pinned", post_id)
-                await r.zrem("bulletin:pinned_order", post_id)
-
-        # Save updated post
-        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-
-        # Update Supabase
-        try:
-            from supabase import create_client
-
-            supabase_url = os.getenv("SUPABASE_URL")
-            supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
-            if supabase_url and supabase_key:
-                client = create_client(supabase_url, supabase_key)
-                client.table("bulletin_posts").upsert(post_dict).execute()
-        except Exception as e:
-            logger.error(f"Supabase bulletin post update failed: {e}")
-
-        return Post(**post_dict)
-
-    @staticmethod
-    async def delete_post(post_id: str) -> bool:
-        """Delete a post (soft delete - move to archive)."""
-        r = await BulletinBoardManager._get_redis()
-        data = await r.hget("rmi:bulletin_posts", post_id)
-        if not data:
-            return False
-
-        post_dict = json.loads(data)
-        post_dict["status"] = "archived"
-        post_dict["archived_at"] = datetime.utcnow().isoformat()
-
-        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-        await r.srem("bulletin:status:published", post_id)
-        await r.srem("bulletin:status:draft", post_id)
-        await r.srem("bulletin:status:review", post_id)
-        await r.sadd("bulletin:status:archived", post_id)
-        await r.srem("bulletin:pinned", post_id)
-
-        return True
-
-    @staticmethod
-    async def list_posts(
-        category: str | None = None,
-        status: str | None = None,
-        target_audience: str | None = None,
-        author_id: str | None = None,
-        pinned_only: bool = False,
-        search_query: str | None = None,
-        tags: list[str] | None = None,
-        limit: int = 50,
-        offset: int = 0,
-        sort_by: str = "created_at",
-        sort_order: str = "desc",
-    ) -> dict[str, Any]:
-        """List posts with filtering."""
-        r = await BulletinBoardManager._get_redis()
-
-        # Start with all posts or filtered set
-        if pinned_only:
-            post_ids = await r.smembers("bulletin:pinned")
-        elif category:
-            post_ids = await r.smembers(f"bulletin:category:{category}")
-        elif status:
-            post_ids = await r.smembers(f"bulletin:status:{status}")
-        elif author_id:
-            post_ids = await r.smembers(f"bulletin:author:{author_id}")
-        elif target_audience:
-            post_ids = await r.smembers(f"bulletin:audience:{target_audience}")
-        elif search_query:
-            # Search by words
-            words = re.findall(r"\w+", search_query.lower())
-            if words:
-                sets = [f"bulletin:search:{w}" for w in words if len(w) > 2]
-                if sets:
-                    post_ids = await r.sinter(sets)
-                else:
-                    post_ids = set()
-            else:
-                post_ids = set()
-        else:
-            all_posts = await r.hgetall("rmi:bulletin_posts")
-            post_ids = set(all_posts.keys())
-
-        # Apply additional filters
-        if tags:
-            tagged_posts = set()
-            all_posts = await r.hgetall("rmi:bulletin_posts")
-            for pid, data in all_posts.items():
-                post_dict = json.loads(data)
-                if any(tag in post_dict.get("tags", []) for tag in tags):
-                    tagged_posts.add(pid)
-            post_ids = post_ids.intersection(tagged_posts)
-
-        # Fetch and sort posts
-        posts = []
-        all_posts = await r.hgetall("rmi:bulletin_posts")
-        for pid in post_ids:
-            data = all_posts.get(pid)
-            if data:
-                post_dict = json.loads(data)
-                posts.append(post_dict)
-
-        # Sort
-        reverse = sort_order == "desc"
-        posts.sort(key=lambda x: x.get(sort_by, ""), reverse=reverse)
-
-        # Pinned posts first if not pinned_only
-        if not pinned_only:
-            pinned_ids = await r.smembers("bulletin:pinned")
-            posts.sort(
-                key=lambda x: (x["post_id"] not in pinned_ids, x.get(sort_by, "")),
-                reverse=not reverse,
-            )
-
-        total = len(posts)
-        posts = posts[offset : offset + limit]
-
-        return {
-            "posts": posts,
-            "total": total,
-            "limit": limit,
-            "offset": offset,
-        }
-
-    @staticmethod
-    async def publish_scheduled() -> list[str]:
-        """Publish posts that are scheduled for now."""
-        r = await BulletinBoardManager._get_redis()
-        now = int(time.time())
-
-        # Get scheduled posts that are due
-        due = await r.zrangebyscore("bulletin:scheduled", 0, now)
-        published = []
-
-        for post_id in due:
-            data = await r.hget("rmi:bulletin_posts", post_id)
-            if data:
-                post_dict = json.loads(data)
-                post_dict["status"] = "published"
-                post_dict["published_at"] = datetime.utcnow().isoformat()
-                post_dict["scheduled_at"] = None
-
-                await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-                await r.srem("bulletin:status:scheduled", post_id)
-                await r.sadd("bulletin:status:published", post_id)
-                await r.zrem("bulletin:scheduled", post_id)
-
-                published.append(post_id)
-
-        return published
-
-    @staticmethod
-    async def archive_expired() -> list[str]:
-        """Archive posts that have expired."""
-        r = await BulletinBoardManager._get_redis()
-        now = datetime.utcnow().isoformat()
-
-        all_posts = await r.hgetall("rmi:bulletin_posts")
-        archived = []
-
-        for post_id, data in all_posts.items():
-            post_dict = json.loads(data)
-            if post_dict.get("status") == "published" and post_dict.get("expires_at") and post_dict["expires_at"] < now:
-                post_dict["status"] = "archived"
-                post_dict["archived_at"] = now
-
-                await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-                await r.srem("bulletin:status:published", post_id)
-                await r.sadd("bulletin:status:archived", post_id)
-                await r.srem("bulletin:pinned", post_id)
-
-                archived.append(post_id)
-
-        return archived
-
-    @staticmethod
-    async def add_comment(
-        post_id: str,
-        author_id: str,
-        author_name: str,
-        author_email: str,
-        content: str,
-        parent_id: str | None = None,
-    ) -> Comment | None:
-        """Add a comment to a post."""
-        r = await BulletinBoardManager._get_redis()
-
-        # Check if post exists and allows comments
-        post_data = await r.hget("rmi:bulletin_posts", post_id)
-        if not post_data:
-            return None
-
-        post_dict = json.loads(post_data)
-        if not post_dict.get("allow_comments", False):
-            return None
-
-        comment_id = f"comment_{int(time.time())}_{os.urandom(4).hex()}"
-        now = datetime.utcnow().isoformat()
-
-        comment = Comment(
-            comment_id=comment_id,
-            post_id=post_id,
-            author_id=author_id,
-            author_name=author_name,
-            author_email=author_email,
-            content=ContentSanitizer.sanitize(content)[:2000],
-            created_at=now,
-            updated_at=now,
-            parent_id=parent_id,
-        )
-
-        await r.hset("rmi:bulletin_comments", comment_id, json.dumps(comment.to_dict()))
-        await r.sadd(f"bulletin:post_comments:{post_id}", comment_id)
-
-        # Update post comment count
-        post_dict["comments_count"] = post_dict.get("comments_count", 0) + 1
-        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-
-        return comment
-
-    @staticmethod
-    async def get_comments(post_id: str, limit: int = 100) -> list[dict]:
-        """Get comments for a post."""
-        r = await BulletinBoardManager._get_redis()
-        comment_ids = await r.smembers(f"bulletin:post_comments:{post_id}")
-
-        comments = []
-        for cid in comment_ids:
-            data = await r.hget("rmi:bulletin_comments", cid)
-            if data:
-                comments.append(json.loads(data))
-
-        comments.sort(key=lambda x: x.get("created_at", ""), reverse=True)
-        return comments[:limit]
-
-    @staticmethod
-    async def get_stats() -> dict[str, Any]:
-        """Get bulletin board statistics."""
-        r = await BulletinBoardManager._get_redis()
-
-        stats = {
-            "total_posts": await r.hlen("rmi:bulletin_posts") or 0,
-            "published": await r.scard("bulletin:status:published") or 0,
-            "drafts": await r.scard("bulletin:status:draft") or 0,
-            "review": await r.scard("bulletin:status:review") or 0,
-            "archived": await r.scard("bulletin:status:archived") or 0,
-            "scheduled": await r.scard("bulletin:status:scheduled") or 0,
-            "pinned": await r.scard("bulletin:pinned") or 0,
-            "total_comments": await r.hlen("rmi:bulletin_comments") or 0,
-            "categories": {},
-        }
-
-        # Count by category
-        for cat in PostCategory:
-            count = await r.scard(f"bulletin:category:{cat.value}") or 0
-            stats["categories"][cat.value] = count
-
-        return stats
-
-    @staticmethod
-    async def track_engagement(post_id: str, action: str) -> bool:
-        """Track engagement (view, click, etc.)."""
-        r = await BulletinBoardManager._get_redis()
-        data = await r.hget("rmi:bulletin_posts", post_id)
-        if not data:
-            return False
-
-        post_dict = json.loads(data)
-
-        if action == "view":
-            post_dict["view_count"] = post_dict.get("view_count", 0) + 1
-        elif action == "click":
-            post_dict["click_count"] = post_dict.get("click_count", 0) + 1
-
-        # Calculate engagement score
-        views = post_dict.get("view_count", 0)
-        clicks = post_dict.get("click_count", 0)
-        post_dict["engagement_score"] = round((clicks / max(views, 1)) * 100, 2)
-
-        await r.hset("rmi:bulletin_posts", post_id, json.dumps(post_dict))
-        return True
-
-
-# ═══════════════════════════════════════════════════════════
-# BADGES & X402 BOT PAYMENTS
-# ═══════════════════════════════════════════════════════════
-
-BADGES = {
-    "first_post": {
-        "name": "First Words",
-        "icon": "💬",
-        "desc": "Made your first post",
-        "tier": "bronze",
-    },
-    "10_posts": {"name": "Chatterbox", "icon": "📢", "desc": "10 posts", "tier": "bronze"},
-    "50_posts": {"name": "Board Regular", "icon": "🎙️", "desc": "50 posts", "tier": "silver"},
-    "100_posts": {
-        "name": "Terminally Online",
-        "icon": "🖥️",
-        "desc": "100 posts - touch grass",
-        "tier": "gold",
-    },
-    "10_upvotes": {
-        "name": "Approved",
-        "icon": "👍",
-        "desc": "10 upvotes on a post",
-        "tier": "bronze",
-    },
-    "50_upvotes": {"name": "Crowd Favorite", "icon": "⭐", "desc": "50 upvotes", "tier": "silver"},
-    "100_upvotes": {"name": "Legendary", "icon": "👑", "desc": "100 upvotes", "tier": "gold"},
-    "rug_reporter": {
-        "name": "Rug Detective",
-        "icon": "🔍",
-        "desc": "Reported 3 verified rugs",
-        "tier": "silver",
-    },
-    "scam_buster": {
-        "name": "Scam Buster",
-        "icon": "🛡️",
-        "desc": "10 scam alerts verified",
-        "tier": "gold",
-    },
-    "honeypot_hunter": {
-        "name": "Honeypot Hunter",
-        "icon": "🍯",
-        "desc": "Found 5 honeypots",
-        "tier": "silver",
-    },
-    "whale_watcher": {
-        "name": "Whale Watcher",
-        "icon": "🐋",
-        "desc": "Tracked 10 whale moves",
-        "tier": "silver",
-    },
-    "alpha_caller": {
-        "name": "Alpha Caller",
-        "icon": "📈",
-        "desc": "Called 5 pumps",
-        "tier": "gold",
-    },
-    "degen": {
-        "name": "Certified Degen",
-        "icon": "🎰",
-        "desc": "Posted in every category",
-        "tier": "gold",
-    },
-    "rug_survivor": {
-        "name": "Rug Survivor",
-        "icon": "💀",
-        "desc": "Posted about getting rugged",
-        "tier": "bronze",
-    },
-    "diamond_hands": {
-        "name": "Diamond Hands",
-        "icon": "💎",
-        "desc": "Held through -90%",
-        "tier": "diamond",
-    },
-    "based": {
-        "name": "Based",
-        "icon": "🧠",
-        "desc": "Called a 10x before it happened",
-        "tier": "diamond",
-    },
-    "bot_verified": {
-        "name": "Verified Bot",
-        "icon": "🤖",
-        "desc": "Registered x402 bot",
-        "tier": "silver",
-    },
-    "bot_pro": {"name": "Bot Pro", "icon": "⚡", "desc": "100+ API calls", "tier": "gold"},
-}
-BADGE_TIERS = {"bronze": "#CD7F32", "silver": "#C0C0C0", "gold": "#FFD700", "diamond": "#B9F2FF"}
-
-
-async def get_user_badges(user_id: str) -> list:
-    try:
-        r = await BulletinBoardManager._get_redis()
-        ids = await r.smembers(f"bb:badges:{user_id}")
-        await r.close()
-        return [{"id": bid, **BADGES[bid]} for bid in ids if bid in BADGES]
-    except Exception:
-        return []
-
-
-async def award_badge(user_id: str, badge_id: str) -> bool:
-    if badge_id not in BADGES:
-        return False
-    try:
-        r = await BulletinBoardManager._get_redis()
-        ok = await r.sadd(f"bb:badges:{user_id}", badge_id)
-        await r.close()
-        return ok > 0
-    except Exception:
-        return False
-
-
-async def get_user_reputation(user_id: str) -> dict:
-    try:
-        r = await BulletinBoardManager._get_redis()
-        p = r.pipeline()
-        p.get(f"bb:rep:{user_id}")
-        p.scard(f"bb:badges:{user_id}")
-        p.get(f"bb:posts:{user_id}")
-        rep, bc, pc = await p.execute()
-        await r.close()
-        return {"reputation": int(rep or 0), "badge_count": bc or 0, "post_count": int(pc or 0)}
-    except Exception:
-        return {"reputation": 0, "badge_count": 0, "post_count": 0}
-
-
-X402_BB_POST_PRICE = "$1.00"
-
-
-async def verify_x402_bot(tx_hash: str, bot_addr: str) -> bool:
-    try:
-        r = await BulletinBoardManager._get_redis()
-        if await r.get(f"bb:x402:tx:{tx_hash}"):
-            await r.close()
-            return False
-        await r.setex(f"bb:x402:tx:{tx_hash}", 86400, bot_addr)
-        await r.setex(f"bb:x402:bot:{bot_addr}", 2592000, "1")  # 30 day auth
-        await r.close()
-        return True
-    except Exception:
-        return False

app/domains/databus/__init__.py

@@ -1,69 +0,0 @@
-"""
-RMI DataBus - The Single Source of Truth for ALL Data
-=====================================================
-
-Every API call, MCP tool, x402 tool, scanner, and frontend hook routes through here.
-No raw HTTP calls to external APIs anywhere else. Period.
-
-Architecture (request flow):
-  Request → SecurityGate → CreditGate → CacheLayer → ProviderChain → Result
-        ↓                      ↓             ↓              ↓            ↓
-   admin check         free-first logic   L1→L2→L3      local-first!   RAG index
-   vault keys          auto-rotate      Redis+R2      OUR data 1st    WS broadcast
-
-What it REPLACES (do NOT use these anymore):
-  - app/cache_manager.py (RMICache)          → databus.cache
-  - app/caching_shield/unified_layer.py       → databus.fetch()
-  - app/caching_shield/data_fallback.py       → databus.fetch()
-  - app/caching_shield/api_registry.py        → databus.key_pool
-  - app/caching_shield/rate_limiter.py         → databus.rate_limiter
-  - app/arkham_connector.py                    → databus.providers.arkham
-  - app/coingecko_connector.py                → databus.providers.coingecko
-  - Direct .env reads for API keys            → databus.vault (encrypted in-memory)
-
-OWN DATA FIRST:
-  Our crown jewels - Wallet Memory Bank, RAG (17K docs), SENTINEL scanner,
-  Consensus RPC, Funding Tracer, ClickHouse, Price Consensus, News Network,
-  Bundle Detection, Label Import (169K+) - these are FAST, FREE, and OURS.
-  They go FIRST in every fallback chain. External APIs only augment or fill gaps.
-
-Usage:
-    from app.databus import databus
-
-    # Simple fetch - auto-selects best provider chain
-    result = await databus.fetch("token_price", mint="So11111111111111111111111111111111111111111")
-
-    # Explicit chain override
-    result = await databus.fetch("wallet_labels", address="7EcD...",
-                                  chain="local_first")
-
-    # Admin-only data (requires admin key in request)
-    result = await databus.fetch("arkham_entity", address="0x...",
-                                  admin_key="...")
-
-    # Check system health
-    health = await databus.health()
-
-    # Get capacity report (credits, rate limits, recommendations)
-    report = databus.capacity_report()
-"""
-
-from app.domains.databus.access_control import AccessController, ConsumerType, access_controller
-from app.domains.databus.core import DataBus, databus
-from app.domains.databus.key_affinity import KeyAffinitySelector, key_affinity
-from app.domains.databus.response_schema import SchemaValidator, schema_validator
-from app.domains.databus.social import SocialDataAggregator, XTwitterProvider
-
-__all__ = [
-    "AccessController",
-    "ConsumerType",
-    "DataBus",
-    "KeyAffinitySelector",
-    "SchemaValidator",
-    "SocialDataAggregator",
-    "XTwitterProvider",
-    "access_controller",
-    "databus",
-    "key_affinity",
-    "schema_validator",
-]

app/domains/databus/_generated/provider_chains.py

@@ -8,7 +8,7 @@ Source: app/databus/provider_chains.py (legacy shim).
 
 import logging
 
-from app.domains.databus.provider_core import (
+from app.databus.provider_core import (
     Provider,
     ProviderChain,
     ProviderTier,
@@ -17,7 +17,7 @@ from app.domains.databus.provider_core import (
     _rate_limiters,
     _RateLimiter,
 )
-from app.domains.databus.provider_implementations import (
+from app.databus.provider_implementations import (
     _alchemy_token_balances,
     _alchemy_token_metadata,
     _arkham_counterparties,
@@ -63,7 +63,7 @@ from app.domains.databus.provider_implementations import (
     _solana_tracker_trending,
     _virustotal_url_scan,
 )
-from app.domains.databus.spl_metadata_decoder import _spl_metadata_decoder_provider
+from app.databus.spl_metadata_decoder import _spl_metadata_decoder_provider
 
 logger = logging.getLogger("databus.providers.chains")
 def build_provider_chains() -> dict[str, ProviderChain]:
@@ -271,7 +271,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── Bitquery chains (activate free plan at bitquery.io/pricing) ──
     try:
-        from app.domains.databus.bitquery_provider import BitqueryProvider
+        from app.databus.bitquery_provider import BitqueryProvider
 
         _bq = BitqueryProvider()
 
@@ -758,7 +758,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── PREMIUM SCANNER - 10 high-value detection chains ──
     try:
-        from app.domains.databus.premium_scanner import (
+        from app.databus.premium_scanner import (
             detect_bot_farms,
             detect_bundles,
             detect_copy_trading,
@@ -842,11 +842,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── WEBHOOK SYSTEM ──
     try:
-        from app.domains.databus.webhooks import (  # noqa: F401
-            handle_webhook,
-            list_webhooks,
-            setup_webhook,
-        )
+        from app.databus.webhooks import handle_webhook, list_webhooks, setup_webhook  # noqa: F401
 
         chains["webhook_handler"] = ProviderChain(
             "webhook_handler",
@@ -870,7 +866,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── ARKHAM WEBSOCKET - real-time intelligence streaming ──
     try:
-        from app.domains.databus.arkham_ws import arkham_ws_subscribe
+        from app.databus.arkham_ws import arkham_ws_subscribe
 
         chains["arkham_ws"] = ProviderChain(
             "arkham_ws",
@@ -894,7 +890,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Volume Authenticity (Fake Volume %) ──
     try:
-        from app.domains.databus.volume_authenticity import analyze_volume_authenticity
+        from app.databus.volume_authenticity import analyze_volume_authenticity
 
         chains["volume_authenticity"] = ProviderChain(
             "volume_authenticity",
@@ -915,7 +911,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: OHLCV Engine ──
     try:
-        from app.domains.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
+        from app.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
 
         chains["ohlcv"] = ProviderChain(
             "ohlcv",
@@ -943,7 +939,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Token Security Matrix (37+ checks) ──
     try:
-        from app.domains.databus.token_security import get_check_matrix_endpoint, run_full_scan
+        from app.databus.token_security import get_check_matrix_endpoint, run_full_scan
 
         chains["token_security"] = ProviderChain(
             "token_security",
@@ -971,8 +967,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS INTELLIGENCE - 10 Premium Features ──
     try:
-        from app.domains.databus.data_quality import enhanced_token_report, get_tier_comparison
-        from app.domains.databus.rugcharts_intel import (
+        from app.databus.data_quality import enhanced_token_report, get_tier_comparison
+        from app.databus.rugcharts_intel import (
             cross_chain_entity,
             developer_reputation,
             holder_health_score,
@@ -1144,7 +1140,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS & MARKET DATA - Free APIs ──
     try:
-        from app.domains.databus.news_provider import (
+        from app.databus.news_provider import (
             get_fear_greed,
             get_full_news_feed,
             get_market_brief,
@@ -1245,7 +1241,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS INTELLIGENCE ENGINE - multi-source, quality-scored, sentiment-tagged ──
     try:
-        from app.domains.databus.news_intel import (
+        from app.databus.news_intel import (
             add_comment,
             add_reaction,
             aggregate_all_news,
@@ -1369,7 +1365,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── X/CT INTELLIGENCE - Crypto Twitter Rundown ──
     try:
-        from app.domains.databus.x_intel import fetch_ct_rundown, track_ct_accounts
+        from app.databus.x_intel import fetch_ct_rundown, track_ct_accounts
 
         chains["ct_rundown"] = ProviderChain(
             "ct_rundown",
@@ -1405,8 +1401,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── SOCIAL INTELLIGENCE - KOL tracking, shill detection, Daily Intel ──
     try:
-        from app.domains.databus.daily_intel import generate_daily_intel
-        from app.domains.databus.social_intel import (
+        from app.databus.daily_intel import generate_daily_intel
+        from app.databus.social_intel import (
             detect_shill_campaigns,
             get_kol_leaderboard,
             get_kol_profile,
@@ -1529,7 +1525,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── MODEL REGISTRY - Smart free model routing, quality review ──
     try:
-        from app.domains.databus.model_registry import ai_call, get_usage_stats, review_content
+        from app.databus.model_registry import ai_call, get_usage_stats, review_content
 
         chains["ai_task"] = ProviderChain(
             "ai_task",
@@ -1585,7 +1581,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RAG INGESTION - nightly indexing, health checks ──
     try:
-        from app.domains.databus.rag_ingestion import nightly_rag_index, rag_health_check
+        from app.databus.rag_ingestion import nightly_rag_index, rag_health_check
 
         chains["rag_nightly"] = ProviderChain(
             "rag_nightly",

app/domains/databus/providers/__init__.py

@@ -38,7 +38,7 @@ import httpx
 logger = logging.getLogger("databus.providers")
 
 # Import SPL metadata decoder
-from app.domains.databus.spl_metadata_decoder import _spl_metadata_decoder_provider  # noqa: E402
+from app.databus.spl_metadata_decoder import _spl_metadata_decoder_provider  # noqa: E402
 
 
 class ProviderTier(Enum):
@@ -362,10 +362,7 @@ async def _passthrough_news(**kwargs) -> dict | None:
 async def _passthrough_alerts(**kwargs) -> dict | None:
     """Alerts from our real alert pipeline."""
     try:
-        from app.domains.intelligence.alert_pipeline import (
-            get_active_alert_count,
-            get_recent_alerts,
-        )
+        from app.alert_pipeline import get_active_alert_count, get_recent_alerts
 
         count = await get_active_alert_count()
         recent = await get_recent_alerts(limit=kwargs.get("limit", 20))
@@ -1542,7 +1539,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── Bitquery chains (activate free plan at bitquery.io/pricing) ──
     try:
-        from app.domains.databus.bitquery_provider import BitqueryProvider
+        from app.databus.bitquery_provider import BitqueryProvider
 
         _bq = BitqueryProvider()
 
@@ -2029,7 +2026,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── PREMIUM SCANNER - 10 high-value detection chains ──
     try:
-        from app.domains.databus.premium_scanner import (
+        from app.databus.premium_scanner import (
             detect_bot_farms,
             detect_bundles,
             detect_copy_trading,
@@ -2113,11 +2110,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── WEBHOOK SYSTEM ──
     try:
-        from app.domains.databus.webhooks import (  # noqa: F401
-            handle_webhook,
-            list_webhooks,
-            setup_webhook,
-        )
+        from app.databus.webhooks import handle_webhook, list_webhooks, setup_webhook  # noqa: F401
 
         chains["webhook_handler"] = ProviderChain(
             "webhook_handler",
@@ -2141,7 +2134,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── ARKHAM WEBSOCKET - real-time intelligence streaming ──
     try:
-        from app.domains.databus.arkham_ws import arkham_ws_subscribe
+        from app.databus.arkham_ws import arkham_ws_subscribe
 
         chains["arkham_ws"] = ProviderChain(
             "arkham_ws",
@@ -2165,7 +2158,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Volume Authenticity (Fake Volume %) ──
     try:
-        from app.domains.databus.volume_authenticity import analyze_volume_authenticity
+        from app.databus.volume_authenticity import analyze_volume_authenticity
 
         chains["volume_authenticity"] = ProviderChain(
             "volume_authenticity",
@@ -2186,7 +2179,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: OHLCV Engine ──
     try:
-        from app.domains.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
+        from app.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
 
         chains["ohlcv"] = ProviderChain(
             "ohlcv",
@@ -2214,7 +2207,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Token Security Matrix (37+ checks) ──
     try:
-        from app.domains.databus.token_security import get_check_matrix_endpoint, run_full_scan
+        from app.databus.token_security import get_check_matrix_endpoint, run_full_scan
 
         chains["token_security"] = ProviderChain(
             "token_security",
@@ -2242,8 +2235,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS INTELLIGENCE - 10 Premium Features ──
     try:
-        from app.domains.databus.data_quality import enhanced_token_report, get_tier_comparison
-        from app.domains.databus.rugcharts_intel import (
+        from app.databus.data_quality import enhanced_token_report, get_tier_comparison
+        from app.databus.rugcharts_intel import (
             cross_chain_entity,
             developer_reputation,
             holder_health_score,
@@ -2415,7 +2408,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS & MARKET DATA - Free APIs ──
     try:
-        from app.domains.databus.news_provider import (
+        from app.databus.news_provider import (
             get_fear_greed,
             get_full_news_feed,
             get_market_brief,
@@ -2516,7 +2509,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS INTELLIGENCE ENGINE - multi-source, quality-scored, sentiment-tagged ──
     try:
-        from app.domains.databus.news_intel import (
+        from app.databus.news_intel import (
             add_comment,
             add_reaction,
             aggregate_all_news,
@@ -2640,7 +2633,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── X/CT INTELLIGENCE - Crypto Twitter Rundown ──
     try:
-        from app.domains.databus.x_intel import fetch_ct_rundown, track_ct_accounts
+        from app.databus.x_intel import fetch_ct_rundown, track_ct_accounts
 
         chains["ct_rundown"] = ProviderChain(
             "ct_rundown",
@@ -2676,8 +2669,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── SOCIAL INTELLIGENCE - KOL tracking, shill detection, Daily Intel ──
     try:
-        from app.domains.databus.daily_intel import generate_daily_intel
-        from app.domains.databus.social_intel import (
+        from app.databus.daily_intel import generate_daily_intel
+        from app.databus.social_intel import (
             detect_shill_campaigns,
             get_kol_leaderboard,
             get_kol_profile,
@@ -2800,7 +2793,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── MODEL REGISTRY - Smart free model routing, quality review ──
     try:
-        from app.domains.databus.model_registry import ai_call, get_usage_stats, review_content
+        from app.databus.model_registry import ai_call, get_usage_stats, review_content
 
         chains["ai_task"] = ProviderChain(
             "ai_task",
@@ -2856,7 +2849,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RAG INGESTION - nightly indexing, health checks ──
     try:
-        from app.domains.databus.rag_ingestion import nightly_rag_index, rag_health_check
+        from app.databus.rag_ingestion import nightly_rag_index, rag_health_check
 
         chains["rag_nightly"] = ProviderChain(
             "rag_nightly",

app/domains/databus/providers/btc.py

@@ -8,7 +8,7 @@ Re-exports the Bitcoin-scoped provider functions from
 Providers:
 - _blockchair_address, _blockchair_stats
 """
-from app.domains.databus.providers import (  # noqa: F401
+from app.databus.providers import (  # noqa: F401
     _blockchair_address,
     _blockchair_stats,
 )

app/domains/databus/providers/evm.py

@@ -12,7 +12,7 @@ Providers:
 - _moralis_wallet_net_worth
 - _etherscan_tx_trace
 """
-from app.domains.databus.providers import (  # noqa: F401
+from app.databus.providers import (  # noqa: F401
     _alchemy_token_balances,
     _alchemy_token_metadata,
     _etherscan_tx_trace,

app/domains/databus/providers/free_tier.py

@@ -15,8 +15,7 @@ Providers:
 - _passthrough_market_overview, _passthrough_trending, _passthrough_news,
   _passthrough_alerts, _passthrough_scanner, _passthrough_rag
 """
-from app.domains.databus.providers import (  # noqa: F401
-    _coindesk_news,
+from app.databus.providers import (  # noqa: F401
     _coingecko_price,
     _defillama_chains,
     _defillama_tvl,
@@ -27,11 +26,12 @@ from app.domains.databus.providers import (  # noqa: F401
     _dexscreener_trades,
     _local_token_price,
     _local_wallet_labels,
-    _messari_news,
     _passthrough_alerts,
     _passthrough_market_overview,
     _passthrough_news,
     _passthrough_rag,
     _passthrough_scanner,
     _passthrough_trending,
+    _coindesk_news,
+    _messari_news,
 )

app/domains/databus/providers/mcp_servers.py

@@ -5,6 +5,6 @@ Phase 3B of AUDIT-2026-Q3.md.
 Re-exports the MCP bridge provider function from
 ``app.databus.providers`` for cleaner import paths.
 """
-from app.domains.databus.providers import (  # noqa: F401
+from app.databus.providers import (  # noqa: F401
     _mcp_bridge,
 )

app/domains/databus/providers/paid_tier.py

@@ -12,7 +12,7 @@ Providers:
 - _dune_early_buyers
 - _virustotal_url_scan
 """
-from app.domains.databus.providers import (  # noqa: F401
+from app.databus.providers import (  # noqa: F401
     _arkham_counterparties,
     _arkham_entity,
     _arkham_intel_search,

app/domains/databus/providers/solana.py

@@ -9,7 +9,7 @@ Providers:
 - _birdeye_overview, _birdeye_price (premium Solana)
 - _solana_tracker_price, _solana_tracker_token, _solana_tracker_trending
 """
-from app.domains.databus.providers import (  # noqa: F401
+from app.databus.providers import (  # noqa: F401
     _birdeye_overview,
     _birdeye_price,
     _solana_tracker_price,

app/domains/intelligence/__init__.py

@@ -1,25 +0,0 @@
-"""Intelligence domain - public API.
-
-Phase 4 domain consolidation: intelligence-related modules -> app.domains.intelligence.
-"""
-from __future__ import annotations
-
-from app.domains.intelligence.alert_pipeline import (
-    get_active_alert_count,
-    get_recent_alerts,
-    push_alert,
-    run_alert_scan,
-    scan_known_scams,
-    scan_solana_new_pairs,
-)
-from app.domains.intelligence.intel_pipeline import run_intelligence_cycle
-
-__all__ = [
-    "get_active_alert_count",
-    "get_recent_alerts",
-    "push_alert",
-    "run_alert_scan",
-    "run_intelligence_cycle",
-    "scan_known_scams",
-    "scan_solana_new_pairs",
-]

app/domains/intelligence/alert_pipeline.py

@@ -1,370 +0,0 @@
-"""
-RMI Alert Pipeline - Real-time threat intelligence from scanners.
-=================================================================
-Feeds the Live Intel panel, WebSocket streams, and alert endpoints.
-
-Data sources:
-  - SENTINEL scanner (risk scores, scam detection)
-  - GoPlus security API (honeypot, tax, proxy checks)
-  - DexScreener new pairs (fresh launches to scan)
-  - Our own RAG scam patterns
-  - Wallet label cross-references
-
-Alert flow:
-  Scanner → alert_pipeline.push_alert() → Redis sorted set + pub/sub
-  Homepage reads from /api/v1/alerts/recent
-  Sidebar reads from /api/v1/alerts/count
-  WebSocket streams to connected clients
-"""
-
-import json
-import logging
-import os
-import time
-from datetime import UTC, datetime
-
-logger = logging.getLogger("alert_pipeline")
-
-REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
-REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
-REDIS_PW = os.getenv("REDIS_PASSWORD", "")
-
-ALERT_KEY = "rmi:alerts:recent"
-ALERT_COUNT_KEY = "rmi:alerts:count:active"
-ALERT_MAX = 500  # Keep last 500 alerts
-
-
-async def _get_redis():
-    """Get async Redis connection."""
-    import redis.asyncio as aioredis
-
-    return aioredis.Redis(
-        host=REDIS_HOST,
-        port=REDIS_PORT,
-        password=REDIS_PW or None,
-        decode_responses=True,
-    )
-
-
-async def get_active_alert_count() -> int:
-    """Get count of active (unacknowledged) alerts."""
-    try:
-        r = await _get_redis()
-        count = await r.zcard(ALERT_KEY)
-        await r.close()
-        return count
-    except Exception as e:
-        logger.debug(f"Alert count error: {e}")
-        return 0
-
-
-async def push_alert(
-    alert_type: str,
-    title: str,
-    description: str = "",
-    severity: str = "high",
-    chain: str = "unknown",
-    token: str = "",
-    token_symbol: str = "",
-    wallet: str = "",
-    risk_score: int = 0,
-    metadata: dict | None = None,
-) -> str:
-    """
-    Push a new alert to the pipeline.
-    Returns alert_id.
-    """
-    alert_id = f"alt_{int(time.time())}_{os.urandom(4).hex()}"
-
-    alert = {
-        "id": alert_id,
-        "type": alert_type,
-        "title": title,
-        "description": description,
-        "severity": severity,
-        "chain": chain,
-        "token": token,
-        "token_symbol": token_symbol,
-        "wallet": wallet,
-        "risk_score": risk_score,
-        "acknowledged": False,
-        "created_at": datetime.now(UTC).isoformat(),
-        **(metadata or {}),
-    }
-
-    try:
-        r = await _get_redis()
-        score = time.time()
-        await r.zadd(ALERT_KEY, {json.dumps(alert): score})
-        # Trim old alerts
-        await r.zremrangebyrank(ALERT_KEY, 0, -(ALERT_MAX + 1))
-        # Pub/sub for WebSocket streaming
-        await r.publish("rmi:ws:alerts", json.dumps(alert))
-        await r.close()
-        logger.info(f"Alert pushed: {alert_type} | {title[:60]}")
-    except Exception as e:
-        logger.error(f"Failed to push alert: {e}")
-
-    return alert_id
-
-
-async def get_recent_alerts(limit: int = 20, severity: str = "") -> list[dict]:
-    """Get recent alerts with optional severity filter. Normalizes old/new formats."""
-    try:
-        r = await _get_redis()
-        raw = await r.zrevrange(ALERT_KEY, 0, limit * 2 - 1)
-        await r.close()
-
-        alerts = []
-        for entry in raw:
-            try:
-                alert = json.loads(entry)
-
-                # Normalize old alert format to new
-                if "title" not in alert:
-                    alert["title"] = alert.get("message", alert.get("event", "Unknown alert"))
-                if "description" not in alert:
-                    desc_parts = []
-                    if alert.get("symbol"):
-                        desc_parts.append(f"Token: {alert['symbol']}")
-                    if alert.get("risk_score"):
-                        desc_parts.append(f"Risk: {alert['risk_score']}/100")
-                    flags = alert.get("risk_flags", [])
-                    if flags:
-                        desc_parts.append("; ".join(str(f)[:80] for f in flags[:2]))
-                    alert["description"] = " | ".join(desc_parts)
-                if "severity" not in alert:
-                    score = alert.get("risk_score", 50)
-                    alert["severity"] = "critical" if score >= 85 else "high" if score >= 65 else "medium"
-                if "chain" not in alert:
-                    alert["chain"] = alert.get("chain", "unknown")
-
-                if severity and alert.get("severity") != severity:
-                    continue
-                alerts.append(alert)
-                if len(alerts) >= limit:
-                    break
-            except json.JSONDecodeError:
-                pass
-
-        return alerts
-    except Exception as e:
-        logger.error(f"get_recent_alerts error: {e}")
-        return []
-
-
-# ── Alert generators - called by cron jobs or on-demand ──────────
-
-
-async def scan_solana_new_pairs(limit: int = 5) -> int:
-    """
-    Scan latest Solana pairs from DexScreener for scam patterns.
-    Pushes alerts for high-risk tokens.
-    Returns number of alerts generated.
-    """
-    import httpx
-
-    pushed = 0
-
-    try:
-        async with httpx.AsyncClient(timeout=15) as client:
-            resp = await client.get("https://api.dexscreener.com/latest/dex/search", params={"q": "SOL USDC"})
-            if resp.status_code != 200:
-                return 0
-
-            data = resp.json()
-            pairs = data.get("pairs", [])[:limit]
-
-            for pair in pairs:
-                token_addr = pair.get("baseToken", {}).get("address", "")
-                token_name = pair.get("baseToken", {}).get("name", "Unknown")
-                token_symbol = pair.get("baseToken", {}).get("symbol", "???")
-                liquidity = float(pair.get("liquidity", {}).get("usd", 0) or 0)
-                volume = float(pair.get("volume", {}).get("h24", 0) or 0)
-                price_change = float(pair.get("priceChange", {}).get("h24", 0) or 0)
-                created_at = pair.get("pairCreatedAt", 0)
-                age_hours = (time.time() - created_at / 1000) / 3600 if created_at else 999
-
-                # Risk heuristics
-                risk_flags = []
-
-                if liquidity < 1000:
-                    risk_flags.append("low_liquidity")
-                if volume == 0:
-                    risk_flags.append("no_volume")
-                if price_change < -80:
-                    risk_flags.append(f"crashed_{abs(price_change):.0f}%")
-                if age_hours < 1 and liquidity < 5000:
-                    risk_flags.append("fresh_launch_low_liq")
-                if price_change > 500:
-                    risk_flags.append(f"pumped_{price_change:.0f}%")
-
-                if risk_flags:
-                    await push_alert(
-                        alert_type="new_pair_risk",
-                        title=f"{token_symbol}: {' | '.join(risk_flags[:2])}",
-                        description=f"New pair {token_name} ({token_symbol}) on Solana. "
-                        f"Liquidity: ${liquidity:,.0f}, Age: {age_hours:.1f}h, "
-                        f"24h change: {price_change:+.1f}%",
-                        severity="critical" if len(risk_flags) >= 3 else "high",
-                        chain="solana",
-                        token=token_addr,
-                        token_symbol=token_symbol,
-                        risk_score=min(90, len(risk_flags) * 25),
-                        metadata={
-                            "risk_flags": risk_flags,
-                            "liquidity_usd": liquidity,
-                            "age_hours": age_hours,
-                        },
-                    )
-                    pushed += 1
-
-        return pushed
-    except Exception as e:
-        logger.warning(f"Solana scan error: {e}")
-        return pushed
-
-
-async def scan_known_scams(limit: int = 3) -> int:
-    """
-    Check our RAG known_scams collection for recently added entries.
-    Pushes alerts for new scam patterns detected.
-    """
-    pushed = 0
-    try:
-        r = await _get_redis()
-        # Check for recent scam pattern additions
-        scam_ids = await r.smembers("rag:idx:known_scams")
-
-        recent_count = 0
-        for sid in list(scam_ids)[:20]:
-            doc = await r.get(f"rag:known_scams:{sid}")
-            if doc:
-                try:
-                    data = json.loads(doc)
-                    added = data.get("metadata", {}).get("added_at", "")
-                    if added:
-                        age_h = (time.time() - datetime.fromisoformat(added.replace("Z", "+00:00")).timestamp()) / 3600
-                        if age_h < 24:
-                            recent_count += 1
-                except Exception:
-                    pass
-
-        await r.close()
-
-        if recent_count > 0:
-            await push_alert(
-                alert_type="scam_pattern_update",
-                title=f"{recent_count} new scam patterns detected in last 24h",
-                description="New rug pull, honeypot, or phishing patterns added to the knowledge base.",
-                severity="high",
-                chain="multi",
-                risk_score=85,
-            )
-            pushed += 1
-
-        return pushed
-    except Exception as e:
-        logger.warning(f"Known scams scan error: {e}")
-        return pushed
-
-
-async def run_alert_scan() -> dict[str, int]:
-    """
-    Run a full alert scan across all sources.
-    Called by cron job every 15 minutes.
-    """
-    results = {}
-
-    # Scan Solana new pairs
-    results["solana_pairs"] = await scan_solana_new_pairs(limit=8)
-
-    # Scan known scams
-    results["known_scams"] = await scan_known_scams()
-
-    total = sum(results.values())
-    logger.info(f"Alert scan complete: {total} new alerts ({results})")
-    return results
-
-
-# ── Seed some initial alerts if Redis is empty ────────────────────
-
-
-async def seed_initial_alerts():
-    """Seed baseline alerts so the system isn't empty on first run."""
-    r = await _get_redis()
-    existing = await r.zcard(ALERT_KEY)
-    await r.close()
-
-    if existing > 0:
-        return  # Already has alerts
-
-    seeds = [
-        (
-            "honeypot",
-            "Honeypot detected on Base: 0xdead...",
-            "Token has sell restrictions and blacklist. Buyers cannot exit.",
-            "critical",
-            "base",
-        ),
-        (
-            "whale",
-            "Whale moved 5M USDC to Binance",
-            "Wallet 0xABCD... transferred $5M USDC to Binance hot wallet. Possible sell pressure.",
-            "high",
-            "ethereum",
-        ),
-        (
-            "rugpull",
-            "Liquidity removed from $SCAM token",
-            "100% of liquidity pool withdrawn by deployer. Token is now worthless.",
-            "critical",
-            "solana",
-        ),
-        (
-            "bundler",
-            "Sniper bundle detected: $NEWLAUNCH",
-            "Coordinated wallet cluster bought 60% of supply in first block.",
-            "high",
-            "solana",
-        ),
-        (
-            "contract",
-            "Unverified proxy contract found",
-            "Token uses upgradeable proxy with unverified implementation. Owner can change logic.",
-            "high",
-            "base",
-        ),
-        (
-            "concentration",
-            "90% supply held by 3 wallets on $DANGER",
-            "Extreme holder concentration. Classic rug pull setup.",
-            "critical",
-            "ethereum",
-        ),
-        (
-            "wash_trade",
-            "Wash trading detected on $FAKEVOL",
-            "95% of volume is self-trading between linked wallets.",
-            "high",
-            "bsc",
-        ),
-        (
-            "phishing",
-            "Fake airdrop targeting $BONK holders",
-            "Phishing site detected posing as official BONK airdrop. Users losing funds.",
-            "critical",
-            "solana",
-        ),
-    ]
-
-    for alert_type, title, desc, severity, chain in seeds:
-        await push_alert(
-            alert_type=alert_type,
-            title=title,
-            description=desc,
-            severity=severity,
-            chain=chain,
-        )
-
-    logger.info(f"Seeded {len(seeds)} initial alerts")