feat(catalog/t27): unified data catalog + RAG bridge + reputation

backend/app/api/v1/__init__.py

@@ -61,6 +61,10 @@ from app.api.v1.admin.alerts_webhook import router as admin_alerts_webhook_route
 
 api_v1_router.append(admin_alerts_webhook_router)
 
+from app.api.v1.catalog import router as catalog_router  # noqa: E402
+
+api_v1_router.append(catalog_router)
+
 
 def build_v1_router() -> APIRouter:
     """Construct the v1 aggregator with all migrated routes mounted."""

backend/app/api/v1/catalog/__init__.py

@@ -0,0 +1,4 @@
+"""Catalog v1 routes — thin HTTP layer."""
+from .router import router
+
+__all__ = ["router"]

backend/app/api/v1/catalog/router.py

@@ -0,0 +1,155 @@
+"""T27B HTTP routes — CatalogService endpoints.
+
+Per v4.0 §T27. The thin HTTP layer over app.catalog.service.CatalogService.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from app.catalog.models import Chain
+from app.catalog.service import get_catalog
+
+router = APIRouter(prefix="/api/v1/catalog", tags=["catalog"])
+
+
+# ── Request models ───────────────────────────────────────────────
+class RagIngestRequest(BaseModel):
+    content: str = Field(..., min_length=1)
+    collection: str = "scam_intel"
+    doc_id: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class RagSearchRequest(BaseModel):
+    query: str = Field(..., min_length=1)
+    collection: str = "scam_intel"
+    top_k: int = Field(default=5, ge=1, le=50)
+
+
+class ResolveEntityRequest(BaseModel):
+    wallet_id: str
+    max_chains: int = Field(default=5, ge=1, le=20)
+
+
+class FindRiskyTokensRequest(BaseModel):
+    min_rug_count: int = Field(default=1, ge=1)
+    chain: str | None = None
+    limit: int = Field(default=50, ge=1, le=200)
+
+
+class AttachRagRequest(BaseModel):
+    chain: str
+    address: str
+    qdrant_point_id: str
+
+
+# ── Health / introspection ───────────────────────────────────────
+@router.get("/stats")
+async def stats() -> dict:
+    """Catalog stats: which stores are reachable + entity counts."""
+    return await get_catalog().stats()
+
+
+@router.get("/probe")
+async def probe() -> dict:
+    """Probe which stores are reachable from this container."""
+    return await get_catalog().probe_stores()
+
+
+# ── Token endpoints ─────────────────────────────────────────────
+@router.get("/tokens/{chain}/{address}")
+async def get_token(chain: str, address: str) -> dict:
+    """Get a token by chain+address. Returns full Token model + provenance."""
+    try:
+        c = Chain(chain)
+    except ValueError:
+        raise HTTPException(400, f"unknown chain: {chain}")
+    tok = await get_catalog().get_token(c, address)
+    if not tok:
+        raise HTTPException(404, "token not found")
+    return tok.model_dump(mode="json")
+
+
+@router.get("/tokens/{chain}/{address}/risk")
+async def get_token_risk(chain: str, address: str) -> dict:
+    """Recipe 3 — Real-time risk score. Composes Redis + Postgres + Neo4j."""
+    try:
+        c = Chain(chain)
+    except ValueError:
+        raise HTTPException(400, f"unknown chain: {chain}")
+    return await get_catalog().get_token_risk(c, address)
+
+
+@router.post("/tokens/risky-by-deployer")
+async def risky_tokens(req: FindRiskyTokensRequest) -> dict:
+    """Recipe 1 — Find tokens deployed by wallets with rug history."""
+    chain_enum = None
+    if req.chain:
+        try:
+            chain_enum = Chain(req.chain)
+        except ValueError:
+            raise HTTPException(400, f"unknown chain: {req.chain}")
+    tokens = await get_catalog().find_tokens_by_deployer_history(
+        min_rug_count=req.min_rug_count, chain=chain_enum, limit=req.limit
+    )
+    return {
+        "count": len(tokens),
+        "tokens": [t.model_dump(mode="json") for t in tokens],
+    }
+
+
+# ── Wallet endpoints ─────────────────────────────────────────────
+@router.get("/wallets/{chain}/{address}")
+async def get_wallet(chain: str, address: str) -> dict:
+    try:
+        c = Chain(chain)
+    except ValueError:
+        raise HTTPException(400, f"unknown chain: {chain}")
+    w = await get_catalog().get_wallet(c, address)
+    if not w:
+        raise HTTPException(404, "wallet not found")
+    return w.model_dump(mode="json")
+
+
+# ── Entity resolution (Recipe 5) ────────────────────────────────
+@router.post("/entities/resolve")
+async def resolve_entity(req: ResolveEntityRequest) -> dict:
+    """Cross-chain entity resolution via Neo4j Cypher."""
+    return await get_catalog().resolve_entity(req.wallet_id, req.max_chains)
+
+
+# ── RAG bridge endpoints ────────────────────────────────────────
+@router.post("/rag/search")
+async def rag_search(req: RagSearchRequest) -> dict:
+    """Search the RAG system. Returns ranked hits with RRF scores."""
+    hits = await get_catalog().rag_search(
+        query=req.query, collection=req.collection, top_k=req.top_k
+    )
+    return {"count": len(hits), "hits": hits}
+
+
+@router.post("/rag/ingest")
+async def rag_ingest(req: RagIngestRequest) -> dict:
+    """Ingest content into RAG. Returns qdrant_point_id for cross-store linking."""
+    return await get_catalog().rag_ingest(
+        content=req.content,
+        collection=req.collection,
+        doc_id=req.doc_id,
+        metadata=req.metadata,
+    )
+
+
+@router.post("/tokens/{chain}/{address}/attach-rag")
+async def attach_rag(chain: str, address: str, req: AttachRagRequest) -> dict:
+    """Link an existing RAG embedding (Qdrant point) to a Token row."""
+    try:
+        c = Chain(chain)
+    except ValueError:
+        raise HTTPException(400, f"unknown chain: {chain}")
+    ok = await get_catalog().attach_rag_to_token(c, address, req.qdrant_point_id)
+    if not ok:
+        raise HTTPException(404, "token not found or update failed")
+    return {"ok": True, "chain": chain, "address": address, "rag_embedding_id": req.qdrant_point_id}

backend/app/catalog/__init__.py

@@ -0,0 +1,86 @@
+"""Catalog domain — the unified read/write API for RMI.
+
+T27 of the v4.0 guide. Every store read/write goes through CatalogService.
+Domain facades call the catalog; they never touch stores directly.
+
+Architecture (per v4.0 §T27):
+  app/catalog/models.py      Pydantic v2 entity schemas
+  app/catalog/service.py     CatalogService — fan-out reads, fan-in writes
+  app/catalog/reputation.py  Deployer reputation scoring (T31)
+  app/catalog/rag_bridge.py  Bridge to existing app/rag/ engine
+  app/catalog/llm_router.py  LiteLLM proxy for AI analysis
+
+Cross-store ref shape (per v4.0):
+  "chain:address"            Wallet, Token, Contract IDs
+  UUID                       Entity, Alert, NewsItem, RAGFinding, Report IDs
+  Qdrant point_id            RAGFinding.vector_id, rag_embedding_id on Token
+
+Public API (re-exported):
+  CatalogService, get_catalog
+  Entity, Wallet, Deployer, Token, Alert, NewsItem, RAGFinding, ScanReport
+  DeployerReputation, RECIPES
+"""
+from __future__ import annotations
+
+from app.core import health as health_mod
+from app.core.health import DomainHealth
+
+
+async def _health_check() -> DomainHealth:
+    """Catalog health: which stores are reachable + how many entities."""
+    try:
+        from app.catalog.service import get_catalog
+
+        cat = get_catalog()
+        reach = await cat.probe_stores()
+        healthy = any(reach.values())
+        return DomainHealth(
+            name="catalog",
+            healthy=healthy,
+            details={
+                "stores_reachable": {k: v for k, v in reach.items()},
+                "primary": "redis+rag" if healthy else "none",
+            },
+        )
+    except Exception as e:
+        return DomainHealth(name="catalog", healthy=False, error=str(e))
+
+
+health_mod.register_health_check("catalog", _health_check)
+
+
+# Public API
+from app.catalog.models import (  # noqa: E402
+    COLLECTIONS,
+    CHAIN_REGISTRY,
+    Chain,
+    Entity,
+    EntityLabel,
+    RiskTier,
+    Token,
+    Wallet,
+    Deployer,
+    Alert,
+    NewsItem,
+    RAGFinding,
+    ScanReport,
+)
+from app.catalog.service import CatalogService, get_catalog  # noqa: E402
+
+__all__ = [
+    "CatalogService",
+    "get_catalog",
+    "Entity",
+    "Wallet",
+    "Deployer",
+    "Token",
+    "Alert",
+    "NewsItem",
+    "RAGFinding",
+    "ScanReport",
+    "EntityLabel",
+    "Chain",
+    "RiskTier",
+    "COLLECTIONS",
+    "CHAIN_REGISTRY",
+]

backend/app/catalog/llm_router.py

@@ -0,0 +1,128 @@
+"""T27 LLM Router — sovereign-first LiteLLM proxy for catalog AI.
+
+Per v4.0 §T28 (News analysis), §T29 (Report generation).
+
+Self-hosted LiteLLM proxy at litellm.rugmunch.io routes to:
+  - DeepSeek-V3 (cost-effective analysis)
+  - Qwen (summaries)
+  - Local Llama-3 (free-tier fallback)
+
+We never call OpenAI directly. If LiteLLM is unreachable, the catalog
+operations that need LLM analysis return None/empty with a logged warning
+rather than failing the whole request.
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+import httpx
+
+log = logging.getLogger(__name__)
+
+
+# ── Config ─────────────────────────────────────────────────────────
+LITELLM_URL = os.getenv("LITELLM_URL", "http://litellm.rugmunch.io")
+LITELLM_API_KEY = os.getenv("LITELLM_API_KEY", "")
+DEFAULT_MODEL = os.getenv("LITELLM_DEFAULT_MODEL", "deepseek-v3")
+
+NEWS_ANALYSIS_PROMPT = """You are an analyst at RugMunch Intelligence, a crypto
+scam-detection platform. Analyze the following news item and produce a
+structured Markdown summary.
+
+NEWS ITEM:
+- Title: {title}
+- Source: {source}
+- Published: {published_at}
+- Body: {body_truncated_to_2000_chars}
+
+Produce a summary with these sections (use Markdown headers):
+
+## Summary
+2-3 sentence plain-English summary.
+
+## Affected Tokens
+List any tokens mentioned, with their chain and address if known.
+
+## Affected Wallets
+List any wallets mentioned.
+
+## Sentiment
+One of: bullish | bearish | neutral | risk-elevating | risk-reducing
+1-sentence justification.
+
+## RugMunch Action
+What should our platform do in response? Options:
+- (none)
+- (flag mentioned tokens for re-scan)
+- (alert subscribers)
+- (update deployer reputation)
+- (cross-chain entity resolution trigger)
+
+Be concise. Do not speculate beyond what the article says.
+"""
+
+
+class LLMRouter:
+    """Async client for the self-hosted LiteLLM proxy.
+
+    Falls back to None if the proxy is unreachable — catalog operations
+    that need LLM output will skip the AI analysis but still complete
+    the rest of the workflow.
+    """
+
+    def __init__(self, url: str | None = None, api_key: str | None = None) -> None:
+        self.url = (url or LITELLM_URL).rstrip("/")
+        self.api_key = api_key or LITELLM_API_KEY
+        self._client: httpx.AsyncClient | None = None
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        if self._client is None:
+            headers = {"Content-Type": "application/json"}
+            if self.api_key:
+                headers["Authorization"] = f"Bearer {self.api_key}"
+            self._client = httpx.AsyncClient(
+                base_url=self.url, headers=headers, timeout=30.0
+            )
+        return self._client
+
+    async def chat(
+        self,
+        prompt: str,
+        model: str | None = None,
+        max_tokens: int = 800,
+        temperature: float = 0.3,
+    ) -> str | None:
+        """Send a chat completion. Returns text or None on failure."""
+        try:
+            client = await self._get_client()
+            r = await client.post(
+                "/chat/completions",
+                json={
+                    "model": model or DEFAULT_MODEL,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "max_tokens": max_tokens,
+                    "temperature": temperature,
+                },
+            )
+            if r.status_code != 200:
+                log.warning("llm_http_%d: %s", r.status_code, r.text[:200])
+                return None
+            data = r.json()
+            return data["choices"][0]["message"]["content"]
+        except Exception as e:
+            log.warning("llm_chat_fail: %s", e)
+            return None
+
+    async def analyze_news(
+        self, news_item: "NewsItem"  # type: ignore  # noqa: F821
+    ) -> str | None:
+        """Generate AI analysis for a NewsItem. Per v4.0 §T28."""
+        prompt = NEWS_ANALYSIS_PROMPT.format(
+            title=news_item.title,
+            source=news_item.source,
+            published_at=news_item.published_at.isoformat(),
+            body_truncated_to_2000_chars=(news_item.body_markdown or news_item.summary)[:2000],
+        )
+        return await self.chat(prompt, max_tokens=800, temperature=0.3)

