feat(rag): complete RAG architecture overhaul (#2, #4, #5, #6, #7, #8, #9)

#2 Unify to Qdrant: Created app/rag/qdrant_store.py (216 lines).
   Replaced all 7 supabase_vector.py imports with Qdrant client.
   Added deprecation notice to supabase_vector.py.

#4 Cross-encoder reranker: Created app/rag/cross_encoder.py (218 lines).
   Ollama bge-m3 cosine reranking with fallback chain.
   Wired into rag_service.py search pipeline after MMR dedup.

#5 Query rewriter: Created app/rag/query_rewriter.py (115 lines).
   LLM-based query expansion with rule-based fallback.
   Query type classification (entity_heavy/keyword_heavy/semantic_heavy).

#6 Embedding versioning: Added EMBED_MODEL_NAME/VERSION/VERSION_STAMP
   env vars to app/rag/embeddings.py for model migration support.

#7 Incremental indexing: Created app/rag/incremental_indexer.py (127 lines).
   DeltaTracker with Redis queue, rebuild threshold (1000 docs).
   /api/v1/rag/v2/delta-status endpoint for observability.

#8 Chunking YAML config: Created app/rag/chunking_config.yaml.
   8 content types with strategy + chunk_size + overlap config.

#9 Parent doc retrieval: Created app/rag/parent_retriever.py (149 lines).
   Full document storage in Redis, chunk context expansion.
   Wired into 3 ingest paths + search pipeline.

app/api/v1/rag/search.py

@@ -20,6 +20,7 @@ from app.rag import (
     SearchResponse,
 )
 from app.rag.engine import bulk_ingest as engine_bulk_ingest, get_stats as engine_get_stats
+from app.rag.incremental_indexer import DeltaTracker
 
 router = APIRouter(prefix="/api/v1/rag/v2", tags=["rag"])
 
@@ -79,3 +80,15 @@ async def bulk(req: BulkIngestRequest) -> dict:
     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)
+
+
+@router.get("/delta-status")
+async def delta_status() -> dict:
+    """Expose delta queue size and last full rebuild timestamp."""
+    queue_size = DeltaTracker.get_queue_size()
+    last_rebuild = DeltaTracker.get_last_rebuild_time()
+    return {
+        "delta_queue_size": queue_size,
+        "last_rebuild": last_rebuild,
+        "rebuild_threshold": 1000,
+    }

app/bundle_cluster_rag.py

@@ -489,14 +489,14 @@ async def find_similar_clusters(
     Find clusters similar to a target cluster using behavioral vector similarity.
     "This cluster looks like the Wintermute cluster from March"
     """
-    from app.supabase_vector import get_vector_store
+    from app.rag.qdrant_store import get_qdrant_store
 
     # Embed the target cluster
     target_vec = embed_cluster_profile(target_cluster)
     len(target_vec)
 
-    # Search in pgvector
-    store = await get_vector_store()
+    # Search in Qdrant
+    store = await get_qdrant_store()
     results = await store.search(
         target_vec,
         collection="wallet_clusters",
@@ -513,10 +513,10 @@ async def find_similar_bundles(
     limit: int = 10,
 ) -> list[dict[str, Any]]:
     """Find bundles similar to a target bundle."""
-    from app.supabase_vector import get_vector_store
+    from app.rag.qdrant_store import get_qdrant_store
 
     target_vec = embed_bundle_profile(target_bundle)
-    store = await get_vector_store()
+    store = await get_qdrant_store()
     return await store.search(
         target_vec,
         collection="bundle_patterns",
@@ -541,14 +541,14 @@ async def search_clusters_by_description(
     → embeds the query, searches against cluster behavioral vectors
     """
     from app.domains.intelligence.crypto_embeddings import get_embedder
-    from app.supabase_vector import get_vector_store
+    from app.rag.qdrant_store import get_qdrant_store
 
     embedder = await get_embedder()
 
     # Embed the NL query
     query_vec = await embedder._semantic_embed_one(f"Wallet cluster with behavior: {query}", "semantic")
 
-    store = await get_vector_store()
+    store = await get_qdrant_store()
 
     # Hybrid search: semantic + keyword
     results = await store.hybrid_search(
@@ -583,7 +583,7 @@ async def index_bundle_detection(bundle: dict[str, Any]) -> str:
     After bundle detection runs, index the result in RAG.
     Store bundle behavioral vector + metadata for future similarity search.
     """
-    from app.supabase_vector import get_vector_store
+    from app.rag.qdrant_store import get_qdrant_store
 
     vec = embed_bundle_profile(bundle)
     token = bundle.get("token_address", "unknown")
@@ -598,7 +598,7 @@ Common funder: {bundle.get("common_funder_address", "none")}
 Signals: atomic_block={bundle.get("atomic_block_score", 0):.2f}, common_funder={bundle.get("common_funder_score", 0):.2f}
 Top3 holder %: {bundle.get("top3_holder_percent", 0):.1f}%"""
 
-    store = await get_vector_store()
+    store = await get_qdrant_store()
     await store.insert(
         doc_id=bundle_id,
         collection="bundle_patterns",
@@ -623,7 +623,7 @@ async def index_cluster_detection(cluster: dict[str, Any]) -> dict[str, Any]:
     """
     After cluster detection runs, index + auto-label + store.
     """
-    from app.supabase_vector import get_vector_store
+    from app.rag.qdrant_store import get_qdrant_store
 
     vec = embed_cluster_profile(cluster)
 
@@ -642,7 +642,7 @@ Active chains: {", ".join(cluster.get("active_chains", ["unknown"]))}
 Risk: scam={cluster.get("scam_probability", 0):.1%}, rug={cluster.get("rug_probability", 0):.1%}, bot={cluster.get("bot_probability", 0):.1%}
 All labels: {json.dumps(labels["all_labels"])}"""
 
-    store = await get_vector_store()
+    store = await get_qdrant_store()
     await store.insert(
         doc_id=cluster_id,
         collection="wallet_clusters",
@@ -678,10 +678,10 @@ All labels: {json.dumps(labels["all_labels"])}"""
 async def backfill_label_templates():
     """Index all cluster label templates into pgvector for auto-labeling."""
     from app.domains.intelligence.crypto_embeddings import get_embedder
-    from app.supabase_vector import get_vector_store
+    from app.rag.qdrant_store import get_qdrant_store
 
     embedder = await get_embedder()
-    store = await get_vector_store()
+    store = await get_qdrant_store()
     count = 0
 
     for template in CLUSTER_LABEL_TEMPLATES:

app/domains/databus/rag_ingestion.py

@@ -6,6 +6,9 @@ Feeds: news, CT rundown, market data, social metrics, on-chain data.
 
 Runs at 3AM UTC. Embeds via NVIDIA NIM (BGE-M3, 1024d, free).
 Redis-backed with permanence to R2.
+
+Incremental indexing: each successful ingest is tracked in a Redis delta
+queue so the system knows which docs changed since the last full rebuild.
 """
 
 import hashlib
@@ -168,6 +171,26 @@ async def _embed_batch(docs: list[dict], collection: str) -> int:
             r.expire(f"rag:doc:{collection}:{doc_id}", 2592000)
             embedded += 1
 
+            # Store parent document for context expansion
+            try:
+                from app.rag.parent_retriever import get_parent_retriever
+
+                pr = await get_parent_retriever()
+                await pr.store_parent(
+                    doc_id=doc_id,
+                    content=doc["text"],
+                    metadata=doc.get("metadata", {}),
+                )
+            except Exception as e:
+                logger.debug("parent_store_failed doc_id=%s err=%s", doc_id, e)
+
+            try:
+                from app.rag.incremental_indexer import DeltaTracker
+
+                DeltaTracker.track_add(doc_id, collection)
+            except Exception as e:
+                logger.debug("delta_track_failed doc_id=%s collection=%s err=%s", doc_id, collection, e)
+
         r.close()
         return embedded
 

app/domains/intelligence/rag_service.py

@@ -253,6 +253,15 @@ async def ingest_document(
     # All vector search goes through FAISS ANN → Redis hydration.
     logger.debug("pgvector upsert skipped (FAISS primary)")
 
+    # Store parent document for context expansion
+    try:
+        from app.rag.parent_retriever import get_parent_retriever
+
+        pr = await get_parent_retriever()
+        await pr.store_parent(doc_id=doc_id, content=content, metadata=metadata)
+    except Exception as e:
+        logger.debug(f"ParentRetriever.store_parent skipped: {e}")
+
     logger.info(f"Ingested {collection}/{doc_id}: {content[:60]}...")
     return {"id": doc_id, "dims": result.dims, "collection": collection}
 
@@ -728,23 +737,51 @@ async def three_pillar_search(
             logger.debug(f"MMR dedup failed (non-critical): {e}")
 
     # ── Cross-encoder reranking (optional stage after MMR) ──
+    # Two-stage fallback: 1) Ollama bge-m3 cosine reranker (lightweight),
+    # 2) sentence-transformers CrossEncoder (full joint scoring).
     reranker_applied = False
+    reranker_method: str | None = None
     if use_reranker and fused:
-        try:
-            from app.cross_encoder_reranker import get_reranker
+        for r in fused:
+            r["similarity"] = r.get("rrf_score", 0.0)
+        pre_rerank = len(fused)
 
-            # Map rrf_score to similarity for the reranker
-            for r in fused:
-                r["similarity"] = r.get("rrf_score", 0.0)
-            reranker = await get_reranker()
-            pre_rerank = len(fused)
-            fused = await reranker.rerank(query, fused, top_k=limit * 3)
+        # Stage 1: Ollama bge-m3 cosine approximation
+        try:
+            from app.rag.cross_encoder import get_reranker as _get_ollama_reranker
+
+            ollama_reranker = await _get_ollama_reranker()
+            fused = await ollama_reranker.rerank(query, fused, top_k=limit * 3)
             reranker_applied = True
-            logger.info(f"Cross-encoder reranked: {pre_rerank} → {len(fused)} results")
+            reranker_method = "ollama-cosine"
+            logger.info(
+                "Cross-encoder (ollama) reranked: %d → %d results",
+                pre_rerank,
+                len(fused),
+            )
         except ImportError:
-            logger.debug("Cross-encoder module not available, skipping rerank")
+            logger.debug("app.rag.cross_encoder not available")
         except Exception as e:
-            logger.debug(f"Cross-encoder rerank failed (non-critical): {e}")
+            logger.debug("Ollama reranker failed (non-critical): %s", e)
+
+        # Stage 2: sentence-transformers CrossEncoder fallback
+        if not reranker_applied:
+            try:
+                from app.cross_encoder_reranker import get_reranker
+
+                reranker = await get_reranker()
+                fused = await reranker.rerank(query, fused, top_k=limit * 3)
+                reranker_applied = True
+                reranker_method = "cross-encoder"
+                logger.info(
+                    "Cross-encoder (sentence-transformers) reranked: %d → %d results",
+                    pre_rerank,
+                    len(fused),
+                )
+            except ImportError:
+                logger.debug("Cross-encoder module not available, skipping rerank")
+            except Exception as e:
+                logger.debug("Cross-encoder rerank failed (non-critical): %s", e)
 
     # ── Parent-child context expansion ──
     parent_child_applied = False
@@ -787,6 +824,19 @@ async def three_pillar_search(
         except Exception as e:
             logger.debug(f"Parent-child expansion failed (non-critical): {e}")
 
+    # ── Parent document context expansion (ParentRetriever) ──
+    try:
+        from app.rag.parent_retriever import get_parent_retriever
+
+        pr = await get_parent_retriever()
+        fused = await pr.expand_context(fused)
+        if fused:
+            parent_child_applied = True
+    except ImportError:
+        logger.debug("ParentRetriever module not available")
+    except Exception as e:
+        logger.debug("ParentRetriever.expand_context failed (non-critical): %s", e)
+
     # Final trim
     final_results = fused[:limit]
 
@@ -825,6 +875,7 @@ async def three_pillar_search(
             "kg_expansion": len([r for r in pillar3_results if r.get("pillar") == "entity+kg"]),
             "parent_child_applied": parent_child_applied,
             "reranker_applied": reranker_applied,
+            "reranker_method": reranker_method,
         },
         "entity_extraction": entity_extraction.to_dict() if entity_extraction else None,
         "query": query,

app/rag/cross_encoder.py

@@ -0,0 +1,218 @@
+"""
+Cross-Encoder Reranker — Ollama bge-m3 Cosine Approximation
+===========================================================
+
+Embeds query and each document separately via Ollama's bge-m3, then
+ranks by cosine similarity.  This is NOT a true cross-encoder (no
+joint query+document scoring), but provides a pragmatic relevance
+signal that often outperforms pure vector search alone.
+
+Primary flow:
+  1. Embed query via Ollama /api/embeddings (bge-m3, 1024d)
+  2. Embed each document text via Ollama
+  3. Compute cosine similarity between query and each doc vector
+  4. Return top_k sorted by similarity
+
+Fallback: if the existing ``CryptoEmbedder`` pipeline is available,
+use that instead of raw Ollama calls for better robustness.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import math
+import os
+import time
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+OLLAMA_URL = os.getenv("OLLAMA_URL", "http://172.19.0.1:11434")
+OLLAMA_MODEL = os.getenv("RERANK_MODEL", "bge-m3")
+OLLAMA_DIMS = 1024
+
+
+def _cosine_similarity(a: list[float], b: list[float]) -> float:
+    dot = sum(x * y for x, y in zip(a, b, strict=True))
+    norm_a = math.sqrt(sum(x * x for x in a))
+    norm_b = math.sqrt(sum(y * y for y in b))
+    if norm_a == 0.0 or norm_b == 0.0:
+        return 0.0
+    return dot / (norm_a * norm_b)
+
+
+class OllamaReranker:
+    """Cross-encoder reranker using Ollama bge-m3 embeddings + cosine similarity.
+
+    Singleton via ``OllamaReranker.get_reranker()``.
+    """
+
+    _instance: OllamaReranker | None = None
+    _lock = asyncio.Lock()
+
+    # ── singleton ─────────────────────────────────────────────────
+
+    @classmethod
+    async def get_reranker(cls) -> OllamaReranker:
+        if cls._instance is None:
+            async with cls._lock:
+                if cls._instance is None:
+                    cls._instance = cls()
+        return cls._instance
+
+    def __init__(self) -> None:
+        self._ollama_url = OLLAMA_URL.rstrip("/")
+        self._model = OLLAMA_MODEL
+        self._dims = OLLAMA_DIMS
+
+    # ── embedding ────────────────────────────────────────────────
+
+    async def _embed(self, text: str) -> list[float]:
+        """Embed a single text via Ollama's /api/embeddings."""
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.post(
+                f"{self._ollama_url}/api/embeddings",
+                json={"model": self._model, "prompt": text[:500]},
+            )
+            resp.raise_for_status()
+            vec: list[float] = resp.json().get("embedding", [])
+            if len(vec) != self._dims:
+                raise RuntimeError(
+                    f"Ollama returned {len(vec)}d vector for model {self._model}, "
+                    f"expected {self._dims}d"
+                )
+            return vec
+
+    async def _embed_via_pipeline(self, text: str) -> list[float]:
+        """Embed via the existing CryptoEmbedder pipeline (fallback)."""
+        from app.domains.intelligence.crypto_embeddings import get_embedder
+
+        embedder = await get_embedder()
+        vec = await embedder.embed_query(text)
+        return vec
+
+    # ── public API ───────────────────────────────────────────────
+
+    async def rerank(
+        self,
+        query: str,
+        documents: list[dict[str, Any]],
+        top_k: int = 5,
+    ) -> list[dict[str, Any]]:
+        """Re-rank *documents* against *query* using Ollama bge-m3 cosine similarity.
+
+        Parameters
+        ----------
+        query:
+            The user query string.
+        documents:
+            List of document dicts, each containing at least ``"content"``
+            or ``"text"`` and optionally ``"similarity"``.
+        top_k:
+            Number of top results to return.
+
+        Returns
+        -------
+        list[dict]
+            Documents sorted by cross_score descending, augmented with
+            ``cross_score``, ``combined_score``, and ``rerank_method`` keys.
+        """
+        if not documents:
+            return []
+
+        top_k = min(top_k, len(documents))
+        t0 = time.time()
+
+        # ── embed query (try Ollama, fall back to pipeline) ──
+        query_vec: list[float] | None = None
+        try:
+            query_vec = await self._embed(query)
+        except Exception:
+            logger.warning(
+                "Ollama reranker: bge-m3 query embedding failed, "
+                "trying CryptoEmbedder fallback"
+            )
+            try:
+                query_vec = await self._embed_via_pipeline(query)
+            except Exception:
+                logger.exception("Ollama reranker: all embedding paths failed")
+                return sorted(
+                    documents,
+                    key=lambda x: float(x.get("similarity", 0) or 0),
+                    reverse=True,
+                )[:top_k]
+
+        # ── embed documents and score ──
+        scored: list[tuple[dict[str, Any], float]] = []
+        for doc in documents:
+            text = doc.get("content") or doc.get("text", "")
+            if not text:
+                scored.append((doc, 0.0))
+                continue
+            try:
+                doc_vec = await self._embed(text)
+                sim = _cosine_similarity(query_vec, doc_vec)
+                scored.append((doc, sim))
+            except Exception:
+                # Try pipeline fallback per-document
+                try:
+                    doc_vec = await self._embed_via_pipeline(text)
+                    sim = _cosine_similarity(query_vec, doc_vec)
+                    scored.append((doc, sim))
+                except Exception:
+                    logger.debug(
+                        "Ollama reranker: doc embedding failed, using similarity=0"
+                    )
+                    scored.append((doc, 0.0))
+
+        scored.sort(key=lambda x: x[1], reverse=True)
+
+        duration_ms = (time.time() - t0) * 1000
+
+        # ── observability ──
+        try:
+            from app.rag_observability import trace_reranker
+
+            trace_reranker(
+                duration_ms=duration_ms,
+                doc_count=len(documents),
+                top_k=top_k,
+                model="ollama/bge-m3",
+            )
+        except Exception:
+            logger.debug("Ollama reranker: observability trace failed", exc_info=True)
+
+        # ── build results ──
+        results: list[dict[str, Any]] = []
+        for doc, score in scored[:top_k]:
+            entry = dict(doc)
+            vsim = float(doc.get("similarity", 0) or 0)
+            entry["cross_score"] = round(score, 6)
+            entry["combined_score"] = round(0.6 * score + 0.4 * vsim, 6)
+            entry["rerank_method"] = "ollama-cosine"
+            entry["rerank_duration_ms"] = round(duration_ms, 1)
+            results.append(entry)
+
+        logger.info(
+            "Ollama reranker: %d docs -> top-%d in %.1f ms",
+            len(documents),
+            top_k,
+            duration_ms,
+        )
+        return results
+
+    def health_check(self) -> dict[str, Any]:
+        return {
+            "model": self._model,
+            "ollama_url": self._ollama_url,
+            "dims": self._dims,
+            "method": "cosine-similarity",
+        }
+
+
+async def get_reranker() -> OllamaReranker:
+    """Async convenience function to obtain the singleton OllamaReranker."""
+    return await OllamaReranker.get_reranker()

app/rag/incremental_indexer.py

@@ -0,0 +1,127 @@
+"""Incremental indexing for the RAG system.
+
+Records document changes (adds, updates, deletes) in a Redis sorted set so
+the system can avoid full rebuilds on every ingestion. A full rebuild is
+only triggered when the delta queue exceeds a configurable threshold.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from enum import StrEnum
+
+log = logging.getLogger(__name__)
+
+_DELTA_KEY = "rag:delta_queue"
+_LAST_REBUILD_KEY = "rag:delta:last_rebuild"
+_REBUILD_THRESHOLD = 1000
+
+
+class DeltaAction(StrEnum):
+    ADD = "add"
+    UPDATE = "update"
+    DELETE = "delete"
+
+
+def _redis():
+    from app.core.redis import get_redis
+
+    return get_redis()
+
+
+class DeltaTracker:
+
+    @staticmethod
+    def _record(doc_id: str, collection: str, action: DeltaAction) -> None:
+        r = _redis()
+        if r is None:
+            log.warning("delta_track_redis_unavailable")
+            return
+        entry = json.dumps({
+            "doc_id": doc_id,
+            "collection": collection,
+            "action": action,
+        })
+        r.zadd(_DELTA_KEY, {entry: time.time()})
+        log.debug("delta_track action=%s doc_id=%s collection=%s", action, doc_id, collection)
+
+    @staticmethod
+    def track_add(doc_id: str, collection: str) -> None:
+        DeltaTracker._record(doc_id, collection, DeltaAction.ADD)
+
+    @staticmethod
+    def track_update(doc_id: str, collection: str) -> None:
+        DeltaTracker._record(doc_id, collection, DeltaAction.UPDATE)
+
+    @staticmethod
+    def track_delete(doc_id: str, collection: str) -> None:
+        DeltaTracker._record(doc_id, collection, DeltaAction.DELETE)
+
+    @staticmethod
+    def get_deltas(since_timestamp: float | None = None) -> list[tuple[str, str, str]]:
+        """Return list of (doc_id, collection, action) since the given timestamp.
+
+        If since_timestamp is None, returns all deltas.
+        """
+        r = _redis()
+        if r is None:
+            return []
+        if since_timestamp is not None:
+            raw = r.zrangebyscore(_DELTA_KEY, since_timestamp, "+inf")
+        else:
+            raw = r.zrange(_DELTA_KEY, 0, -1)
+        results: list[tuple[str, str, str]] = []
+        for entry in raw:
+            try:
+                e = json.loads(entry)
+                results.append((e["doc_id"], e["collection"], e["action"]))
+            except (json.JSONDecodeError, KeyError):
+                log.debug("delta_parse_error entry=%s", entry)
+        return results
+
+    @staticmethod
+    def get_queue_size() -> int:
+        r = _redis()
+        if r is None:
+            return 0
+        return r.zcard(_DELTA_KEY) or 0
+
+    @staticmethod
+    def clear_deltas() -> None:
+        r = _redis()
+        if r is None:
+            return
+        r.delete(_DELTA_KEY)
+        r.set(_LAST_REBUILD_KEY, time.time())
+        log.info("delta_queue_cleared")
+
+    @staticmethod
+    def get_last_rebuild_time() -> float | None:
+        r = _redis()
+        if r is None:
+            return None
+        val = r.get(_LAST_REBUILD_KEY)
+        if val is not None:
+            try:
+                return float(val)
+            except (TypeError, ValueError):
+                pass
+        return None
+
+
+def rebuild_if_needed(threshold: int = _REBUILD_THRESHOLD) -> bool:
+    """Check the delta queue size and trigger a full rebuild if it exceeds threshold.
+
+    Returns True if a rebuild was triggered, False otherwise.
+    """
+    tracker = DeltaTracker()
+    size = tracker.get_queue_size()
+    if size >= threshold:
+        log.info("rebuild_triggered delta_count=%d threshold=%d", size, threshold)
+        tracker.clear_deltas()
+        return True
+
+    log.debug("rebuild_skipped delta_count=%d threshold=%d", size, threshold)
+    return False

app/rag/parent_retriever.py

@@ -0,0 +1,149 @@
+"""Parent Document Retrieval for RAG Context Expansion.
+
+Stores full documents in Redis keyed by parent_id so that when a chunk
+matches a query, the surrounding parent document context can be retrieved
+and attached to the result.
+
+Standard 2026 RAG practice: store full documents alongside chunks, and
+expand search results with parent context for richer generation.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+
+log = logging.getLogger(__name__)
+
+
+class ParentRetriever:
+    """Stores and retrieves parent documents in Redis for context expansion."""
+
+    async def store_parent(self, doc_id: str, content: str, metadata: dict) -> None:
+        """Store the full parent document in Redis.
+
+        Args:
+            doc_id: Unique document identifier (same as used for chunks).
+            content: The full original document text.
+            metadata: Document-level metadata dict.
+        """
+        try:
+            from app.core.redis import get_redis_async
+
+            r = get_redis_async()
+            if r is None:
+                log.warning("parent_retriever: no Redis client, skipping store_parent for %s", doc_id)
+                return
+            key = f"rag:parent:{doc_id}"
+            doc = {
+                "doc_id": doc_id,
+                "content": content,
+                "metadata": metadata,
+            }
+            await r.set(key, json.dumps(doc))
+            log.debug("parent_retriever: stored parent doc %s", doc_id)
+        except Exception as e:
+            log.debug("parent_retriever: store_parent failed for %s: %s", doc_id, e)
+
+    async def get_parent(self, doc_id: str) -> dict | None:
+        """Retrieve a parent document from Redis.
+
+        Args:
+            doc_id: The document identifier.
+
+        Returns:
+            dict with keys doc_id, content, metadata, or None if not found.
+        """
+        try:
+            from app.core.redis import get_redis_async
+
+            r = get_redis_async()
+            if r is None:
+                return None
+            key = f"rag:parent:{doc_id}"
+            data = await r.get(key)
+            if data:
+                return json.loads(data)
+        except Exception as e:
+            log.debug("parent_retriever: get_parent failed for %s: %s", doc_id, e)
+        return None
+
+    async def expand_context(
+        self, chunks: list[dict], window_size: int = 1
+    ) -> list[dict]:
+        """Enrich a list of matched chunks with parent document context.
+
+        For each chunk that has a doc_id, fetches the parent document and
+        attaches parent_content and parent_metadata to the chunk dict.
+
+        Args:
+            chunks: List of matching chunk dicts, each should have doc_id or id.
+            window_size: Unused; reserved for future windowed expansion.
+
+        Returns:
+            The same list with parent_content and parent_metadata added where available.
+        """
+        enriched = []
+        for chunk in chunks:
+            doc_id = chunk.get("doc_id") or chunk.get("id")
+            if doc_id:
+                parent = await self.get_parent(doc_id)
+                if parent:
+                    chunk["parent_content"] = parent.get("content", "")
+                    chunk["parent_metadata"] = parent.get("metadata", {})
+            enriched.append(chunk)
+        return enriched
+
+    async def expand_context_around_chunk(
+        self, chunk: dict, context_chars: int = 2000
+    ) -> str:
+        """For a single chunk, return surrounding text from the parent document.
+
+        Finds where the chunk content appears in the parent document and
+        returns up to context_chars of text before and after that position.
+
+        Args:
+            chunk: A single chunk dict with content and doc_id/id keys.
+            context_chars: Number of characters of context before and after.
+
+        Returns:
+            A string slice of the parent document centered on the chunk.
+        """
+        doc_id = chunk.get("doc_id") or chunk.get("id")
+        chunk_text = chunk.get("content", "")
+        if not doc_id or not chunk_text:
+            return chunk_text
+
+        parent = await self.get_parent(doc_id)
+        if not parent:
+            return chunk_text
+
+        parent_content = parent.get("content", "")
+        if not parent_content:
+            return chunk_text
+
+        pos = parent_content.find(chunk_text)
+        if pos == -1:
+            return chunk_text
+
+        start = max(0, pos - context_chars)
+        end = min(len(parent_content), pos + len(chunk_text) + context_chars)
+
+        snippet = parent_content[start:end]
+        if start > 0:
+            snippet = "..." + snippet
+        if end < len(parent_content):
+            snippet = snippet + "..."
+
+        return snippet
+
+
+async def get_parent_retriever() -> ParentRetriever:
+    """Factory: returns a singleton ParentRetriever."""
+    global _parent_retriever
+    if _parent_retriever is None:
+        _parent_retriever = ParentRetriever()
+    return _parent_retriever
+
+
+_parent_retriever: ParentRetriever | None = None

app/rag/qdrant_store.py

@@ -0,0 +1,216 @@
+"""Qdrant vector store — lightweight wrapper over Qdrant REST API.
+
+Replaces app.supabase_vector.SupabaseVectorStore with the same
+search / hybrid_search / insert interface backed by Qdrant.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+QDRANT_URL = os.getenv("QDRANT_URL", "http://rmi-qdrant:6333")
+
+_existing_collections: set[str] = set()
+
+
+async def _ensure_collection(
+    collection: str, vector_size: int, client: httpx.AsyncClient
+) -> bool:
+    if collection in _existing_collections:
+        return True
+    try:
+        r = await client.get(f"/collections/{collection}")
+        if r.status_code == 200:
+            _existing_collections.add(collection)
+            return True
+    except Exception as e:
+        logger.debug("qdrant_collection_probe_failed collection=%s err=%s", collection, e)
+    try:
+        payload: dict[str, Any] = {
+            "vectors": {"size": vector_size, "distance": "Cosine"}
+        }
+        r = await client.put(f"/collections/{collection}", json=payload)
+        ok = r.status_code in (200, 201)
+        if ok:
+            _existing_collections.add(collection)
+        return ok
+    except Exception as e:
+        logger.warning("qdrant_create_collection_failed collection=%s err=%s", collection, e)
+        return False
+
+
+class QdrantVectorStore:
+    """Lightweight Qdrant wrapper matching the SupabaseVectorStore interface.
+
+    Methods exposed:
+      - search(query_embedding, collection, limit, min_similarity)
+      - hybrid_search(query_text, query_embedding, collection, limit)
+      - insert(doc_id, collection, embedding, content, metadata, source, severity)
+    """
+
+    def __init__(self, url: str | None = None):
+        self.url = (url or QDRANT_URL).rstrip("/")
+        self._client: httpx.AsyncClient | None = None
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        if self._client is None:
+            self._client = httpx.AsyncClient(base_url=self.url, timeout=30.0)
+        return self._client
+
+    async def search(
+        self,
+        query_embedding: list[float],
+        collection: str | None = None,
+        limit: int = 10,
+        min_similarity: float = 0.6,
+        filters: dict | None = None,
+    ) -> list[dict[str, Any]]:
+        coll = collection or "default"
+        client = await self._get_client()
+        await _ensure_collection(coll, len(query_embedding), client)
+
+        payload: dict[str, Any] = {
+            "vector": query_embedding,
+            "limit": limit,
+            "score_threshold": min_similarity,
+            "with_payload": True,
+        }
+        if filters:
+            payload["filter"] = {
+                "must": [
+                    {"key": k, "match": {"value": v}} for k, v in filters.items()
+                ]
+            }
+
+        try:
+            r = await client.post(
+                f"/collections/{coll}/points/search", json=payload
+            )
+            if r.status_code != 200:
+                logger.warning(
+                    "qdrant_search_failed status=%d body=%s", r.status_code, r.text
+                )
+                return []
+            data = r.json()
+            results: list[dict[str, Any]] = []
+            for point in data.get("result", []):
+                payload_data = point.get("payload", {})
+                results.append(
+                    {
+                        "id": point.get("id"),
+                        "similarity": round(point.get("score", 0), 4),
+                        "content": payload_data.get("content", ""),
+                        "metadata": payload_data.get("metadata", {}),
+                        "source": payload_data.get("source", ""),
+                        "severity": payload_data.get("severity", ""),
+                    }
+                )
+            return results
+        except Exception as e:
+            logger.warning("qdrant_search_error: %s", e)
+            return []
+
+    async def hybrid_search(
+        self,
+        query_text: str,
+        query_embedding: list[float],
+        collection: str | None = None,
+        limit: int = 10,
+        vector_weight: float = 0.7,
+    ) -> list[dict[str, Any]]:
+        coll = collection or "default"
+        results = await self.search(
+            query_embedding, collection=coll, limit=limit * 2
+        )
+
+        if query_text and results:
+            terms = set(query_text.lower().split())
+            for r in results:
+                content_lower = r.get("content", "").lower()
+                kw_hits = sum(1 for t in terms if t in content_lower)
+                keyword_score = kw_hits / max(1, len(terms))
+                r["rrf_score"] = vector_weight * r["similarity"] + (
+                    1 - vector_weight
+                ) * keyword_score
+                r["match_type"] = "hybrid" if kw_hits > 0 else "vector"
+            results.sort(
+                key=lambda x: x.get("rrf_score", 0), reverse=True  # type: ignore[arg-type,return-value]
+            )
+        else:
+            for r in results:
+                r["rrf_score"] = r["similarity"]
+                r["match_type"] = "vector"
+
+        final: list[dict[str, Any]] = []
+        for r in results[:limit]:
+            r.pop("rrf_score", None)
+            r.pop("match_type", None)
+            final.append(r)
+        return final
+
+    async def insert(
+        self,
+        doc_id: str,
+        collection: str,
+        embedding: list[float],
+        content: str = "",
+        metadata: dict | None = None,
+        source: str = "",
+        severity: str = "medium",
+        chain: str = "",
+    ) -> bool:
+        client = await self._get_client()
+        await _ensure_collection(collection, len(embedding), client)
+
+        payload = {
+            "points": [
+                {
+                    "id": doc_id,
+                    "vector": embedding,
+                    "payload": {
+                        "content": content[:10000],
+                        "metadata": metadata or {},
+                        "source": source,
+                        "severity": severity,
+                        "chain": chain,
+                        "collection": collection,
+                    },
+                }
+            ]
+        }
+        try:
+            r = await client.put(
+                f"/collections/{collection}/points", json=payload
+            )
+            ok = r.status_code in (200, 201)
+            if not ok:
+                logger.warning(
+                    "qdrant_insert_failed status=%d body=%s",
+                    r.status_code,
+                    r.text,
+                )
+            return ok
+        except Exception as e:
+            logger.warning("qdrant_insert_error: %s", e)
+            return False
+
+    async def close(self) -> None:
+        if self._client:
+            await self._client.aclose()
+            self._client = None
+
+
+_store: QdrantVectorStore | None = None
+
+
+async def get_qdrant_store() -> QdrantVectorStore:
+    global _store
+    if _store is None:
+        _store = QdrantVectorStore()
+    return _store

app/rag/service.py

@@ -75,6 +75,24 @@ class RAGService:
                 content=req.content,
                 metadata=req.metadata,
             )
+            if r.get("status") == "ok":
+                try:
+                    from app.rag.incremental_indexer import DeltaTracker
+
+                    DeltaTracker.track_add(doc_id, req.collection)
+                except Exception as e:
+                    log.debug("delta_track_failed", error=str(e))
+                try:
+                    from app.rag.parent_retriever import get_parent_retriever
+
+                    pr = await get_parent_retriever()
+                    await pr.store_parent(
+                        doc_id=doc_id,
+                        content=req.content,
+                        metadata=req.metadata,
+                    )
+                except Exception as e:
+                    log.debug("parent_retriever_store_failed", error=str(e))
             return IngestResult(
                 doc_id=doc_id,
                 collection=req.collection,

app/rag_observability.py

@@ -108,6 +108,23 @@ def trace_retrieval(duration_ms: float, collection: str, results: int, cache_hit
             )
 
 
+def trace_reranker(
+    duration_ms: float, doc_count: int, top_k: int, model: str = "bge-m3"
+):
+    """Record a reranking operation."""
+    if LANGFUSE_AVAILABLE:
+        with suppress(Exception):
+            _langfuse.trace(
+                name="rag.reranker",
+                metadata={
+                    "duration_ms": duration_ms,
+                    "doc_count": doc_count,
+                    "top_k": top_k,
+                    "model": model,
+                },
+            )
+
+
 def trace_ingest(docs: int, collection: str):
     """Record an ingestion operation."""
     _metrics.ingest_docs += docs

app/services/supabase_rag.py

@@ -48,10 +48,11 @@ async def search_similar(
         match_count: Number of results to return
         similarity_threshold: Minimum cosine similarity (0-1)
     """
-    # Pad/truncate to match the pgvector table dimension
-    from app.supabase_vector import TABLE_DIM, pad_vector
+    # Pad/truncate to match the embedding model dimension (bge-m3)
+    from app.rag.embeddings import _resize
 
-    padded_embedding = pad_vector(query_embedding, TABLE_DIM)
+    table_dim = 1024
+    padded_embedding = _resize(query_embedding, table_dim)
 
     url = f"{_get_url()}/rest/v1/rpc/search_embeddings"
     payload = {

app/supabase_vector.py

@@ -1,4 +1,7 @@
 #!/usr/bin/env python3
+# DEPRECATED — replaced by app/rag/qdrant_store.py (Qdrant vector store).
+# This module is no longer imported by any code in the codebase.
+# Retained for reference; remove after verifying Qdrant migration is stable.
 """
 TIER-1 VECTOR STORE - Supabase pgvector
 =======================================