feat(rag/m3): v3 RAG engine — embeddings + ANN + chunking + 3-pillar search

backend/app/api/v1/rag/search.py

@@ -5,9 +5,10 @@ facade exposes the most-used operations: search, ingest, feedback.
 """
 from __future__ import annotations
 
-from typing import Annotated
+from typing import Annotated, Any
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
 
 from app.rag import (
     FeedbackRecord,
@@ -17,6 +18,8 @@ from app.rag import (
     SearchRequest,
     SearchResponse,
 )
+from app.rag.engine import bulk_ingest as engine_bulk_ingest
+from app.rag.engine import get_stats as engine_get_stats
 
 router = APIRouter(prefix="/api/v1/rag/v2", tags=["rag"])
 
@@ -25,6 +28,11 @@ def _service() -> RAGService:
     return RAGService()
 
 
+class BulkIngestRequest(BaseModel):
+    collection: str = "scam_intel"
+    items: list[dict[str, Any]] = Field(default_factory=list)
+
+
 @router.post("/search", response_model=SearchResponse)
 async def search(
     req: SearchRequest,
@@ -55,3 +63,19 @@ async def feedback(
         collection="known_scams",
         status="ok" if ok else "failed",
     )
+
+
+@router.get("/stats")
+async def stats() -> dict:
+    """Per-collection vector counts + active embedder backend."""
+    return engine_get_stats()
+
+
+@router.post("/bulk-ingest")
+async def bulk(req: BulkIngestRequest) -> dict:
+    """Ingest many items into a collection sequentially (max 500 per call)."""
+    if not req.items:
+        raise HTTPException(status_code=400, detail="items must be non-empty")
+    if len(req.items) > 500:
+        raise HTTPException(status_code=400, detail="bulk limit 500 per call")
+    return await engine_bulk_ingest(items=req.items, collection=req.collection)

backend/app/rag/__init__.py

@@ -1,4 +1,12 @@
-"""RAG domain — auto-registers its health check."""
+"""RAG domain — v3 M3 engine surface.
+
+Public API (per v3 unfuck guide §M3):
+- RAGService, init_rag
+- SearchRequest, SearchResponse, SearchHit
+- IngestRequest, IngestResult
+- FeedbackRecord
+- EmbeddingProvider, COLLECTIONS
+"""
 from __future__ import annotations
 
 from app.core import health as health_mod
@@ -6,13 +14,18 @@ from app.core.health import DomainHealth
 
 
 async def _health_check() -> DomainHealth:
-    """RAG health: legacy rag_engine + vector store available."""
+    """RAG health: v3 engine present + Redis reachable."""
     try:
-        from app.rag_engine import search_similar
+        from app.rag.engine import get_stats
+        stats = get_stats()
         return DomainHealth(
             name="rag",
             healthy=True,
-            details={"engine": "legacy", "module": "app.rag_engine"},
+            details={
+                "engine": "v3-m3",
+                "backend": stats.get("backend", "unknown"),
+                "total_docs": stats.get("total_docs", 0),
+            },
         )
     except Exception as e:
         return DomainHealth(name="rag", healthy=False, error=str(e))
@@ -22,7 +35,7 @@ health_mod.register_health_check("rag", _health_check)
 
 
 # Public API
-from app.rag.models import (  # noqa: F401
+from app.rag.models import (  # noqa: E402
     COLLECTIONS,
     EmbeddingProvider,
     FeedbackRecord,
@@ -32,7 +45,7 @@ from app.rag.models import (  # noqa: F401
     SearchRequest,
     SearchResponse,
 )
-from app.rag.service import RAGService, init_rag  # noqa: F401
+from app.rag.service import RAGService, init_rag  # noqa: E402
 
 __all__ = [
     "RAGService",

backend/app/rag/ann_index.py

@@ -0,0 +1,240 @@
+"""M3 RAG ANN Index — numpy-based cosine similarity with Redis persistence.
+
+Why numpy + Redis (not FAISS):
+- 13 collections × ~5K docs = ~65K total — small enough for in-process numpy
+- No FAISS native dep, no rebuild on container restart
+- Vector + metadata stored as JSON in Redis; loaded into a numpy matrix
+  on first search, then kept in process memory for fast repeated queries
+- Cosine sim = dot product on L2-normalized vectors (we normalize on insert)
+
+Persistence shape (per collection):
+  rag:ann:{collection}:meta   → JSON {count, dim}
+  rag:ann:{collection}:docs   → HASH doc_id → {"vector": [...], "metadata": {...}, "text": "..."}
+"""
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+import numpy as np
+
+from app.rag.embeddings import EMBEDDING_DIM
+
+log = logging.getLogger(__name__)
+
+# Lazy import — Redis client is created on first use, after app.core.redis is ready
+_redis_client = None
+
+
+def _get_redis():
+    global _redis_client
+    if _redis_client is None:
+        from app.core.redis import get_redis
+
+        _redis_client = get_redis()
+    return _redis_client
+
+
+@dataclass
+class Hit:
+    """One search result."""
+
+    doc_id: str
+    score: float
+    text: str
+    metadata: dict[str, Any]
+
+
+class ANNIndex:
+    """Per-collection vector index. Holds vectors in memory, persists to Redis.
+
+    Usage:
+        idx = ANNIndex("scam_intel")
+        idx.add("doc1", vector, {"source": "x"}, text="...")
+        hits = idx.search(query_vec, top_k=5)
+    """
+
+    def __init__(self, collection: str, dim: int = EMBEDDING_DIM):
+        self.collection = collection
+        self.dim = dim
+        self._ids: list[str] = []
+        self._matrix: np.ndarray | None = None  # shape (N, dim), L2-normalized
+        self._meta: dict[str, dict] = {}  # doc_id → {text, metadata}
+        self._loaded = False
+
+    # ── Redis keys ────────────────────────────────────────────────────
+    def _key_docs(self) -> str:
+        return f"rag:ann:{self.collection}:docs"
+
+    def _key_meta(self) -> str:
+        return f"rag:ann:{self.collection}:meta"
+
+    # ── Persistence ───────────────────────────────────────────────────
+    def _persist(self) -> None:
+        r = _get_redis()
+        pipe = r.pipeline()
+        for doc_id in self._ids:
+            entry = {
+                "vector": (self._matrix[self._ids.index(doc_id)] if self._matrix is not None else []).tolist(),
+                "metadata": self._meta[doc_id].get("metadata", {}),
+                "text": self._meta[doc_id].get("text", ""),
+            }
+            pipe.hset(self._key_docs(), doc_id, json.dumps(entry))
+        pipe.set(
+            self._key_meta(),
+            json.dumps({"count": len(self._ids), "dim": self.dim}),
+        )
+        pipe.execute()
+
+    def _load(self) -> None:
+        if self._loaded:
+            return
+        try:
+            r = _get_redis()
+            raw = r.hgetall(self._key_docs())
+        except Exception as e:
+            log.warning("ann_load_failed collection=%s err=%s", self.collection, e)
+            raw = {}
+
+        if not raw:
+            self._loaded = True
+            return
+
+        ids: list[str] = []
+        vecs: list[list[float]] = []
+        meta: dict[str, dict] = {}
+        for doc_id, blob in raw.items():
+            try:
+                entry = json.loads(blob)
+                ids.append(doc_id)
+                vecs.append(entry.get("vector", []))
+                meta[doc_id] = {
+                    "metadata": entry.get("metadata", {}),
+                    "text": entry.get("text", ""),
+                }
+            except Exception as e:
+                log.debug("ann_skip doc=%s err=%s", doc_id, e)
+        if vecs:
+            self._matrix = np.asarray(vecs, dtype=np.float32)
+        self._ids = ids
+        self._meta = meta
+        self._loaded = True
+        log.info("ann_loaded collection=%s count=%d", self.collection, len(ids))
+
+    # ── Mutations ─────────────────────────────────────────────────────
+    def add(
+        self,
+        doc_id: str,
+        vector: list[float],
+        metadata: dict[str, Any] | None = None,
+        text: str = "",
+    ) -> None:
+        self._load()
+        arr = np.asarray(vector, dtype=np.float32)
+        if arr.shape[0] != self.dim:
+            # Pad or truncate
+            if arr.shape[0] >= self.dim:
+                arr = arr[: self.dim]
+            else:
+                arr = np.concatenate(
+                    [arr, np.zeros(self.dim - arr.shape[0], dtype=np.float32)]
+                )
+        # L2 normalize
+        n = float(np.linalg.norm(arr))
+        if n > 0:
+            arr = arr / n
+
+        if doc_id in self._ids:
+            # Update — replace vector
+            idx = self._ids.index(doc_id)
+            self._matrix[idx] = arr
+        else:
+            if self._matrix is None:
+                self._matrix = arr.reshape(1, -1)
+            else:
+                self._matrix = np.vstack([self._matrix, arr.reshape(1, -1)])
+            self._ids.append(doc_id)
+
+        self._meta[doc_id] = {"metadata": metadata or {}, "text": text}
+        self._persist()
+
+    def delete(self, doc_id: str) -> bool:
+        self._load()
+        if doc_id not in self._ids:
+            return False
+        idx = self._ids.index(doc_id)
+        self._ids.pop(idx)
+        if self._matrix is not None and self._matrix.shape[0] > 1:
+            self._matrix = np.delete(self._matrix, idx, axis=0)
+        else:
+            self._matrix = None
+        self._meta.pop(doc_id, None)
+        self._persist()
+        return True
+
+    def clear(self) -> None:
+        self._ids = []
+        self._matrix = None
+        self._meta = {}
+        self._loaded = True
+        try:
+            r = _get_redis()
+            r.delete(self._key_docs(), self._key_meta())
+        except Exception as e:
+            log.warning("ann_clear_failed: %s", e)
+
+    # ── Search ────────────────────────────────────────────────────────
+    def search(
+        self,
+        query_vector: list[float],
+        top_k: int = 5,
+        min_similarity: float = 0.0,
+    ) -> list[Hit]:
+        self._load()
+        if self._matrix is None or len(self._ids) == 0:
+            return []
+        q = np.asarray(query_vector, dtype=np.float32)
+        n = float(np.linalg.norm(q))
+        if n > 0:
+            q = q / n
+        # Cosine sim = dot product on normalized vectors
+        scores = self._matrix @ q
+        # Top-k by score
+        k = min(top_k, len(self._ids))
+        # argpartition is faster than full argsort for small top_k
+        idx = np.argpartition(-scores, k - 1)[:k] if k < len(scores) else np.arange(len(scores))
+        idx = idx[np.argsort(-scores[idx])]
+        hits: list[Hit] = []
+        for i in idx:
+            s = float(scores[i])
+            if s < min_similarity:
+                continue
+            doc_id = self._ids[int(i)]
+            meta = self._meta.get(doc_id, {})
+            hits.append(
+                Hit(
+                    doc_id=doc_id,
+                    score=s,
+                    text=meta.get("text", ""),
+                    metadata=meta.get("metadata", {}),
+                )
+            )
+        return hits
+
+    # ── Stats ─────────────────────────────────────────────────────────
+    def count(self) -> int:
+        self._load()
+        return len(self._ids)
+
+
+# ── Per-collection cache ─────────────────────────────────────────────
+_index_cache: dict[str, ANNIndex] = {}
+
+
+def get_index(collection: str) -> ANNIndex:
+    """Get or create the ANNIndex for a collection. Cached per process."""
+    if collection not in _index_cache:
+        _index_cache[collection] = ANNIndex(collection)
+    return _index_cache[collection]

backend/app/rag/chunking.py

@@ -0,0 +1,158 @@
+"""M3 RAG Chunking — recursive text chunking with MD5 dedup.
+
+Why chunking matters:
+- Embedding models have context limits (bge-m3 = 8192 tokens, but we
+  chunk to 512 for granularity and to keep individual FAISS hits useful)
+- Smaller chunks = better retrieval precision (less noise per result)
+- Dedup via content MD5 prevents the same fact being ingested N times
+
+Strategy (per 2026 RAG standards):
+- Recursive character split, paragraph → sentence → word boundaries
+- Overlap window preserves cross-chunk context
+- Quality score: (length / max_chunk) × (1 - special_char_ratio)
+- Skip chunks < min_chars (likely noise)
+"""
+from __future__ import annotations
+
+import hashlib
+import logging
+import re
+from dataclasses import dataclass
+
+log = logging.getLogger(__name__)
+
+DEFAULT_MAX_CHUNK = 512
+DEFAULT_OVERLAP = 64
+DEFAULT_MIN_CHARS = 32
+
+_redis = None
+
+
+def _get_redis():
+    global _redis
+    if _redis is None:
+        from app.core.redis import get_redis
+        _redis = get_redis()
+    return _redis
+
+
+@dataclass
+class Chunk:
+    """A piece of text ready to embed."""
+
+    text: str
+    content_hash: str
+    index: int
+    quality: float = 1.0
+
+
+def content_hash(text: str) -> str:
+    """Stable MD5 of normalized text. Used for dedup."""
+    norm = re.sub(r"\s+", " ", text.strip().lower())
+    return hashlib.md5(norm.encode("utf-8", errors="ignore")).hexdigest()
+
+
+def chunk_text(
+    text: str,
+    max_chunk: int = DEFAULT_MAX_CHUNK,
+    overlap: int = DEFAULT_OVERLAP,
+    min_chars: int = DEFAULT_MIN_CHARS,
+) -> list[Chunk]:
+    """Recursive character chunker.
+
+    Splits on paragraph breaks first, then sentence, then word. Each split
+    is greedy-packed into chunks of <= max_chunk chars, with `overlap` chars
+    carried forward to preserve context.
+    """
+    if not text or not text.strip():
+        return []
+
+    text = text.strip()
+    if len(text) <= max_chunk:
+        h = content_hash(text)
+        return [Chunk(text=text, content_hash=h, index=0, quality=_quality(text, max_chunk))]
+
+    # Build sentence-level segments (split on . ! ? \n)
+    sentences = re.split(r"(?<=[.!?\n])\s+", text)
+    chunks: list[str] = []
+    cur = ""
+    for s in sentences:
+        s = s.strip()
+        if not s:
+            continue
+        # If a single sentence is too long, hard-split on words
+        if len(s) > max_chunk:
+            words = s.split()
+            sub = ""
+            for w in words:
+                if len(sub) + len(w) + 1 > max_chunk and sub:
+                    chunks.append(sub)
+                    sub = w
+                else:
+                    sub = (sub + " " + w).strip()
+            if sub:
+                chunks.append(sub)
+        elif len(cur) + len(s) + 1 > max_chunk:
+            if cur:
+                chunks.append(cur)
+            cur = s
+        else:
+            cur = (cur + " " + s).strip()
+
+    if cur:
+        chunks.append(cur)
+
+    # Apply overlap: each chunk gets the last `overlap` chars of the prior chunk as prefix
+    if overlap > 0 and len(chunks) > 1:
+        overlapped: list[str] = []
+        for i, c in enumerate(chunks):
+            if i == 0:
+                overlapped.append(c)
+            else:
+                tail = chunks[i - 1][-overlap:]
+                overlapped.append(tail + " " + c)
+        chunks = overlapped
+
+    # Filter and produce Chunk objects
+    out: list[Chunk] = []
+    for i, c in enumerate(chunks):
+        if len(c) < min_chars:
+            continue
+        out.append(
+            Chunk(
+                text=c,
+                content_hash=content_hash(c),
+                index=i,
+                quality=_quality(c, max_chunk),
+            )
+        )
+    return out
+
+
+def _quality(text: str, max_chunk: int) -> float:
+    """Quick quality score 0-1. Penalize noise (high special-char ratio)."""
+    if not text:
+        return 0.0
+    n = len(text)
+    special = sum(1 for c in text if not c.isalnum() and not c.isspace())
+    length_score = min(1.0, n / max_chunk)
+    special_score = max(0.0, 1.0 - (special / n))
+    return round(0.7 * length_score + 0.3 * special_score, 3)
+
+
+# ── Dedup via Redis ─────────────────────────────────────────────────
+def is_duplicate(content_hash: str, collection: str) -> bool:
+    """Has this exact content already been ingested into the collection?"""
+    try:
+        return bool(_get_redis().sismember(f"rag:hashes:{collection}", content_hash))
+    except Exception as e:
+        log.debug("dedup_check_failed: %s", e)
+        return False
+
+
+def mark_ingested(content_hash: str, collection: str) -> None:
+    """Record that this content hash is now in the collection."""
+    try:
+        _get_redis().sadd(f"rag:hashes:{collection}", content_hash)
+    except Exception as e:
+        log.debug("dedup_mark_failed: %s", e)

backend/app/rag/embeddings.py

@@ -0,0 +1,183 @@
+"""M3 RAG Embeddings — multi-tier fallback chain.
+
+Tier 1: Ollama bge-m3 (1024d, our own, zero cost) — best quality when reachable
+Tier 2: OpenRouter NVIDIA Nemotron (2048d, free tier) — if API key set
+Tier 3: Hash-based (1024d, deterministic) — always works, no network
+
+Design (per DESIGN.md M3):
+- Single async API: `await get_embedding(text) -> list[float]`
+- BGE-M3 preferred when available; hash is acceptable degradation
+- Embedding dim is reported via EMBEDDING_DIM; collections use same dim
+- All embeddings are float lists (JSON-serializable) for Redis storage
+"""
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+from typing import Final
+
+import httpx
+import numpy as np
+
+log = logging.getLogger(__name__)
+
+# Default dim matches bge-m3 (1024d). Hash is also 1024d for cross-backend
+# compatibility. OpenRouter Nemotron is 2048d but we truncate/pad to 1024
+# during a collection's lifecycle to avoid dim mixing.
+EMBEDDING_DIM: Final = 1024
+
+# Backends
+_BACKEND = os.getenv("RAG_EMBED_BACKEND", "auto").lower()
+OLLAMA_URL = os.getenv("OLLAMA_URL", "http://host.docker.internal:11434").rstrip("/")
+OLLAMA_MODEL = os.getenv("OLLAMA_EMBED_MODEL", "bge-m3")
+OPENROUTER_URL = "https://openrouter.ai/api/v1/embeddings"
+OPENROUTER_MODEL = "nvidia/llama-nemotron-embed-vl-1b-v2:free"
+
+
+def _ollama_reachable() -> bool:
+    """Quick probe: is Ollama listening? Cached per process via env."""
+    if os.getenv("_OLLAMA_PROBED"):
+        return os.getenv("_OLLAMA_OK") == "1"
+    try:
+        with httpx.Client(timeout=2.0) as c:
+            r = c.get(f"{OLLAMA_URL}/api/tags")
+            ok = r.status_code == 200
+    except Exception:
+        ok = False
+    os.environ["_OLLAMA_PROBED"] = "1"
+    os.environ["_OLLAMA_OK"] = "1" if ok else "0"
+    if not ok:
+        log.info("rag_ollama_unreachable url=%s", OLLAMA_URL)
+    return ok
+
+
+def _hash_embed(text: str, dim: int = EMBEDDING_DIM) -> list[float]:
+    """Deterministic hash-based 1024d embedding.
+
+    NOT semantic — purely positional. Used as last-resort fallback so the
+    RAG system stays functional when all neural backends are unreachable.
+    Quality is poor for similarity (cosine sim typically 0.1-0.3) but the
+    API contract holds: same text → same vector every time.
+    """
+    # SHA-512 gives 64 bytes; we expand to dim via repeated hashing.
+    chunks: list[float] = []
+    seed = text.encode("utf-8", errors="ignore")
+    while len(chunks) < dim:
+        seed = hashlib.sha512(seed).digest() + seed
+        # Each byte → float in [-1, 1]
+        for byte in seed[: min(64, dim - len(chunks))]:
+            chunks.append((byte / 127.5) - 1.0)
+    # L2 normalize for cosine compatibility
+    arr = np.asarray(chunks[:dim], dtype=np.float32)
+    n = float(np.linalg.norm(arr))
+    if n > 0:
+        arr = arr / n
+    return arr.tolist()
+
+
+def _ollama_embed(text: str, dim: int = EMBEDDING_DIM) -> list[float] | None:
+    """Try Ollama bge-m3. Returns None on any failure (caller falls back)."""
+    try:
+        with httpx.Client(timeout=10.0) as c:
+            r = c.post(
+                f"{OLLAMA_URL}/api/embeddings",
+                json={"model": OLLAMA_MODEL, "prompt": text[:8000]},
+            )
+            if r.status_code != 200:
+                return None
+            vec = r.json().get("embedding")
+            if not vec or not isinstance(vec, list):
+                return None
+            arr = np.asarray(vec, dtype=np.float32)
+            # Resize to EMBEDDING_DIM (pad with 0 or truncate)
+            if arr.shape[0] >= dim:
+                arr = arr[:dim]
+            else:
+                arr = np.concatenate([arr, np.zeros(dim - arr.shape[0], dtype=np.float32)])
+            n = float(np.linalg.norm(arr))
+            if n > 0:
+                arr = arr / n
+            return arr.tolist()
+    except Exception as e:
+        log.debug("ollama_embed_failed: %s", e)
+        return None
+
+
+def _openrouter_embed(text: str, dim: int = EMBEDDING_DIM) -> list[float] | None:
+    """Try OpenRouter Nemotron. Returns None on any failure."""
+    key = os.getenv("OPENROUTER_API_KEY")
+    if not key:
+        return None
+    try:
+        with httpx.Client(timeout=15.0) as c:
+            r = c.post(
+                OPENROUTER_URL,
+                headers={"Authorization": f"Bearer {key}"},
+                json={"model": OPENROUTER_MODEL, "input": text[:8000]},
+            )
+            if r.status_code != 200:
+                return None
+            data = r.json().get("data", [])
+            if not data:
+                return None
+            vec = data[0].get("embedding")
+            if not vec:
+                return None
+            arr = np.asarray(vec, dtype=np.float32)
+            if arr.shape[0] >= dim:
+                arr = arr[:dim]
+            else:
+                arr = np.concatenate([arr, np.zeros(dim - arr.shape[0], dtype=np.float32)])
+            n = float(np.linalg.norm(arr))
+            if n > 0:
+                arr = arr / n
+            return arr.tolist()
+    except Exception as e:
+        log.debug("openrouter_embed_failed: %s", e)
+        return None
+
+
+async def get_embedding(text: str) -> list[float]:
+    """Async embedding API. Always returns a 1024d float list.
+
+    Tries in order: Ollama → OpenRouter → Hash. Logs which tier served.
+    """
+    if not text or not text.strip():
+        return _hash_embed("", EMBEDDING_DIM)
+
+    if _BACKEND == "hash":
+        return _hash_embed(text)
+    if _BACKEND == "ollama":
+        v = _ollama_embed(text)
+        return v if v is not None else _hash_embed(text)
+    if _BACKEND == "openrouter":
+        v = _openrouter_embed(text)
+        return v if v is not None else _hash_embed(text)
+
+    # auto: try ollama → openrouter → hash
+    if _ollama_reachable():
+        v = _ollama_embed(text)
+        if v is not None:
+            return v
+    v = _openrouter_embed(text)
+    if v is not None:
+        return v
+    return _hash_embed(text)
+
+
+async def get_embeddings(texts: list[str]) -> list[list[float]]:
+    """Batch embedding. Sequential calls — Ollama and OpenRouter handle
+    larger requests poorly and we'd rather keep memory bounded."""
+    return [await get_embedding(t) for t in texts]
+
+
+def current_backend() -> str:
+    """Which tier is currently active? For observability."""
+    if _BACKEND != "auto":
+        return _BACKEND
+    if _ollama_reachable():
+        return "ollama"
+    if os.getenv("OPENROUTER_API_KEY"):
+        return "openrouter"
+    return "hash"

backend/app/rag/engine.py

@@ -0,0 +1,308 @@
+"""M3 RAG Engine — three-pillar search + ingest.
+
+The missing module that app/rag/service.py was importing (the legacy
+app.rag_engine was nuked during consolidation but never replaced).
+
+Three-Pillar Search (per 2026 RAG standards):
+  Pillar 1: ANN vector similarity (semantic match)
+  Pillar 2: BM25-lite keyword search (lexical match)
+  Pillar 3: Metadata filter (structured constraints)
+  Fusion: Reciprocal Rank Fusion (RRF) — k=60
+
+Ingest:
+  - Chunk → embed → store in ANN index + Redis doc store
+  - MD5 dedup per collection
+  - Per-chunk doc_id = "{collection}:{hash}:{idx}"
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any
+
+from app.rag.ann_index import Hit, get_index
+from app.rag.chunking import chunk_text, content_hash, is_duplicate, mark_ingested
+from app.rag.embeddings import current_backend, get_embedding
+
+log = logging.getLogger(__name__)
+
+# RRF constant
+_RRF_K = 60
+
+
+# ── Three-pillar search ──────────────────────────────────────────────
+async def three_pillar_search(
+    query: str,
+    collection: str = "scam_intel",
+    top_k: int = 5,
+    min_similarity: float = 0.0,
+    filters: dict[str, Any] | None = None,
+) -> list[dict]:
+    """Run the three-pillar search and return RRF-fused hits.
+
+    Each returned hit: {doc_id, score, text, metadata, source_pillars}
+    """
+    start = time.monotonic()
+
+    # Pillar 1: ANN vector search
+    qvec = await get_embedding(query)
+    idx = get_index(collection)
+    ann_hits = idx.search(qvec, top_k=top_k * 3, min_similarity=min_similarity)
+
+    # Pillar 2: keyword (BM25-lite via Redis text scan on stored docs)
+    keyword_hits = await _keyword_search(query, collection, top_k * 3)
+
+    # Pillar 3: metadata filter (apply to ann+keyword results)
+    filtered = _apply_filters(ann_hits + keyword_hits, filters or {})
+
+    # RRF fusion
+    fused = _reciprocal_rank_fusion([ann_hits, keyword_hits], top_k=top_k)
+
+    # Apply filters post-fusion
+    if filters:
+        fused = [h for h in fused if _matches_filters(h, filters)]
+
+    took_ms = int((time.monotonic() - start) * 1000)
+    log.info(
+        "rag_search_done collection=%s ann=%d kw=%d fused=%d took_ms=%d backend=%s",
+        collection, len(ann_hits), len(keyword_hits), len(fused), took_ms, current_backend(),
+    )
+    return [
+        {
+            "doc_id": h.doc_id,
+            "score": h.score,
+            "text": h.text,
+            "metadata": h.metadata,
+        }
+        for h in fused
+    ]
+
+
+async def _keyword_search(
+    query: str, collection: str, limit: int
+) -> list[Hit]:
+    """BM25-lite keyword search. Simple TF scoring on stored text.
+
+    Returns Hits with score in [0, 1] (normalized). We don't pretend this
+    is real BM25 — but it's good enough for a fallback that surfaces
+    lexically-matching docs the ANN might miss.
+    """
+    try:
+        from app.core.redis import get_redis
+
+        r = get_redis()
+        # Pull all doc texts for this collection's ANN store
+        raw = r.hgetall(f"rag:ann:{collection}:docs")
+    except Exception as e:
+        log.debug("keyword_search_redis_failed: %s", e)
+        return []
+
+    if not raw:
+        return []
+
+    import json
+    terms = [t.lower() for t in query.split() if len(t) > 2]
+    if not terms:
+        return []
+
+    scored: list[tuple[float, str, str, dict]] = []
+    for doc_id, blob in raw.items():
+        try:
+            entry = json.loads(blob)
+        except Exception:
+            continue
+        text = entry.get("text", "")
+        if not text:
+            continue
+        text_l = text.lower()
+        # Simple TF
+        tf = sum(text_l.count(t) for t in terms)
+        if tf == 0:
+            continue
+        # Normalize by length
+        score = tf / max(10, len(text_l.split()))
+        scored.append((score, doc_id, text, entry.get("metadata", {})))
+
+    scored.sort(key=lambda x: -x[0])
+    out: list[Hit] = []
+    max_score = scored[0][0] if scored else 1.0
+    for score, doc_id, text, metadata in scored[:limit]:
+        out.append(
+            Hit(
+                doc_id=doc_id,
+                score=min(1.0, score / max_score) if max_score > 0 else 0.0,
+                text=text,
+                metadata=metadata,
+            )
+        )
+    return out
+
+
+def _apply_filters(hits: list[Hit], filters: dict[str, Any]) -> list[Hit]:
+    """Pre-filter: keep hits whose metadata matches all filter key=val pairs."""
+    if not filters:
+        return hits
+    return [h for h in hits if _matches_filters(h, filters)]
+
+
+def _matches_filters(hit: Hit, filters: dict[str, Any]) -> bool:
+    for k, v in filters.items():
+        if hit.metadata.get(k) != v:
+            return False
+    return True
+
+
+def _reciprocal_rank_fusion(
+    pillar_hits: list[list[Hit]], top_k: int, k: int = _RRF_K
+) -> list[Hit]:
+    """Reciprocal Rank Fusion across multiple ranked lists.
+
+    RRF score(d) = sum( 1 / (k + rank_i(d)) ) for each pillar that contains d.
+    """
+    scores: dict[str, float] = {}
+    by_id: dict[str, Hit] = {}
+    for pillar in pillar_hits:
+        for rank, h in enumerate(pillar, start=1):
+            scores[h.doc_id] = scores.get(h.doc_id, 0.0) + 1.0 / (k + rank)
+            if h.doc_id not in by_id:
+                by_id[h.doc_id] = h
+
+    ranked = sorted(scores.items(), key=lambda x: -x[1])
+    out: list[Hit] = []
+    for doc_id, rrf_score in ranked[:top_k]:
+        h = by_id[doc_id]
+        # Normalize to [0, 1] roughly — RRF max is ~3/k for 3 pillars
+        norm = min(1.0, rrf_score * k / 3.0)
+        out.append(
+            Hit(
+                doc_id=h.doc_id,
+                score=norm,
+                text=h.text,
+                metadata=h.metadata,
+            )
+        )
+    return out
+
+
+# ── Ingest ───────────────────────────────────────────────────────────
+async def ingest_document(
+    collection: str,
+    doc_id: str,
+    content: str,
+    metadata: dict[str, Any] | None = None,
+    chunk: bool = True,
+) -> dict:
+    """Ingest a document into the RAG system.
+
+    - Chunks the content (recursive split, dedup)
+    - Embeds each chunk
+    - Adds to the collection's ANN index
+    - Marks each chunk's hash as ingested for dedup
+    """
+    if not content or not content.strip():
+        return {"doc_id": doc_id, "collection": collection, "status": "empty", "chunks": 0}
+
+    metadata = metadata or {}
+    idx = get_index(collection)
+    chunks = chunk_text(content) if chunk else [
+        # Single-chunk path: still dedup
+        __import__("app.rag.chunking", fromlist=["Chunk"]).Chunk(
+            text=content, content_hash=content_hash(content), index=0, quality=1.0
+        )
+    ]
+
+    # Dedup
+    new_chunks = [c for c in chunks if not is_duplicate(c.content_hash, collection)]
+    skipped = len(chunks) - len(new_chunks)
+    if not new_chunks:
+        return {
+            "doc_id": doc_id,
+            "collection": collection,
+            "status": "duplicate",
+            "chunks": 0,
+            "skipped": skipped,
+        }
+
+    # Embed + insert
+    added = 0
+    for c in new_chunks:
+        vec = await get_embedding(c.text)
+        chunk_doc_id = f"{doc_id}:{c.content_hash[:8]}:{c.index}"
+        chunk_meta = {
+            **metadata,
+            "chunk_index": c.index,
+            "content_hash": c.content_hash,
+            "quality": c.quality,
+        }
+        idx.add(chunk_doc_id, vec, chunk_meta, text=c.text)
+        mark_ingested(c.content_hash, collection)
+        added += 1
+
+    return {
+        "doc_id": doc_id,
+        "collection": collection,
+        "status": "ok",
+        "chunks": added,
+        "skipped": skipped,
+    }
+
+
+# ── Stats ────────────────────────────────────────────────────────────
+def get_collection_stats(collection: str) -> dict:
+    """Get stats for one collection: vector count + dedup hashes count."""
+    idx = get_index(collection)
+    count = idx.count()
+    try:
+        from app.core.redis import get_redis
+
+        hashes = get_redis().scard(f"rag:hashes:{collection}")
+    except Exception:
+        hashes = 0
+    return {
+        "collection": collection,
+        "vector_count": count,
+        "dedup_hashes": hashes,
+    }
+
+
+def get_stats(collections: list[str] | None = None) -> dict:
+    """Get stats for all (or specified) collections + the active embedder backend."""
+    from app.rag.models import COLLECTIONS as DEFAULT_COLLECTIONS
+
+    cols = collections or DEFAULT_COLLECTIONS
+    return {
+        "total_docs": sum(get_collection_stats(c)["vector_count"] for c in cols),
+        "backend": current_backend(),
+        "collections": [get_collection_stats(c) for c in cols],
+    }
+
+
+# ── Background helpers (used by ingest_cron worker) ──────────────────
+async def bulk_ingest(
+    items: list[dict],
+    collection: str = "scam_intel",
+) -> dict:
+    """Ingest many items sequentially. Returns summary counts."""
+    if not items:
+        return {"total": 0, "ok": 0, "duplicate": 0, "empty": 0, "errors": 0}
+
+    counts = {"total": len(items), "ok": 0, "duplicate": 0, "empty": 0, "errors": 0}
+    for it in items:
+        try:
+            r = await ingest_document(
+                collection=collection,
+                doc_id=it.get("doc_id") or it.get("id", f"bulk:{int(time.time()*1000)}"),
+                content=it.get("content") or it.get("text", ""),
+                metadata=it.get("metadata") or {},
+            )
+            if r["status"] == "ok":
+                counts["ok"] += 1
+            elif r["status"] == "duplicate":
+                counts["duplicate"] += 1
+            elif r["status"] == "empty":
+                counts["empty"] += 1
+        except Exception as e:
+            log.warning("bulk_ingest_item_failed: %s", e)
+            counts["errors"] += 1
+    return counts

backend/app/rag/service.py

@@ -1,8 +1,7 @@
-"""RAG service — facade over the 14 legacy RAG modules.
+"""RAG service — HTTP facade over the v3 RAG engine.
 
-Provides a clean async Pydantic surface for search, ingest, feedback,
-firehose, and embeddings. Delegates to the proven implementations
-in app.rag_engine and friends.
+Provides a clean async Pydantic surface for search, ingest, feedback.
+Delegates to the M3 engine in app.rag.engine (built per v3 unfuck plan).
 """
 from __future__ import annotations
 
@@ -11,6 +10,18 @@ import time
 from typing import Any
 
 from app.core.logging import get_logger
+from app.rag.engine import (
+    bulk_ingest as engine_bulk_ingest,
+)
+from app.rag.engine import (
+    get_stats as engine_get_stats,
+)
+from app.rag.engine import (
+    ingest_document as engine_ingest_document,
+)
+from app.rag.engine import (
+    three_pillar_search as engine_three_pillar_search,
+)
 from app.rag.models import (
     FeedbackRecord,
     IngestRequest,
@@ -24,7 +35,7 @@ log = get_logger(__name__)
 
 
 class RAGService:
-    """Async facade over the RAG system."""
+    """Async facade over the v3 RAG engine (M3 deliverable)."""
 
     async def search(self, req: SearchRequest) -> SearchResponse:
         """Search the RAG system. Returns a Pydantic response."""
@@ -36,12 +47,12 @@ class RAGService:
         )
         start = time.monotonic()
         try:
-            from app.rag_engine import three_pillar_search
-            raw = await three_pillar_search(
+            raw = await engine_three_pillar_search(
                 query=req.query,
                 collection=req.collection,
                 top_k=req.top_k,
                 min_similarity=req.min_similarity,
+                filters=req.filters or None,
             )
         except Exception as e:
             log.warning("rag_search_failed", error=str(e))
@@ -64,9 +75,8 @@ class RAGService:
             content_len=len(req.content),
         )
         try:
-            from app.rag_engine import ingest_document
-            doc_id = req.doc_id or f"doc:{int(time.time())}"
-            await ingest_document(
+            doc_id = req.doc_id or f"doc:{int(time.time()*1000)}"
+            r = await engine_ingest_document(
                 collection=req.collection,
                 doc_id=doc_id,
                 content=req.content,
@@ -75,8 +85,8 @@ class RAGService:
             return IngestResult(
                 doc_id=doc_id,
                 collection=req.collection,
-                status="ok",
-                chunks=1,
+                status=r.get("status", "ok"),
+                chunks=r.get("chunks", 0),
             )
         except Exception as e:
             log.warning("rag_ingest_failed", error=str(e))
@@ -87,6 +97,14 @@ class RAGService:
                 error=str(e),
             )
 
+    async def stats(self) -> dict:
+        """Return engine stats (collection counts + active embedder backend)."""
+        try:
+            return engine_get_stats()
+        except Exception as e:
+            log.warning("rag_stats_failed", error=str(e))
+            return {"total_docs": 0, "backend": "unknown", "collections": [], "error": str(e)}
+
     async def record_feedback(self, record: FeedbackRecord) -> bool:
         """Record a scanner → RAG feedback (ingest a known scam)."""
         log.info(
@@ -141,16 +159,17 @@ class RAGService:
 
 
 async def init_rag() -> None:
-    """Initialize the RAG system. Called from app.core.lifespan."""
-    log.info("rag_init_started")
+    """Initialize the RAG system. Called from app.core.lifespan.
+
+    v3: no-op (the new engine is lazy — collections load on first search).
+    Kept for backward compat with lifespan hooks.
+    """
+    log.info("rag_init_started", note="v3 engine is lazy — no warmup needed")
     try:
-        from app.rag_engine import init_rag as _legacy_init
-        if asyncio.iscoroutinefunction(_legacy_init):
-            await _legacy_init()
-        else:
-            result = _legacy_init()
-            if asyncio.iscoroutine(result):
-                await result
+        # Touch the redis client to fail fast if misconfigured
+        from app.core.redis import get_redis
+
+        get_redis().ping()
         log.info("rag_init_complete")
     except Exception as e:
-        log.warning("rag_init_failed", error=str(e))
+        log.warning("rag_init_redis_probe_failed", error=str(e))