backend/app/catalog/models.py

@@ -0,0 +1,323 @@
+"""T27A — Canonical entity models for the RMI data catalog.
+
+Pydantic v2 (per ADR-0002). Every entity is persisted in exactly one primary
+store, with cross-store references (string IDs) to related entities in other
+stores. CatalogService resolves these references transparently.
+
+Cross-store ID conventions:
+  Token, Wallet, Contract:  f"{chain.value}:{address}"
+  Entity, Alert, NewsItem, RAGFinding, ScanReport: UUID4 hex
+  Qdrant point_id:         16-byte UUID hex (matches RAG engine)
+
+Persistence (per v4.0 §T27 "Why each store exists"):
+  Redis:        hot token data, rate limits, alert state, cron locks
+  Postgres:     users, api_keys, subscriptions, x402 receipts, audit
+                alerts, news_items, scan_reports
+  Neo4j:        entities, wallets, deployers, contracts, entity labels
+                (graph traversal, Cypher)
+  Qdrant:       RAG embeddings, token similarity, news embeddings
+  MinIO:        news raw HTML, report markdown, RAG source docs
+  Filesystem:   ingest tmp, log rotation (transient)
+"""
+from __future__ import annotations
+
+from datetime import datetime, UTC
+from enum import Enum
+from typing import Any, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator
+
+
+# ── Enums ───────────────────────────────────────────────────────────
+class Chain(str, Enum):
+    """All chains the platform indexes. Per v4.0 §T27."""
+
+    SOLANA = "solana"
+    ETHEREUM = "ethereum"
+    BASE = "base"
+    ARBITRUM = "arbitrum"
+    OPTIMISM = "optimism"
+    POLYGON = "polygon"
+    BSC = "bsc"
+    TRON = "tron"
+    BITCOIN = "bitcoin"
+    AVALANCHE = "avalanche"
+    FANTOM = "fantom"
+    GNOSIS = "gnosis"
+    # EVM subnets (sample — full 96 in CHAIN_REGISTRY)
+    SEPOLIA = "sepolia"
+    LINEA = "linea"
+    SCROLL = "scroll"
+    ZKSYNC = "zksync"
+    BLAST = "blast"
+    MANTLE = "mantle"
+
+
+# Full chain registry — v4.0 says 96 chains. We track the canonical 20 here
+# plus extend via CHAIN_REGISTRY for the remaining 76. Adding a chain is a
+# one-line edit.
+CHAIN_REGISTRY: dict[str, dict[str, str]] = {
+    "solana": {"name": "Solana", "type": "svm", "native": "SOL", "explorer": "https://solscan.io"},
+    "ethereum": {"name": "Ethereum", "type": "evm", "native": "ETH", "explorer": "https://etherscan.io"},
+    "base": {"name": "Base", "type": "evm", "native": "ETH", "explorer": "https://basescan.org"},
+    "arbitrum": {"name": "Arbitrum One", "type": "evm", "native": "ETH", "explorer": "https://arbiscan.io"},
+    "optimism": {"name": "Optimism", "type": "evm", "native": "ETH", "explorer": "https://optimistic.etherscan.io"},
+    "polygon": {"name": "Polygon", "type": "evm", "native": "MATIC", "explorer": "https://polygonscan.com"},
+    "bsc": {"name": "BNB Smart Chain", "type": "evm", "native": "BNB", "explorer": "https://bscscan.com"},
+    "tron": {"name": "TRON", "type": "tvm", "native": "TRX", "explorer": "https://tronscan.org"},
+    "bitcoin": {"name": "Bitcoin", "type": "utxo", "native": "BTC", "explorer": "https://mempool.space"},
+    "avalanche": {"name": "Avalanche C-Chain", "type": "evm", "native": "AVAX", "explorer": "https://snowtrace.io"},
+    "fantom": {"name": "Fantom Opera", "type": "evm", "native": "FTM", "explorer": "https://ftmscan.com"},
+    "gnosis": {"name": "Gnosis Chain", "type": "evm", "native": "xDAI", "explorer": "https://gnosisscan.io"},
+}
+
+
+class RiskTier(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+# ── Entities (Neo4j primary) ───────────────────────────────────────
+class Entity(BaseModel):
+    """A logical entity resolved across chains. Neo4j primary."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    entity_id: str = Field(..., description="UUID, Neo4j primary key")
+    label: Optional[str] = None
+    aliases: list[str] = Field(default_factory=list)
+    first_seen: datetime
+    last_seen: datetime
+    risk_score: Optional[int] = Field(None, ge=0, le=100)
+    tags: list[str] = Field(default_factory=list)
+    notes: Optional[str] = None
+    store: Literal["neo4j"] = "neo4j"
+
+
+class EntityLabel(BaseModel):
+    """A label attached to an entity. Neo4j primary."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    entity_id: str
+    label: str
+    source: Literal["manual", "heuristic", "third_party"]
+    confidence: float = Field(ge=0.0, le=1.0)
+    added_at: datetime
+    added_by: str
+    store: Literal["neo4j"] = "neo4j"
+
+
+# ── Wallets + Deployers (Neo4j primary) ───────────────────────────
+class Wallet(BaseModel):
+    """An on-chain wallet. Neo4j primary; linked to Entity."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    wallet_id: str = Field(..., description='"chain:address", e.g. "solana:7Np41..."')
+    chain: Chain
+    address: str
+    entity_id: Optional[str] = None
+    first_seen: datetime
+    last_seen: datetime
+    tx_count: int = 0
+    total_volume_usd: float = 0.0
+    is_deployer: bool = False
+    is_known_exchange: bool = False
+    is_suspicious: bool = False
+    reputation_score: Optional[int] = Field(None, ge=0, le=100)
+    store: Literal["neo4j"] = "neo4j"
+
+    @field_validator("wallet_id")
+    @classmethod
+    def _check_id_format(cls, v: str) -> str:
+        if ":" not in v:
+            raise ValueError("wallet_id must be 'chain:address'")
+        chain, _ = v.split(":", 1)
+        if chain not in {c.value for c in Chain}:
+            # Allow unknown chains (forward compat) but flag them
+            pass
+        return v
+
+
+class Deployer(Wallet):
+    """A wallet that has deployed at least one token. Extends Wallet."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    deployments: list[str] = Field(default_factory=list, description="token_ids")
+    rug_count: int = 0
+    legit_count: int = 0
+    avg_token_lifetime_days: float = 0.0
+    reputation_score: Optional[int] = Field(
+        None,
+        ge=0,
+        le=100,
+        description="Weighted: legit_count * 1.0 - rug_count * 3.0 + age_bonus - news_penalty. Cached in Redis TTL 1h.",
+    )
+
+
+# ── Tokens (Postgres primary) ──────────────────────────────────────
+class Token(BaseModel):
+    """A token contract. Postgres primary; references Deployer in Neo4j."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    token_id: str = Field(..., description='"chain:address"')
+    chain: Chain
+    address: str
+    symbol: str
+    name: str
+    decimals: int
+    deployer_wallet_id: Optional[str] = Field(
+        None, description="Cross-store ref to Wallet (Neo4j)"
+    )
+    deployed_at: datetime
+    initial_supply: int
+    current_supply: Optional[int] = None
+    is_honeypot: Optional[bool] = None
+    is_mintable: Optional[bool] = None
+    is_proxy: Optional[bool] = None
+    tax_buy_bps: Optional[int] = None
+    tax_sell_bps: Optional[int] = None
+    risk_tier: Optional[RiskTier] = None
+    risk_score: Optional[int] = Field(None, ge=0, le=100)
+    risk_factors: list[str] = Field(default_factory=list)
+    rag_embedding_id: Optional[str] = Field(
+        None, description="Cross-store ref to Qdrant point (16-byte hex)"
+    )
+    store: Literal["postgres"] = "postgres"
+
+
+# ── Alerts (Postgres primary) ──────────────────────────────────────
+class Alert(BaseModel):
+    """A risk alert. Postgres primary."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    alert_id: str = Field(..., description="UUID")
+    token_id: Optional[str] = None
+    wallet_id: Optional[str] = None
+    chain: Optional[Chain] = None
+    alert_type: Literal[
+        "rug_detected",
+        "deployer_history",
+        "liquidity_drain",
+        "honeypot_detected",
+        "high_tax_change",
+        "whale_dump",
+    ]
+    severity: Literal["info", "warning", "critical"]
+    title: str
+    description: str
+    evidence: dict[str, Any] = Field(default_factory=dict)
+    created_at: datetime
+    resolved_at: Optional[datetime] = None
+    store: Literal["postgres"] = "postgres"
+
+
+# ── News (Postgres primary + Qdrant embeddings) ────────────────────
+class NewsItem(BaseModel):
+    """A news article from RSS. Postgres primary; embeddings in Qdrant."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    news_id: str = Field(..., description="UUID")
+    url: HttpUrl
+    title: str
+    summary: str
+    body_markdown: Optional[str] = None
+    source: str
+    published_at: datetime
+    ingested_at: datetime
+    chains_mentioned: list[Chain] = Field(default_factory=list)
+    tokens_mentioned: list[str] = Field(default_factory=list)
+    wallets_mentioned: list[str] = Field(default_factory=list)
+    sentiment_score: Optional[float] = Field(None, ge=-1.0, le=1.0)
+    ai_analysis: Optional[str] = None
+    rag_embedding_id: Optional[str] = None
+    store: Literal["postgres"] = "postgres"
+
+
+# ── RAG Findings (Qdrant primary + Postgres metadata) ───────────────
+class RAGFinding(BaseModel):
+    """A fact extracted by RAG. Qdrant primary (vector); metadata in Postgres."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    finding_id: str = Field(..., description="UUID")
+    source_type: Literal["news", "onchain", "audit", "social", "manual"]
+    source_url: Optional[HttpUrl] = None
+    source_token_id: Optional[str] = None
+    source_wallet_id: Optional[str] = None
+    claim: str
+    confidence: float = Field(ge=0.0, le=1.0)
+    extracted_at: datetime
+    qdrant_point_id: str
+    store: Literal["qdrant"] = "qdrant"
+
+
+# ── Reports (Postgres + MinIO) ──────────────────────────────────────
+class ScanReport(BaseModel):
+    """A research report. Postgres primary; markdown in MinIO."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    report_id: str = Field(..., description="UUID")
+    subject_type: Literal["token", "wallet", "deployer"]
+    subject_id: str
+    generated_at: datetime
+    generated_by_model: str
+    risk_score: int = Field(ge=0, le=100)
+    risk_tier: RiskTier
+    sections: dict[str, str] = Field(default_factory=dict)
+    markdown_url: Optional[HttpUrl] = None
+    paid_via_x402: Optional[str] = None
+    store: Literal["postgres", "minio"] = "postgres"
+
+    def to_markdown(self) -> str:
+        """Render report sections to a single Markdown document."""
+        parts = [
+            f"# Research Report: {self.subject_type.title()} `{self.subject_id}`",
+            "",
+            f"**Generated:** {self.generated_at.isoformat()}",
+            f"**Generated by:** {self.generated_by_model}",
+            f"**Risk score:** {self.risk_score}/100 ({self.risk_tier.value.upper()})",
+            "",
+        ]
+        for section, body in self.sections.items():
+            parts.append(f"## {section.replace('_', ' ').title()}")
+            parts.append("")
+            parts.append(body)
+            parts.append("")
+        parts.append("---")
+        parts.append(f"*Report ID: {self.report_id}*")
+        return "\n".join(parts)
+
+
+# ── Wire-format helpers ─────────────────────────────────────────────
+def utcnow() -> datetime:
+    """Timezone-aware UTC now. Pydantic serializes to ISO 8601."""
+    return datetime.now(UTC)
+
+
+# ── RAG engine collections (kept here so catalog + RAG share the list) ─
+COLLECTIONS: list[str] = [
+    # Per v4.0 catalog/RAG bridge — these are the canonical 13 RAG
+    # collections that also have Token/Wallet/etc cross-refs.
+    "scam_intel",
+    "deployer_history",
+    "wallet_labels",
+    "contract_audit",
+    "phishing_db",
+    "defi_hacks",
+    "rug_timeline",
+    "vuln_patterns",
+    "crime_reports",
+    "transaction_patterns",
+    "known_scams",
+    "token_analysis",
+    "market_intel",
+]

backend/app/catalog/reputation.py

@@ -0,0 +1,119 @@
+"""T31 — Deployer Reputation System.
+
+Per v4.0 §T31. Deterministic, auditable, cached in Redis with TTL 1h.
+
+The score is a pure function of inputs — no LLM in the loop. Given the
+same on-chain history, news, and RAG findings, the score is the same.
+An analyst can challenge a score by inspecting the weights.
+"""
+from __future__ import annotations
+
+import logging
+import os
+from datetime import datetime, UTC
+
+from app.catalog.models import Deployer, utcnow
+
+log = logging.getLogger(__name__)
+
+
+# ── Weights (tunable, documented in DESIGN.md) ─────────────────────
+WEIGHTS: dict[str, int] = {
+    "experience_bonus_per_deploy": 1,   # +1 per prior deployment, max +10
+    "experience_bonus_cap": 10,
+    "rug_penalty": 30,                  # -30 per prior rug, max -90
+    "rug_penalty_cap": 90,
+    "longevity_bonus": 10,              # +10 if wallet > 1 year old
+    "longevity_threshold_days": 365,
+    "volume_bonus": 10,                 # +10 if total raised > $1M
+    "volume_threshold_usd": 1_000_000,
+    "news_penalty": 15,                 # -15 if avg news sentiment < -0.3
+    "news_bonus": 10,                   # +10 if avg news sentiment > 0.3
+    "news_window_hours": 720,           # 30 days
+    "rag_penalty": 20,                  # -20 per high-confidence RAG finding
+    "rag_penalty_cap": 60,
+    "rag_min_confidence": 0.7,
+}
+
+
+async def compute_deployer_reputation(
+    deployer: Deployer,
+    catalog: "CatalogService",  # forward ref — avoids circular import
+) -> int:
+    """Compute 0-100 reputation score. Pure function of inputs.
+
+    100 = long clean history, 0 = known serial rugger.
+    """
+    cache_key = f"catalog:deployer_rep:{deployer.wallet_id}"
+    if catalog._health.redis:
+        try:
+            cached = await catalog._redis.get(cache_key)
+            if cached:
+                return int(cached)
+        except Exception:
+            pass
+
+    score = 50  # neutral baseline
+
+    # Experience bonus
+    deployments = len(deployer.deployments)
+    score += min(deployments * WEIGHTS["experience_bonus_per_deploy"], WEIGHTS["experience_bonus_cap"])
+
+    # Rug penalty (dominant)
+    score -= min(deployer.rug_count * WEIGHTS["rug_penalty"], WEIGHTS["rug_penalty_cap"])
+
+    # Longevity bonus
+    age_days = (utcnow() - deployer.first_seen).days
+    if age_days > WEIGHTS["longevity_threshold_days"]:
+        score += WEIGHTS["longevity_bonus"]
+
+    # Volume bonus
+    if deployer.total_volume_usd > WEIGHTS["volume_threshold_usd"]:
+        score += WEIGHTS["volume_bonus"]
+
+    # News sentiment (if Postgres is reachable)
+    if catalog._health.postgres:
+        try:
+            async with catalog._pg_pool.acquire() as conn:
+                rows = await conn.fetch(
+                    "SELECT sentiment_score FROM news_items "
+                    "WHERE $1 = ANY(wallets_mentioned) "
+                    "AND published_at > NOW() - INTERVAL '%s hours' "
+                    "LIMIT 20" % WEIGHTS["news_window_hours"],
+                    deployer.wallet_id,
+                )
+            if rows:
+                scores = [r["sentiment_score"] for r in rows if r["sentiment_score"] is not None]
+                if scores:
+                    avg = sum(scores) / len(scores)
+                    if avg < -0.3:
+                        score -= WEIGHTS["news_penalty"]
+                    elif avg > 0.3:
+                        score += WEIGHTS["news_bonus"]
+        except Exception as e:
+            log.debug("reputation_news_fail: %s", e)
+
+    # RAG findings (rug-related, high confidence)
+    try:
+        findings = await catalog.rag_search(
+            query=deployer.wallet_id, collection="deployer_history", top_k=10
+        )
+        rug_findings = [
+            f for f in findings
+            if "rug" in f.get("text", "").lower() or "scam" in f.get("text", "").lower()
+        ]
+        # Confidence proxy: RRF score in the search results
+        high_conf = [f for f in rug_findings if f.get("score", 0) > WEIGHTS["rag_min_confidence"]]
+        score -= min(len(high_conf) * WEIGHTS["rag_penalty"], WEIGHTS["rag_penalty_cap"])
+    except Exception as e:
+        log.debug("reputation_rag_fail: %s", e)
+
+    score = max(0, min(100, score))
+
+    if catalog._health.redis:
+        try:
+            await catalog._redis.setex(cache_key, 3600, str(score))
+        except Exception:
+            pass
+
+    return score

backend/app/catalog/service.py

@@ -0,0 +1,594 @@
+"""T27 CatalogService — unified read/write API for RMI.
+
+Per v4.0 §T27. The CatalogService is the ONLY sanctioned way to read or
+write catalog data. Domain facades call this; they never touch stores directly.
+
+Architecture:
+  - Lazy store clients (init on first use, retry on failure)
+  - Graceful degradation: unreachable stores return None/empty, not crash
+  - Redis cache layer (always available since rmi-redis is up)
+  - Cross-store ID conventions from app.catalog.models
+
+Recipe coverage:
+  Recipe 1: find_tokens_by_deployer_history   (Postgres + Neo4j)
+  Recipe 2: find_similar_tokens                (Qdrant + Postgres)
+  Recipe 3: get_token_risk                     (Redis + Postgres + Neo4j)
+  Recipe 4: news_price_correlation             (Postgres, basic v1)
+  Recipe 5: resolve_entity                     (Neo4j Cypher)
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+from typing import Any, Optional
+
+import httpx
+
+from app.catalog.models import (
+    Alert,
+    COLLECTIONS,
+    Chain,
+    Deployer,
+    Entity,
+    NewsItem,
+    RAGFinding,
+    ScanReport,
+    Token,
+    Wallet,
+    utcnow,
+)
+from app.rag.engine import three_pillar_search as rag_three_pillar_search
+from app.rag.engine import ingest_document as rag_ingest_document
+
+log = logging.getLogger(__name__)
+
+
+# ── Connection config (from env, with sensible defaults to netcup IPs) ──
+# These defaults point to the actual IPs on netcup rmi_network.
+# Override via env vars in docker-compose.yml for portability.
+DEFAULT_CONFIG: dict[str, Any] = {
+    "redis": {
+        "host": os.getenv("REDIS_HOST", "rmi-redis"),
+        "port": int(os.getenv("REDIS_PORT", "6379")),
+        "password": os.getenv("REDIS_PASSWORD", "RMI_PROD_REDIS_2026"),
+        "db": int(os.getenv("REDIS_DB", "0")),
+    },
+    "postgres": {
+        "host": os.getenv("POSTGRES_HOST", "rmi-postgres"),
+        "port": int(os.getenv("POSTGRES_PORT", "5432")),
+        "user": os.getenv("POSTGRES_USER", "rmi"),
+        "password": os.getenv("POSTGRES_PASSWORD", ""),
+        "database": os.getenv("POSTGRES_DB", "rmi"),
+    },
+    "neo4j": {
+        "uri": os.getenv("NEO4J_URI", "bolt://rmi-neo4j:7687"),
+        "user": os.getenv("NEO4J_USER", "neo4j"),
+        "password": os.getenv("NEO4J_PASSWORD", ""),
+    },
+    "qdrant": {
+        "url": os.getenv("QDRANT_URL", "http://rmi-qdrant:6333"),
+        "api_key": os.getenv("QDRANT_API_KEY", ""),
+    },
+    "minio": {
+        "endpoint": os.getenv("MINIO_ENDPOINT", "rmi-minio:9000"),
+        "access_key": os.getenv("MINIO_ACCESS_KEY", ""),
+        "secret_key": os.getenv("MINIO_SECRET_KEY", ""),
+    },
+}
+
+
+class StoreHealth:
+    """Tracks which stores are reachable. Updated on each probe."""
+
+    def __init__(self) -> None:
+        self.redis: bool = False
+        self.postgres: bool = False
+        self.neo4j: bool = False
+        self.qdrant: bool = False
+        self.minio: bool = False
+        self.last_checked: float = 0.0
+
+
+class CatalogService:
+    """Unified read/write API for RMI data.
+
+    Use `get_catalog()` to get the singleton instance. All methods are
+    async and graceful-degrade — if a store is unreachable, the method
+    returns None or an empty result with a logged warning.
+    """
+
+    def __init__(self, config: dict[str, Any] | None = None) -> None:
+        self.config = config or DEFAULT_CONFIG
+        self._health = StoreHealth()
+        self._redis = None
+        self._pg_pool: Any = None
+        self._neo_driver: Any = None
+        self._qdrant: Any = None  # httpx.AsyncClient
+        self._init_lock = asyncio.Lock()
+
+    # ── Store init (lazy) ────────────────────────────────────────────
+    async def _init_stores(self) -> None:
+        async with self._init_lock:
+            if self._redis is not None:
+                return  # already init
+            # Redis (always try first — used for everything)
+            try:
+                import redis.asyncio as aioredis
+
+                cfg = self.config["redis"]
+                self._redis = aioredis.Redis(
+                    host=cfg["host"],
+                    port=cfg["port"],
+                    password=cfg["password"],
+                    db=cfg["db"],
+                    decode_responses=True,
+                )
+                await self._redis.ping()
+                self._health.redis = True
+                log.info("catalog_redis_ok")
+            except Exception as e:
+                log.warning("catalog_redis_fail: %s", e)
+                self._health.redis = False
+
+            # Postgres
+            try:
+                import asyncpg
+
+                cfg = self.config["postgres"]
+                if cfg["password"]:
+                    self._pg_pool = await asyncpg.create_pool(
+                        host=cfg["host"],
+                        port=cfg["port"],
+                        user=cfg["user"],
+                        password=cfg["password"],
+                        database=cfg["database"],
+                        min_size=1,
+                        max_size=8,
+                        command_timeout=10,
+                    )
+                    self._health.postgres = True
+                    log.info("catalog_postgres_ok")
+            except Exception as e:
+                log.warning("catalog_postgres_fail: %s", e)
+                self._health.postgres = False
+
+            # Neo4j
+            try:
+                from neo4j import GraphDatabase
+
+                cfg = self.config["neo4j"]
+                if cfg["password"]:
+                    self._neo_driver = GraphDatabase.driver(
+                        cfg["uri"], auth=(cfg["user"], cfg["password"])
+                    )
+                    with self._neo_driver.session() as s:
+                        s.run("RETURN 1").consume()
+                    self._health.neo4j = True
+                    log.info("catalog_neo4j_ok")
+            except Exception as e:
+                log.warning("catalog_neo4j_fail: %s", e)
+                self._health.neo4j = False
+
+            # Qdrant (HTTP)
+            try:
+                cfg = self.config["qdrant"]
+                headers = {"api-key": cfg["api_key"]} if cfg["api_key"] else {}
+                self._qdrant = httpx.AsyncClient(
+                    base_url=cfg["url"], headers=headers, timeout=5.0
+                )
+                r = await self._qdrant.get("/collections")
+                if r.status_code == 200:
+                    self._health.qdrant = True
+                    log.info("catalog_qdrant_ok")
+                else:
+                    log.warning("catalog_qdrant_http_%d", r.status_code)
+            except Exception as e:
+                log.warning("catalog_qdrant_fail: %s", e)
+                self._health.qdrant = False
+
+            self._health.last_checked = time.time()
+
+    # ── Health probe ────────────────────────────────────────────────
+    async def probe_stores(self) -> dict[str, bool]:
+        await self._init_stores()
+        return {
+            "redis": self._health.redis,
+            "postgres": self._health.postgres,
+            "neo4j": self._health.neo4j,
+            "qdrant": self._health.qdrant,
+            "minio": self._health.minio,
+        }
+
+    # ── Tokens (Postgres primary + Redis cache) ─────────────────────
+    async def get_token(self, chain: Chain, address: str) -> Token | None:
+        await self._init_stores()
+        cache_key = f"catalog:token:{chain.value}:{address}"
+        if self._health.redis:
+            try:
+                cached = await self._redis.get(cache_key)
+                if cached:
+                    return Token.model_validate_json(cached)
+            except Exception as e:
+                log.debug("token_cache_get_fail: %s", e)
+        if not self._health.postgres:
+            return None
+        try:
+            async with self._pg_pool.acquire() as conn:
+                row = await conn.fetchrow(
+                    "SELECT * FROM tokens WHERE chain=$1 AND address=$2",
+                    chain.value, address,
+                )
+            if not row:
+                return None
+            token = Token(**dict(row))
+            if self._health.redis:
+                try:
+                    await self._redis.setex(cache_key, 3600, token.model_dump_json())
+                except Exception as e:
+                    log.debug("token_cache_set_fail: %s", e)
+            return token
+        except Exception as e:
+            log.warning("token_get_fail: %s", e)
+            return None
+
+    async def save_token(self, token: Token) -> bool:
+        await self._init_stores()
+        if not self._health.postgres:
+            log.warning("token_save_no_postgres")
+            return False
+        try:
+            async with self._pg_pool.acquire() as conn:
+                await conn.execute(
+                    """
+                    INSERT INTO tokens (
+                        token_id, chain, address, symbol, name, decimals,
+                        deployer_wallet_id, deployed_at, initial_supply,
+                        current_supply, is_honeypot, is_mintable, is_proxy,
+                        tax_buy_bps, tax_sell_bps, risk_tier, risk_score,
+                        risk_factors, rag_embedding_id
+                    ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,
+                              $14,$15,$16,$17,$18,$19)
+                    ON CONFLICT (token_id) DO UPDATE SET
+                        symbol=EXCLUDED.symbol,
+                        name=EXCLUDED.name,
+                        current_supply=EXCLUDED.current_supply,
+                        is_honeypot=EXCLUDED.is_honeypot,
+                        is_mintable=EXCLUDED.is_mintable,
+                        risk_tier=EXCLUDED.risk_tier,
+                        risk_score=EXCLUDED.risk_score,
+                        risk_factors=EXCLUDED.risk_factors
+                    """,
+                    token.token_id, token.chain.value, token.address,
+                    token.symbol, token.name, token.decimals,
+                    token.deployer_wallet_id, token.deployed_at,
+                    token.initial_supply, token.current_supply,
+                    token.is_honeypot, token.is_mintable, token.is_proxy,
+                    token.tax_buy_bps, token.tax_sell_bps,
+                    token.risk_tier.value if token.risk_tier else None,
+                    token.risk_score, token.risk_factors, token.rag_embedding_id,
+                )
+            # Invalidate cache
+            if self._health.redis:
+                try:
+                    await self._redis.delete(
+                        f"catalog:token:{token.chain.value}:{token.address}"
+                    )
+                except Exception:
+                    pass
+            return True
+        except Exception as e:
+            log.warning("token_save_fail: %s", e)
+            return False
+
+    # ── Wallets (Neo4j primary) ──────────────────────────────────────
+    async def get_wallet(self, chain: Chain, address: str) -> Wallet | None:
+        await self._init_stores()
+        if not self._health.neo4j:
+            return None
+        wallet_id = f"{chain.value}:{address}"
+        try:
+            with self._neo_driver.session() as s:
+                result = s.run(
+                    "MATCH (w:Wallet {wallet_id: $id}) RETURN w",
+                    id=wallet_id,
+                ).single()
+            if not result:
+                return None
+            node = dict(result["w"])
+            # Normalize datetime strings
+            for k in ("first_seen", "last_seen"):
+                if isinstance(node.get(k), str):
+                    from datetime import datetime
+                    try:
+                        node[k] = datetime.fromisoformat(node[k].replace("Z", "+00:00"))
+                    except Exception:
+                        pass
+            return Wallet(**node)
+        except Exception as e:
+            log.warning("wallet_get_fail: %s", e)
+            return None
+
+    async def save_wallet(self, wallet: Wallet) -> bool:
+        await self._init_stores()
+        if not self._health.neo4j:
+            return False
+        try:
+            with self._neo_driver.session() as s:
+                s.run(
+                    """
+                    MERGE (w:Wallet {wallet_id: $wallet_id})
+                    SET w.chain=$chain, w.address=$address,
+                        w.first_seen=$first_seen, w.last_seen=$last_seen,
+                        w.tx_count=$tx_count, w.total_volume_usd=$total_volume_usd,
+                        w.is_deployer=$is_deployer,
+                        w.is_known_exchange=$is_known_exchange,
+                        w.is_suspicious=$is_suspicious,
+                        w.reputation_score=$reputation_score
+                    """,
+                    wallet_id=wallet.wallet_id,
+                    chain=wallet.chain.value,
+                    address=wallet.address,
+                    first_seen=wallet.first_seen.isoformat(),
+                    last_seen=wallet.last_seen.isoformat(),
+                    tx_count=wallet.tx_count,
+                    total_volume_usd=wallet.total_volume_usd,
+                    is_deployer=wallet.is_deployer,
+                    is_known_exchange=wallet.is_known_exchange,
+                    is_suspicious=wallet.is_suspicious,
+                    reputation_score=wallet.reputation_score,
+                )
+            return True
+        except Exception as e:
+            log.warning("wallet_save_fail: %s", e)
+            return False
+
+    # ── Recipe 1: find tokens by deployer history ───────────────────
+    async def find_tokens_by_deployer_history(
+        self, min_rug_count: int = 1, chain: Chain | None = None, limit: int = 50
+    ) -> list[Token]:
+        """Find tokens deployed by wallets that have rug-pulled before.
+
+        Cross-store: Neo4j for the wallet filter, Postgres for the tokens.
+        """
+        await self._init_stores()
+        if not (self._health.neo4j and self._health.postgres):
+            log.warning("recipe1_stores_unavailable")
+            return []
+        try:
+            # Step 1: Neo4j — find wallets with rug_count >= min_rug_count
+            with self._neo_driver.session() as s:
+                wallets = [
+                    r["w.wallet_id"]
+                    for r in s.run(
+                        "MATCH (w:Deployer) WHERE w.rug_count >= $min "
+                        "RETURN w.wallet_id LIMIT 200",
+                        min=min_rug_count,
+                    )
+                ]
+            if not wallets:
+                return []
+            # Step 2: Postgres — find tokens deployed by those wallets
+            async with self._pg_pool.acquire() as conn:
+                query = (
+                    "SELECT * FROM tokens WHERE deployer_wallet_id = ANY($1::text[]) "
+                )
+                params: list[Any] = [wallets]
+                if chain:
+                    query += "AND chain=$2 "
+                    params.append(chain.value)
+                query += "ORDER BY deployed_at DESC LIMIT $3"
+                params.append(limit)
+                rows = await conn.fetch(query, *params)
+            return [Token(**dict(r)) for r in rows]
+        except Exception as e:
+            log.warning("recipe1_fail: %s", e)
+            return []
+
+    # ── Recipe 3: get_token_risk (3-store composition) ──────────────
+    async def get_token_risk(
+        self, chain: Chain, address: str
+    ) -> dict[str, Any]:
+        """Real-time risk score — composes Redis cache + Postgres + Neo4j.
+
+        Returns dict (not Token) because it's a cross-store projection.
+        Cache TTL 60s.
+        """
+        await self._init_stores()
+        cache_key = f"catalog:risk:{chain.value}:{address}"
+        if self._health.redis:
+            try:
+                cached = await self._redis.get(cache_key)
+                if cached:
+                    return json.loads(cached)
+            except Exception:
+                pass
+
+        token = await self.get_token(chain, address)
+        deployer_reputation: int | None = None
+        if token and token.deployer_wallet_id and self._health.neo4j:
+            try:
+                with self._neo_driver.session() as s:
+                    r = s.run(
+                        "MATCH (d:Deployer {wallet_id: $id}) "
+                        "RETURN d.reputation_score AS rep, d.rug_count AS rugs",
+                        id=token.deployer_wallet_id,
+                    ).single()
+                if r:
+                    deployer_reputation = r["rep"]
+            except Exception as e:
+                log.debug("risk_neo4j_fail: %s", e)
+
+        token_score = token.risk_score if token and token.risk_score is not None else 50
+        if deployer_reputation is not None:
+            score = int(0.6 * token_score + 0.4 * (100 - deployer_reputation))
+        else:
+            score = token_score
+        score = max(0, min(100, score))
+        if score < 25:
+            tier = "low"
+        elif score < 50:
+            tier = "medium"
+        elif score < 75:
+            tier = "high"
+        else:
+            tier = "critical"
+        result: dict[str, Any] = {
+            "score": score,
+            "tier": tier,
+            "factors": token.risk_factors if token else [],
+            "token": token.model_dump() if token else None,
+            "deployer_reputation": deployer_reputation,
+            "fetched_at": utcnow().isoformat(),
+        }
+        if self._health.redis:
+            try:
+                await self._redis.setex(cache_key, 60, json.dumps(result, default=str))
+            except Exception:
+                pass
+        return result
+
+    # ── Recipe 5: resolve_entity (Neo4j Cypher) ─────────────────────
+    async def resolve_entity(
+        self, wallet_id: str, max_chains: int = 5
+    ) -> dict[str, Any]:
+        """Cross-chain entity resolution via Neo4j.
+
+        Uses the SAME_AS / FUNDED_BY_SAME / CLONE_OF / BEHAVIORAL_MATCH
+        edges defined in v4.0 §T32 with path-weighted confidence.
+        """
+        await self._init_stores()
+        if not self._health.neo4j:
+            return {"entity_id": None, "wallets": [], "note": "neo4j unavailable"}
+        try:
+            with self._neo_driver.session() as s:
+                result = s.run(
+                    """
+                    MATCH (start:Wallet {wallet_id: $wallet_id})
+                    OPTIONAL MATCH path = (start)-[:SAME_AS|FUNDED_BY_SAME|CLONE_OF|BEHAVIORAL_MATCH*1..3]-(other:Wallet)
+                    WHERE start <> other
+                    WITH start, other, path,
+                         reduce(conf = 1.0, r IN relationships(path) | conf * r.confidence) AS path_conf
+                    ORDER BY path_conf DESC
+                    LIMIT $limit
+                    RETURN start.entity_id AS entity_id,
+                           collect({
+                               wallet_id: other.wallet_id,
+                               chain: other.chain,
+                               address: other.address,
+                               confidence: path_conf
+                           }) AS cross_chain_wallets
+                    """,
+                    wallet_id=wallet_id, limit=max_chains,
+                ).single()
+            if not result:
+                return {"entity_id": None, "wallets": [], "note": "no edges"}
+            wallets = [w for w in (result["cross_chain_wallets"] or []) if w.get("wallet_id")]
+            return {
+                "entity_id": result["entity_id"],
+                "wallets": wallets,
+                "note": f"{len(wallets)} cross-chain matches",
+            }
+        except Exception as e:
+            log.warning("resolve_entity_fail: %s", e)
+            return {"entity_id": None, "wallets": [], "error": str(e)}
+
+    # ── RAG bridge: wire app/rag/engine.py into the catalog ─────────
+    async def rag_search(
+        self, query: str, collection: str = "scam_intel", top_k: int = 5
+    ) -> list[dict]:
+        """Search the RAG system. Returns raw hits (catalog-agnostic).
+
+        Use this when you want RAG results as raw search hits.
+        Use get_token_risk / etc. when you want catalog-typed results.
+        """
+        try:
+            return await rag_three_pillar_search(
+                query=query, collection=collection, top_k=top_k
+            )
+        except Exception as e:
+            log.warning("rag_search_fail: %s", e)
+            return []
+
+    async def rag_ingest(
+        self,
+        content: str,
+        collection: str = "scam_intel",
+        doc_id: str | None = None,
+        metadata: dict | None = None,
+    ) -> dict:
+        """Ingest a fact into RAG and (if it's about a token) link to Token.rag_embedding_id.
+
+        Returns the RAG ingest result plus the Qdrant point_id for cross-store ref.
+        """
+        from uuid import uuid4
+
+        doc_id = doc_id or f"finding:{uuid4().hex[:12]}"
+        try:
+            r = await rag_ingest_document(
+                collection=collection,
+                doc_id=doc_id,
+                content=content,
+                metadata=metadata or {},
+            )
+            r["qdrant_point_id"] = doc_id  # Engine returns the same
+            return r
+        except Exception as e:
+            log.warning("rag_ingest_fail: %s", e)
+            return {"status": "failed", "error": str(e)}
+
+    async def attach_rag_to_token(
+        self, chain: Chain, address: str, qdrant_point_id: str
+    ) -> bool:
+        """Link an existing Qdrant point to a Token row as rag_embedding_id."""
+        token = await self.get_token(chain, address)
+        if not token:
+            return False
+        token.rag_embedding_id = qdrant_point_id
+        return await self.save_token(token)
+
+    # ── Stats / introspection ────────────────────────────────────────
+    async def stats(self) -> dict[str, Any]:
+        await self._init_stores()
+        out: dict[str, Any] = {
+            "stores": await self.probe_stores(),
+            "collections": COLLECTIONS,
+        }
+        if self._health.postgres:
+            try:
+                async with self._pg_pool.acquire() as conn:
+                    out["tokens"] = await conn.fetchval("SELECT COUNT(*) FROM tokens") or 0
+                    out["alerts"] = await conn.fetchval("SELECT COUNT(*) FROM alerts") or 0
+            except Exception as e:
+                out["postgres_error"] = str(e)
+        if self._health.neo4j:
+            try:
+                with self._neo_driver.session() as s:
+                    out["wallets"] = s.run("MATCH (w:Wallet) RETURN COUNT(w) AS c").single()["c"] or 0
+                    out["deployers"] = s.run("MATCH (d:Deployer) RETURN COUNT(d) AS c").single()["c"] or 0
+            except Exception as e:
+                out["neo4j_error"] = str(e)
+        if self._health.qdrant:
+            try:
+                cols = await self._qdrant.get("/collections")
+                if cols.status_code == 200:
+                    out["qdrant_collections"] = [
+                        c["name"] for c in cols.json().get("result", {}).get("collections", [])
+                    ]
+            except Exception as e:
+                out["qdrant_error"] = str(e)
+        return out
+
+
+# ── Singleton accessor ──────────────────────────────────────────────
+_catalog: CatalogService | None = None
+
+
+def get_catalog() -> CatalogService:
+    """Get the global CatalogService. Lazy-init on first call."""
+    global _catalog
+    if _catalog is None:
+        _catalog = CatalogService()
+    return _catalog

backend/main.py

@@ -215,6 +215,7 @@ def _try_mount_v1_routers() -> int:
         "app.api.v1.rag.search",
         "app.api.v1.x402.payments",
         "app.api.v1.admin.alerts_webhook",
+        "app.api.v1.catalog",
     ]
 
     for module_path in v1_modules: