From 504d276bf6ecbb86d2518811db2eb0579b502fec Mon Sep 17 00:00:00 2001
From: sanjeevafk <strucker08@gmail.com>
Date: Fri, 4 Sep 2026 18:52:31 +0530
Subject: [PATCH 1/2] feat(retrieval): dense-first fallback, citation
 enforcement, and redis query cache with quotas

- Add dense_search_v5 pgvector SQL function (migration 004) and gate dense-first retrieval
- Enforce [n] citation markers in LLM generation with automatic retry on zero citations
- Implement Redis-backed query result cache and per-key daily token quotas with fail-open semantics
- Update CRAG confidence evaluation for dense cosine similarity vs fused RRF score scales
- Mount migration 004 and Redis service in docker-compose and validate_rag_local.sh
- Add hermetic unit test suites for cache, citations, dense-first fallback, and cognitive depth
---
 api/adapters/pg_adapter.py                |   1 +
 api/config.py                             |   9 +
 api/routers/query.py                      | 238 ++++++++++++++++++----
 api/services/cache.py                     | 144 +++++++++++++
 api/services/inference/inference.py       |  93 +++++++--
 crates/depth_engine/src/retrieval/crag.rs |  48 +++++
 db/migrations/004_dense_search.sql        |  23 +++
 docker-compose.yml                        |   1 +
 scripts/validate_rag_local.sh             |   1 +
 tests/conftest.py                         |  43 ++++
 tests/unit/test_citation_enforcement.py   | 104 ++++++++++
 tests/unit/test_cognitive_depth.py        |  37 +++-
 tests/unit/test_dense_first.py            | 134 ++++++++++++
 tests/unit/test_query_cache.py            | 136 +++++++++++++
 uv.lock                                   |  30 +--
 15 files changed, 966 insertions(+), 76 deletions(-)
 create mode 100644 api/services/cache.py
 create mode 100644 db/migrations/004_dense_search.sql
 create mode 100644 tests/conftest.py
 create mode 100644 tests/unit/test_citation_enforcement.py
 create mode 100644 tests/unit/test_dense_first.py
 create mode 100644 tests/unit/test_query_cache.py

diff --git a/api/adapters/pg_adapter.py b/api/adapters/pg_adapter.py
index bf138ed..3ed8f41 100644
--- a/api/adapters/pg_adapter.py
+++ b/api/adapters/pg_adapter.py
@@ -26,6 +26,7 @@ def get_pool() -> asyncpg.Pool:
     "hybrid_search_trusted_v5",
     "hybrid_search_with_graph_v5",
     "hybrid_search_trusted_with_graph_v5",
+    "dense_search_v5",
     "queue_document",
     "dequeue_document",
     "complete_document",
diff --git a/api/config.py b/api/config.py
index 89a4976..f7f350e 100644
--- a/api/config.py
+++ b/api/config.py
@@ -37,6 +37,15 @@ class Settings(BaseSettings):
     stream_fallback_budget_seconds: int = 8
     trusted_proxies: str = ""
 
+    # Retrieval strategy: dense-first with hybrid fallback.
+    dense_hit_min_results: int = 5
+    dense_hit_min_similarity: float = 0.5
+    rerank_skip_similarity: float = 0.8
+
+    # Query-result cache (Redis) and per-key daily token quotas.
+    query_cache_ttl_seconds: int = 3600
+    quota_enabled: bool = True
+
     redis_url: str = "redis://localhost:6379"
     database_url: str = "postgresql://depthapi:depthapi@localhost:5432/depthapi"
     cache_ttl: int = 86400
diff --git a/api/routers/query.py b/api/routers/query.py
index 5a49b90..cadd4bd 100644
--- a/api/routers/query.py
+++ b/api/routers/query.py
@@ -9,7 +9,13 @@
 from pydantic import BaseModel, Field
 
 from api.adapters.pg_adapter import execute_rpc, get_pool
-from api.services.inference.inference import generate_response, generate_stream_response
+from api.config import get_settings
+from api.services import cache as query_cache
+from api.services.inference.inference import (
+    generate_response,
+    generate_stream_response,
+    has_citation_markers,
+)
 from api.services.rag.context_processing import reorder_lost_in_the_middle
 from api.services.rag.embeddings import embed_texts
 from api.services.rag.graph.router import detect_graph_hops
@@ -29,6 +35,27 @@
 log = logging.getLogger(__name__)
 
 
+def _confidence_from_scores(contexts: list[dict[str, Any]]) -> str:
+    """Map retrieval scores to a confidence band.
+
+    Dense cosine similarity (~0.3-1.0) and hybrid RRF-fused scores (<0.05)
+    live on different scales, so each uses its own thresholds.
+    """
+    if any(c.get("match_source") == "dense" for c in contexts):
+        top = max((c.get("score") or 0.0) for c in contexts)
+        if top < 0.55:
+            return "low"
+        if top < 0.7:
+            return "medium"
+        return "high"
+    top = max(c.get("score", 0.0) for c in contexts)
+    if top < 0.012:
+        return "low"
+    if top < 0.020:
+        return "medium"
+    return "high"
+
+
 class QueryRequest(BaseModel):
     query: str = Field(..., min_length=1, max_length=8000)
     collection_id: str | None = None
@@ -46,7 +73,7 @@ class QueryRequest(BaseModel):
         default=3,
         ge=1,
         le=5,
-        description="Cognitive depth level (1-2: direct concept summaries, 3-4: scoped hybrid 1-hop, 5: deep 2-hop graph + rerank).",
+        description="Cognitive depth level (1-2: direct concept summaries, 3-4: dense-first hybrid, 5: deep retrieval + forced rerank; graph hops only on detected intent or manual override).",
     )
     save_to_wiki: bool = Field(
         default=False,
@@ -64,6 +91,30 @@ class QueryResponse(BaseModel):
 
 @router.post("/query", response_model=QueryResponse)
 async def query(req: QueryRequest, request: Request, _api_key: ApiKeyRecord = Depends(verify_api_key)) -> QueryResponse:
+    settings = get_settings()
+    model = settings.llm_model
+    # save_to_wiki has a side effect, so it never reads or populates the cache.
+    use_cache = not req.bypass_cache and not req.save_to_wiki
+    ckey = query_cache.cache_key(
+        _api_key.id, req.query, req.collection_id, req.depth, req.temperature,
+        req.rerank, req.use_trusted_corpus, req.graph_hops, model,
+    )
+    if settings.quota_enabled:
+        query_cache.check_quota(_api_key.id, _api_key.is_pro, query_cache.count_tokens(req.query, model))
+    if use_cache:
+        hit = query_cache.get_cached(ckey)
+        if hit is not None:
+            try:
+                cached_resp = QueryResponse(**hit)
+            except Exception as exc:
+                log.warning("Cached payload invalid, treating as miss: %s", exc)
+                cached_resp = None
+            if cached_resp is not None:
+                if settings.quota_enabled:
+                    query_cache.consume_quota(_api_key.id, query_cache.count_tokens(req.query, model))
+                cached_resp.cached = True
+                return cached_resp
+
     contexts, ordered_contexts, response_metadata, collection_filter, confidence = await _retrieve(req, _api_key)
 
     if confidence == "insufficient":
@@ -89,7 +140,18 @@ async def query(req: QueryRequest, request: Request, _api_key: ApiKeyRecord = De
             log.warning("save_to_wiki failed: %s", exc)
 
     citations = [{"source": row.get("source_url") or str(row.get("document_id") or "unknown")} for row in contexts]
-    return QueryResponse(answer=answer, contexts=contexts, citations=citations, metadata=response_metadata)
+    response_metadata["citations_enforced"] = (
+        not contexts or confidence == "insufficient" or has_citation_markers(answer)
+    )
+    resp = QueryResponse(answer=answer, contexts=contexts, citations=citations, metadata=response_metadata)
+    if settings.quota_enabled:
+        query_cache.consume_quota(
+            _api_key.id,
+            query_cache.count_tokens(req.query, model) + query_cache.count_tokens(answer, model),
+        )
+    if use_cache:
+        query_cache.put_cached(ckey, resp.model_dump())
+    return resp
 
 
 async def _retrieve(
@@ -101,16 +163,15 @@ async def _retrieve(
     except ValueError as exc:
         raise HTTPException(400, "collection_id must be a UUID") from exc
 
-    # Determine graph hops and mode based on cognitive depth and request params
+    # Determine graph hops: explicit override wins; depths 1-2 never traverse;
+    # otherwise follow the intent router. Graph expansion is off by default and
+    # only engages on detected lineage/dependency intent or manual override.
     if req.graph_hops is not None:
         effective_hops = req.graph_hops
         graph_mode = "manual"
     elif req.depth in (1, 2):
         effective_hops = 0
         graph_mode = "concept_direct"
-    elif req.depth == 5:
-        effective_hops = 2
-        graph_mode = "auto"
     else:
         effective_hops = detect_graph_hops(req.query)
         graph_mode = "auto"
@@ -183,48 +244,87 @@ async def _retrieve(
                     for c in vault_matches
                 ]
 
-    # Cognitive Depth 3-5 or fallback when Depths 1-2 found no concept notes
+    # Cognitive Depth 3-5 or fallback when Depths 1-2 found no concept notes.
+    # Dense-first: a strong dense hit is used directly; the hybrid (dense +
+    # lexical RRF) functions serve as the backstop when dense coverage is weak.
+    retrieval_mode = "concept" if contexts else "dense"
     if not contexts:
-        params: dict[str, Any] = {
-            "query_text": req.query,
-            "query_embedding": (await embed_texts([req.query]))[0],
+        settings = get_settings()
+        query_embedding = (await embed_texts([req.query]))[0]
+        base_params: dict[str, Any] = {
             "collection_filter": collection_filter,
             "api_key_filter": UUID(_api_key.id),
         }
-        if effective_hops > 0:
-            params["graph_hops"] = effective_hops
-            rpc_fn = "hybrid_search_trusted_with_graph_v5" if req.use_trusted_corpus else "hybrid_search_with_graph_v5"
-        else:
-            rpc_fn = "hybrid_search_trusted_v5" if req.use_trusted_corpus else "hybrid_search_v5"
+        dense_contexts: list[dict[str, Any]] = []
+        if effective_hops == 0:
+            try:
+                dense_contexts = await execute_rpc(
+                    "dense_search_v5", {"query_embedding": query_embedding, **base_params}
+                )
+                for row in dense_contexts:
+                    row["match_source"] = "dense"
+            except Exception as exc:
+                log.warning("Dense search failed, falling back to hybrid: %s", exc)
+                dense_contexts = []
 
-        try:
-            raw_contexts = await execute_rpc(rpc_fn, params)
-            if req.depth in (1, 2):
-                contexts = raw_contexts[:(1 if req.depth == 1 else 2)]
-            else:
-                contexts = raw_contexts
-        except Exception:
+        dense_top = max((c.get("score", 0.0) or 0.0) for c in dense_contexts) if dense_contexts else 0.0
+        if len(dense_contexts) >= settings.dense_hit_min_results and dense_top >= settings.dense_hit_min_similarity:
+            contexts = dense_contexts[:10]
+            retrieval_mode = "dense"
+        else:
+            retrieval_mode = "hybrid"
+            params: dict[str, Any] = {
+                "query_text": req.query,
+                "query_embedding": query_embedding,
+                **base_params,
+            }
             if effective_hops > 0:
-                fallback_params = {k: v for k, v in params.items() if k != "graph_hops"}
-                fallback_fn = "hybrid_search_trusted_v5" if req.use_trusted_corpus else "hybrid_search_v5"
-                try:
-                    contexts = await execute_rpc(fallback_fn, fallback_params)
-                except Exception as exc:
-                    raise HTTPException(503, "PostgreSQL retrieval is unavailable") from exc
+                params["graph_hops"] = effective_hops
+                rpc_fn = "hybrid_search_trusted_with_graph_v5" if req.use_trusted_corpus else "hybrid_search_with_graph_v5"
             else:
-                raise HTTPException(503, "PostgreSQL retrieval is unavailable")
+                rpc_fn = "hybrid_search_trusted_v5" if req.use_trusted_corpus else "hybrid_search_v5"
 
-    # Reranking: Depths 1-2 bypass reranking for sub-200ms latency.
-    # Depth 5 forces cross-encoder rerank; Depth 3-4 respects req.rerank.
-    should_rerank = (req.depth == 5) or (req.rerank and req.depth >= 3)
+            try:
+                raw_contexts = await execute_rpc(rpc_fn, params)
+                if req.depth in (1, 2):
+                    contexts = raw_contexts[:(1 if req.depth == 1 else 2)]
+                else:
+                    contexts = raw_contexts
+            except Exception:
+                if effective_hops > 0:
+                    fallback_params = {k: v for k, v in params.items() if k != "graph_hops"}
+                    fallback_fn = "hybrid_search_trusted_v5" if req.use_trusted_corpus else "hybrid_search_v5"
+                    try:
+                        contexts = await execute_rpc(fallback_fn, fallback_params)
+                    except Exception as exc:
+                        raise HTTPException(503, "PostgreSQL retrieval is unavailable") from exc
+                else:
+                    raise HTTPException(503, "PostgreSQL retrieval is unavailable")
+
+    # Reranking: depths 1-2 skip for sub-200ms latency; depth 5 forces it.
+    # At depths 3-4 the cross-encoder runs as a rescue for marginal hits, but
+    # is skipped on clear dense matches (cosine similarity at/above the
+    # threshold) where reordering adds latency without measurable gain.
+    def _clear_dense_hit(rows: list[dict[str, Any]]) -> bool:
+        threshold = get_settings().rerank_skip_similarity
+        return any(
+            (row.get("match_source") == "dense") and ((row.get("score") or 0.0) >= threshold)
+            for row in rows
+        )
+
+    rerank_applied = False
+    should_rerank = (req.depth == 5) or (req.rerank and req.depth >= 3 and not _clear_dense_hit(contexts))
     if should_rerank and contexts:
         try:
             top_n = 7 if req.depth == 5 else 5
             contexts = await get_reranker_service().rerank(req.query, contexts, top_n=top_n)
+            rerank_applied = True
         except Exception as exc:
             log.warning("Rerank failed, using retrieval order: %s", exc)
 
-    # Evaluate retrieval confidence (CRAG - Corrective RAG gating)
+    # Evaluate retrieval confidence (CRAG - Corrective RAG gating).
+    # Score scales differ by source: cosine similarity (~0.3-1.0) for dense
+    # hits versus small RRF-fused scores for hybrid/graph retrieval.
     confidence = "high"
     if not contexts:
         confidence = "insufficient"
@@ -240,11 +340,7 @@ async def _retrieve(
                 elif max_score < 0.0:
                     confidence = "medium"
             elif any("score" in c for c in contexts):
-                max_score = max(c.get("score", 0.0) for c in contexts)
-                if max_score < 0.012:
-                    confidence = "low"
-                elif max_score < 0.020:
-                    confidence = "medium"
+                confidence = _confidence_from_scores(contexts)
     elif any("rerank_score" in c for c in contexts):
         max_score = max(c.get("rerank_score", -999.0) for c in contexts)
         if max_score < -2.0:
@@ -252,11 +348,7 @@ async def _retrieve(
         elif max_score < 0.0:
             confidence = "medium"
     elif any("score" in c for c in contexts):
-        max_score = max(c.get("score", 0.0) for c in contexts)
-        if max_score < 0.012:
-            confidence = "low"
-        elif max_score < 0.020:
-            confidence = "medium"
+        confidence = _confidence_from_scores(contexts)
 
     # Apply Lost-in-the-Middle U-shaped ordering before prompt synthesis
     ordered_contexts = reorder_lost_in_the_middle(contexts)
@@ -269,12 +361,56 @@ async def _retrieve(
         "graph_mode": graph_mode,
         "confidence": confidence,
         "prompt_ordering": "lost_in_the_middle",
+        "retrieval_mode": retrieval_mode,
+        "rerank_applied": rerank_applied,
     }
     return contexts, ordered_contexts, response_metadata, collection_filter, confidence
 
 
 @router.post("/query/stream")
 async def query_stream(req: QueryRequest, request: Request, _api_key: ApiKeyRecord = Depends(verify_api_key)) -> StreamingResponse:
+    settings = get_settings()
+    model = settings.llm_model
+    use_cache = not req.bypass_cache and not req.save_to_wiki
+    ckey = query_cache.cache_key(
+        _api_key.id, req.query, req.collection_id, req.depth, req.temperature,
+        req.rerank, req.use_trusted_corpus, req.graph_hops, model,
+    )
+    if settings.quota_enabled:
+        query_cache.check_quota(_api_key.id, _api_key.is_pro, query_cache.count_tokens(req.query, model))
+    replay: QueryResponse | None = None
+    if use_cache:
+        hit = query_cache.get_cached(ckey)
+        if hit is not None:
+            try:
+                replay = QueryResponse(**hit)
+                replay.cached = True
+            except Exception as exc:
+                log.warning("Cached payload invalid, treating as miss: %s", exc)
+                replay = None
+        if replay is not None:
+            if settings.quota_enabled:
+                query_cache.consume_quota(_api_key.id, query_cache.count_tokens(req.query, model))
+
+            async def replay_events():
+                yield ": stream start\n\n"
+                yield f"data: {json.dumps({'delta': replay.answer}, default=str)}\n\n"
+                final = {
+                    "answer": replay.answer,
+                    "contexts": replay.contexts,
+                    "citations": replay.citations,
+                    "cached": True,
+                    "metadata": replay.metadata,
+                }
+                yield f"data: {json.dumps(final, default=str)}\n\n"
+                yield "data: [DONE]\n\n"
+
+            return StreamingResponse(
+                replay_events(),
+                media_type="text/event-stream",
+                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+            )
+
     contexts, ordered_contexts, response_metadata, collection_filter, confidence = await _retrieve(req, _api_key)
     citations = [{"source": row.get("source_url") or str(row.get("document_id") or "unknown")} for row in contexts]
 
@@ -283,6 +419,7 @@ async def events():
         yield ": stream start\n\n"
         if confidence == "insufficient":
             answer = "I could not find sufficient matching documentation in your collection to answer this query reliably."
+            response_metadata["citations_enforced"] = True
             yield f"data: {json.dumps({'delta': answer}, default=str)}\n\n"
             final = {"answer": answer, "contexts": contexts, "citations": citations, "metadata": response_metadata}
             yield f"data: {json.dumps(final, default=str)}\n\n"
@@ -294,6 +431,11 @@ async def events():
                 if await request.is_disconnected():
                     break
             answer = "".join(parts)
+            # No retry on the streaming path (it would double latency and repeat
+            # deltas); enforcement retry lives on POST /api/query. Flag it here.
+            response_metadata["citations_enforced"] = (
+                not contexts or confidence == "insufficient" or has_citation_markers(answer)
+            )
             if req.save_to_wiki and answer:
                 try:
                     ref_concepts = [c.get("concept_name") for c in contexts if c.get("concept_name")]
@@ -307,8 +449,14 @@ async def events():
                     log.warning("save_to_wiki failed: %s", exc)
             final = {"answer": answer, "contexts": contexts, "citations": citations, "metadata": response_metadata}
             yield f"data: {json.dumps(final, default=str)}\n\n"
+        if settings.quota_enabled:
+            query_cache.consume_quota(
+                _api_key.id,
+                query_cache.count_tokens(req.query, model) + query_cache.count_tokens(answer, model),
+            )
+        if use_cache:
+            query_cache.put_cached(ckey, final)
         yield "data: [DONE]\n\n"
-
     return StreamingResponse(
         events(),
         media_type="text/event-stream",
diff --git a/api/services/cache.py b/api/services/cache.py
new file mode 100644
index 0000000..bbfc22f
--- /dev/null
+++ b/api/services/cache.py
@@ -0,0 +1,144 @@
+"""Query-result cache and per-key token quotas backed by Redis.
+
+Both features fail open: if Redis is unreachable, queries run uncached and
+quotas are not enforced (with a loud warning). Availability first; enforcement
+is best-effort until Redis becomes a hard deployment dependency.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import HTTPException
+
+from api.config import get_settings
+
+log = logging.getLogger(__name__)
+
+CACHE_VERSION = 1
+
+_client = None
+
+
+def get_client():
+    """Lazy Redis singleton; None when Redis is unreachable. Monkeypatchable."""
+    global _client
+    if _client is not None:
+        return _client
+    try:
+        import redis
+
+        client = redis.Redis.from_url(
+            get_settings().redis_url,
+            socket_connect_timeout=2,
+            socket_timeout=2,
+            decode_responses=True,
+        )
+        client.ping()
+        _client = client
+        return _client
+    except Exception as exc:
+        log.warning("Redis unavailable, running without cache/quotas: %s", exc)
+        return None
+
+
+def reset_client() -> None:
+    """Drop the cached client (tests)."""
+    global _client
+    _client = None
+
+
+def cache_key(api_key_id: str, query: str, collection_id: str | None, depth: int,
+              temperature: float, rerank: bool, use_trusted: bool,
+              graph_hops: int | None, llm_model: str) -> str:
+    material = "|".join([
+        f"v{CACHE_VERSION}", api_key_id, query.strip(),
+        collection_id or "", str(depth), f"{temperature:.2f}",
+        str(rerank), str(use_trusted), str(graph_hops), llm_model,
+    ])
+    return "depthapi:q:" + hashlib.sha256(material.encode("utf-8")).hexdigest()
+
+
+def get_cached(key: str) -> dict[str, Any] | None:
+    client = get_client()
+    if client is None:
+        return None
+    try:
+        raw = client.get(key)
+        if not raw:
+            return None
+        payload = json.loads(raw)
+        if not isinstance(payload, dict) or "answer" not in payload or "contexts" not in payload:
+            return None
+        return payload
+    except Exception as exc:
+        log.warning("Cache read failed, treating as miss: %s", exc)
+        return None
+
+
+def put_cached(key: str, payload: dict[str, Any]) -> None:
+    client = get_client()
+    if client is None:
+        return
+    try:
+        client.set(key, json.dumps(payload, default=str), ex=get_settings().query_cache_ttl_seconds)
+    except Exception as exc:
+        log.warning("Cache write failed: %s", exc)
+
+
+def quota_limit(is_pro: bool) -> int:
+    settings = get_settings()
+    return settings.pro_daily_token_quota if is_pro else settings.daily_token_quota_per_user
+
+
+def count_tokens(text: str, model: str) -> int:
+    try:
+        import tiktoken
+
+        try:
+            enc = tiktoken.encoding_for_model(model)
+        except Exception:
+            enc = tiktoken.get_encoding("cl100k_base")
+        return len(enc.encode(text or ""))
+    except Exception as exc:
+        log.warning("Token counting failed, assuming 0: %s", exc)
+        return 0
+
+
+def _quota_redis_key(api_key_id: str) -> str:
+    day = datetime.now(timezone.utc).strftime("%Y%m%d")
+    return f"depthapi:quota:{api_key_id}:{day}"
+
+
+def check_quota(api_key_id: str, is_pro: bool, estimated_tokens: int) -> None:
+    """Raise 429 when the key would exceed its daily token budget. Fail-open."""
+    limit = quota_limit(is_pro)
+    if limit <= 0:
+        return
+    client = get_client()
+    if client is None:
+        return
+    try:
+        used = int(client.get(_quota_redis_key(api_key_id)) or 0)
+    except Exception as exc:
+        log.warning("Quota read failed, allowing request: %s", exc)
+        return
+    if used + estimated_tokens > limit:
+        raise HTTPException(429, "Daily token quota exceeded")
+
+
+def consume_quota(api_key_id: str, tokens: int) -> None:
+    if tokens <= 0:
+        return
+    client = get_client()
+    if client is None:
+        return
+    try:
+        key = _quota_redis_key(api_key_id)
+        client.incrby(key, tokens)
+        client.expire(key, 172800)
+    except Exception as exc:
+        log.warning("Quota accounting failed: %s", exc)
diff --git a/api/services/inference/inference.py b/api/services/inference/inference.py
index 23d456e..d793df7 100644
--- a/api/services/inference/inference.py
+++ b/api/services/inference/inference.py
@@ -1,9 +1,18 @@
 """Mode-free response generation for retrieved contexts."""
+import re
 from collections.abc import AsyncIterator
 from typing import Any
 
 from api.config import get_settings
 
+CITATION_PATTERN = re.compile(r"\[\d+\]")
+
+_ABSTENTION_MARKERS = (
+    "could not find sufficient",
+    "no matching knowledge",
+    "insufficient",
+)
+
 
 def _fallback_response(contexts: list[dict[str, Any]]) -> str:
     if not contexts:
@@ -11,7 +20,54 @@ def _fallback_response(contexts: list[dict[str, Any]]) -> str:
     excerpts = [str(context.get("content", "")).strip() for context in contexts]
     return "\n\n".join(excerpt for excerpt in excerpts if excerpt) or "No matching knowledge was found."
 
-async def generate_response(query: str, contexts: list[dict[str, Any]], temperature: float = 0.7) -> str:
+
+def has_citation_markers(answer: str) -> bool:
+    """True when the answer cites at least one numbered source like [1]."""
+    return CITATION_PATTERN.search(answer or "") is not None
+
+
+def looks_like_abstention(answer: str) -> bool:
+    """True for honest no-match answers, which carry no citations by design."""
+    lowered = (answer or "").lower()
+    return any(marker in lowered for marker in _ABSTENTION_MARKERS)
+
+
+def _numbered_sources(contexts: list[dict[str, Any]]) -> str:
+    chunks = []
+    for idx, item in enumerate(contexts, 1):
+        chunks.append(f"[{idx}] {str(item.get('content', ''))[:6000]}")
+    return "\n\n".join(chunks)
+
+
+def _citation_system_prompt() -> str:
+    return (
+        "Answer using only the supplied knowledge, which is numbered [1], [2], ... . "
+        "Cite every factual claim inline with its source number, e.g. [1]. "
+        "If the knowledge is insufficient, say so plainly without citations."
+    )
+
+
+async def _complete_once(
+    client: Any, model: str, temperature: float, query: str, source_text: str, nudge: str = ""
+) -> str | None:
+    user_content = f"Question: {query}\n\nKnowledge:\n{source_text}"
+    if nudge:
+        user_content += f"\n\n{nudge}"
+    response = await client.chat.completions.create(
+        model=model,
+        temperature=temperature,
+        messages=[
+            {"role": "system", "content": _citation_system_prompt()},
+            {"role": "user", "content": user_content},
+        ],
+    )
+    answer = response.choices[0].message.content if response.choices else None
+    return answer.strip() if answer else None
+
+
+async def generate_response(
+    query: str, contexts: list[dict[str, Any]], temperature: float = 0.7, enforce_citations: bool = True
+) -> str:
     settings = get_settings()
     api_key = settings.openai_api_key.get_secret_value()
     if not api_key or not contexts:
@@ -19,17 +75,26 @@ async def generate_response(query: str, contexts: list[dict[str, Any]], temperat
     try:
         from openai import AsyncOpenAI
         client = AsyncOpenAI(api_key=api_key, timeout=settings.llm_timeout_seconds)
-        source_text = "\n\n".join(str(item.get("content", ""))[:6000] for item in contexts)
-        response = await client.chat.completions.create(
-            model=settings.llm_model,
-            temperature=temperature,
-            messages=[
-                {"role": "system", "content": "Answer using only the supplied knowledge. If it is insufficient, say so."},
-                {"role": "user", "content": f"Question: {query}\n\nKnowledge:\n{source_text}"},
-            ],
-        )
-        answer = response.choices[0].message.content if response.choices else None
-        return answer.strip() if answer else _fallback_response(contexts)
+        source_text = _numbered_sources(contexts)
+        answer = await _complete_once(client, settings.llm_model, temperature, query, source_text)
+        if not answer:
+            return _fallback_response(contexts)
+        if (
+            enforce_citations
+            and not has_citation_markers(answer)
+            and not looks_like_abstention(answer)
+        ):
+            retried = await _complete_once(
+                client,
+                settings.llm_model,
+                temperature,
+                query,
+                source_text,
+                nudge="Your previous answer contained no citations like [1]. Answer again, citing every factual claim with its source number.",
+            )
+            if retried and (has_citation_markers(retried) or looks_like_abstention(retried)):
+                return retried
+        return answer
     except Exception:
         return _fallback_response(contexts)
 
@@ -47,13 +112,13 @@ async def generate_stream_response(query: str, contexts: list[dict[str, Any]], t
     try:
         from openai import AsyncOpenAI
         client = AsyncOpenAI(api_key=api_key, timeout=settings.llm_timeout_seconds)
-        source_text = "\n\n".join(str(item.get("content", ""))[:6000] for item in contexts)
+        source_text = _numbered_sources(contexts)
         stream = await client.chat.completions.create(
             model=settings.llm_model,
             temperature=temperature,
             stream=True,
             messages=[
-                {"role": "system", "content": "Answer using only the supplied knowledge. If it is insufficient, say so."},
+                {"role": "system", "content": _citation_system_prompt()},
                 {"role": "user", "content": f"Question: {query}\n\nKnowledge:\n{source_text}"},
             ],
         )
diff --git a/crates/depth_engine/src/retrieval/crag.rs b/crates/depth_engine/src/retrieval/crag.rs
index d516a2b..955f108 100644
--- a/crates/depth_engine/src/retrieval/crag.rs
+++ b/crates/depth_engine/src/retrieval/crag.rs
@@ -54,6 +54,11 @@ pub fn evaluate_confidence(scores: &[f64], is_reranked: bool) -> ConfidenceEvalu
 }
 
 /// Helper to evaluate confidence directly from a vector of context dictionaries.
+///
+/// Score scales differ by retrieval source: dense cosine similarity
+/// (~0.3-1.0, tagged with `"match_source": "dense"`) versus small RRF-fused
+/// scores for hybrid/graph retrieval. Each scale uses its own thresholds so
+/// a strong dense hit is not misread as a weak fused one (or vice versa).
 pub fn evaluate_contexts_confidence(contexts: &[serde_json::Value]) -> ConfidenceEvaluation {
     if contexts.is_empty() {
         return ConfidenceEvaluation {
@@ -72,10 +77,31 @@ pub fn evaluate_contexts_confidence(contexts: &[serde_json::Value]) -> Confidenc
             .collect();
         evaluate_confidence(&scores, true)
     } else {
+        let is_dense = contexts.iter().any(|c| {
+            c.get("match_source").and_then(|v| v.as_str()) == Some("dense")
+        });
         let scores: Vec<f64> = contexts
             .iter()
             .filter_map(|c| c.get("score").and_then(|v| v.as_f64()))
             .collect();
+        if is_dense {
+            if scores.is_empty() {
+                return evaluate_confidence(&[], false);
+            }
+            let max_score = scores.iter().cloned().fold(f64::NEG_INFINITY, f64::max);
+            let tier = if max_score < 0.55 {
+                "low"
+            } else if max_score < 0.7 {
+                "medium"
+            } else {
+                "high"
+            };
+            return ConfidenceEvaluation {
+                confidence: tier.to_string(),
+                is_insufficient: false,
+                max_score: Some(max_score),
+            };
+        }
         evaluate_confidence(&scores, false)
     }
 }
@@ -117,4 +143,26 @@ mod tests {
         let low = evaluate_confidence(&[0.005, 0.002], false);
         assert_eq!(low.confidence, "low");
     }
+
+    fn dense_context(score: f64) -> serde_json::Value {
+        serde_json::json!({"content": "x", "score": score, "match_source": "dense"})
+    }
+
+    #[test]
+    fn test_crag_dense_similarity_scale() {
+        let high = evaluate_contexts_confidence(&[dense_context(0.9), dense_context(0.8)]);
+        assert_eq!(high.confidence, "high");
+
+        let medium = evaluate_contexts_confidence(&[dense_context(0.6)]);
+        assert_eq!(medium.confidence, "medium");
+
+        let low = evaluate_contexts_confidence(&[dense_context(0.4)]);
+        assert_eq!(low.confidence, "low");
+    }
+
+    #[test]
+    fn test_crag_fused_scale_unchanged() {
+        let fused = serde_json::json!({"content": "x", "score": 0.03});
+        assert_eq!(evaluate_contexts_confidence(&[fused]).confidence, "high");
+    }
 }
diff --git a/db/migrations/004_dense_search.sql b/db/migrations/004_dense_search.sql
new file mode 100644
index 0000000..1ffa09e
--- /dev/null
+++ b/db/migrations/004_dense_search.sql
@@ -0,0 +1,23 @@
+-- Migration 004: dense-only retrieval for dense-first strategy.
+-- hybrid_search_v5 fuses dense+lexical in one RRF step, which dilutes a
+-- strong dense signal with a weak lexical one. The application now tries
+-- a dense-only lookup first and falls back to the hybrid functions only
+-- when dense coverage is weak. Score is cosine similarity (1 - distance),
+-- so higher is better, matching the lexical ts_rank direction.
+CREATE OR REPLACE FUNCTION dense_search_v5(
+    query_embedding vector(768),
+    collection_filter uuid DEFAULT NULL,
+    api_key_filter uuid DEFAULT NULL
+) RETURNS TABLE(content text, document_id uuid, source_url text, score real) LANGUAGE sql STABLE AS $$
+  SELECT c.content, c.document_id, d.source_url,
+         (1.0 - (c.embedding <=> query_embedding))::real AS score
+  FROM knowledge_chunks c
+  JOIN knowledge_documents d ON d.id = c.document_id
+  JOIN knowledge_collections k ON k.id = d.collection_id
+  WHERE api_key_filter IS NOT NULL
+    AND k.api_key_id = api_key_filter
+    AND (collection_filter IS NULL OR d.collection_id = collection_filter)
+    AND c.embedding IS NOT NULL
+  ORDER BY c.embedding <=> query_embedding ASC
+  LIMIT 10
+$$;
diff --git a/docker-compose.yml b/docker-compose.yml
index 5da9af3..d71ec1a 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -11,6 +11,7 @@ services:
       - ./db/migrations/001_schema.sql:/docker-entrypoint-initdb.d/001_schema.sql:ro
       - ./db/migrations/002_concepts_graph.sql:/docker-entrypoint-initdb.d/002_concepts_graph.sql:ro
       - ./db/migrations/003_fixes.sql:/docker-entrypoint-initdb.d/003_fixes.sql:ro
+      - ./db/migrations/004_dense_search.sql:/docker-entrypoint-initdb.d/004_dense_search.sql:ro
       - ./db/seed/001_dev_api_key.sql:/docker-entrypoint-initdb.d/004_dev_api_key.sql:ro
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U depthapi"]
diff --git a/scripts/validate_rag_local.sh b/scripts/validate_rag_local.sh
index b30b117..a916269 100755
--- a/scripts/validate_rag_local.sh
+++ b/scripts/validate_rag_local.sh
@@ -20,6 +20,7 @@ else
       -v "$ROOT_DIR/db/migrations/001_schema.sql:/docker-entrypoint-initdb.d/001_schema.sql:ro" \
       -v "$ROOT_DIR/db/migrations/002_concepts_graph.sql:/docker-entrypoint-initdb.d/002_concepts_graph.sql:ro" \
       -v "$ROOT_DIR/db/migrations/003_fixes.sql:/docker-entrypoint-initdb.d/003_fixes.sql:ro" \
+      -v "$ROOT_DIR/db/migrations/004_dense_search.sql:/docker-entrypoint-initdb.d/004_dense_search.sql:ro" \
       -v "$ROOT_DIR/db/seed/001_dev_api_key.sql:/docker-entrypoint-initdb.d/004_dev_api_key.sql:ro" \
       pgvector/pgvector:pg17
   else
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..8c0aee2
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,43 @@
+"""Shared fixtures: keep query tests hermetic (no real Redis)."""
+from __future__ import annotations
+
+import pytest
+
+from api.services import cache as query_cache
+
+
+class FakeRedis:
+    """Minimal in-memory stand-in for the Redis surface query_cache uses."""
+
+    def __init__(self):
+        self.store: dict[str, str] = {}
+
+    def ping(self):
+        return True
+
+    def get(self, key: str):
+        return self.store.get(key)
+
+    def set(self, key: str, value: str, ex: int | None = None):
+        self.store[key] = value
+        return True
+
+    def setex(self, key: str, ttl: int, value: str):
+        self.store[key] = value
+        return True
+
+    def incrby(self, key: str, amount: int):
+        self.store[key] = str(int(self.store.get(key, "0")) + amount)
+        return int(self.store[key])
+
+    def expire(self, key: str, ttl: int):
+        return True
+
+
+@pytest.fixture(autouse=True)
+def _isolated_query_cache(monkeypatch):
+    """Every test gets a fresh fake Redis; prod code paths unchanged."""
+    fake = FakeRedis()
+    monkeypatch.setattr(query_cache, "get_client", lambda: fake)
+    query_cache.reset_client()
+    return fake
diff --git a/tests/unit/test_citation_enforcement.py b/tests/unit/test_citation_enforcement.py
new file mode 100644
index 0000000..9e7af04
--- /dev/null
+++ b/tests/unit/test_citation_enforcement.py
@@ -0,0 +1,104 @@
+"""Unit tests for citation enforcement in response generation."""
+from __future__ import annotations
+
+import sys
+import types
+
+import pytest
+from pydantic import SecretStr
+
+from api.services.inference import inference as inference_module
+
+
+class _StubSettings:
+    openai_api_key = SecretStr("test-key")
+    llm_model = "test-model"
+    llm_timeout_seconds = 60
+
+
+def _install_openai_stub(monkeypatch, answers: list[str | None]):
+    calls: list[dict] = []
+
+    class _Message:
+        def __init__(self, content):
+            self.content = content
+
+    class _Choice:
+        def __init__(self, content):
+            self.message = _Message(content)
+
+    class _Response:
+        def __init__(self, content):
+            self.choices = [_Choice(content)] if content is not None else []
+
+    class _Completions:
+        async def create(self, **kwargs):
+            calls.append(kwargs)
+            return _Response(answers[min(len(calls) - 1, len(answers) - 1)])
+
+    class _Chat:
+        completions = _Completions()
+
+    class AsyncOpenAI:
+        def __init__(self, *args, **kwargs):
+            self.chat = _Chat()
+
+    stub = types.ModuleType("openai")
+    stub.AsyncOpenAI = AsyncOpenAI
+    monkeypatch.setitem(sys.modules, "openai", stub)
+    monkeypatch.setattr(inference_module, "get_settings", lambda: _StubSettings())
+    return calls
+
+
+def test_has_citation_markers():
+    assert inference_module.has_citation_markers("Paris is the capital [1].") is True
+    assert inference_module.has_citation_markers("No markers here.") is False
+    assert inference_module.has_citation_markers("") is False
+
+
+def test_looks_like_abstention():
+    assert inference_module.looks_like_abstention("I could not find sufficient documentation.") is True
+    assert inference_module.looks_like_abstention("No matching knowledge was found.") is True
+    assert inference_module.looks_like_abstention("Paris is the capital [1].") is False
+
+
+@pytest.mark.asyncio
+async def test_generate_response_retries_missing_citations(monkeypatch):
+    calls = _install_openai_stub(monkeypatch, ["Paris is the capital.", "Paris is the capital [1]."])
+    contexts = [{"content": "Paris is the capital of France."}]
+
+    answer = await inference_module.generate_response("Capital?", contexts)
+
+    assert answer == "Paris is the capital [1]."
+    assert len(calls) == 2
+    assert "[1] Paris" in calls[0]["messages"][1]["content"]
+
+
+@pytest.mark.asyncio
+async def test_generate_response_no_retry_when_cited(monkeypatch):
+    calls = _install_openai_stub(monkeypatch, ["Paris is the capital [1]."])
+
+    answer = await inference_module.generate_response("Capital?", [{"content": "Paris."}])
+
+    assert answer == "Paris is the capital [1]."
+    assert len(calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_generate_response_no_retry_on_abstention(monkeypatch):
+    calls = _install_openai_stub(monkeypatch, ["I could not find sufficient documentation."])
+
+    answer = await inference_module.generate_response("Capital?", [{"content": "Paris."}])
+
+    assert "could not find sufficient" in answer
+    assert len(calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_generate_response_keeps_first_answer_if_retry_still_uncited(monkeypatch):
+    calls = _install_openai_stub(monkeypatch, ["First attempt.", "Second attempt."])
+
+    answer = await inference_module.generate_response("Capital?", [{"content": "Paris."}])
+
+    assert answer == "First attempt."
+    assert len(calls) == 2
diff --git a/tests/unit/test_cognitive_depth.py b/tests/unit/test_cognitive_depth.py
index d590f9e..e3aa64f 100644
--- a/tests/unit/test_cognitive_depth.py
+++ b/tests/unit/test_cognitive_depth.py
@@ -87,7 +87,7 @@ async def fake_generate(q, ctxs, temp):
 
 @pytest.mark.asyncio
 async def test_cognitive_depth_5_deep_graph_and_forced_rerank(monkeypatch):
-    """Depth 5 enforces 2-hop graph traversal and cross-encoder rerank."""
+    """Depth 5 with manual graph_hops=2 traverses the graph and forces rerank."""
     rpc_params = None
     rerank_executed = False
 
@@ -114,16 +114,49 @@ async def rerank(self, query: str, candidates: list, top_n: int = 7):
     monkeypatch.setattr(query_module, "generate_response", AsyncMock(return_value="Deep answer"))
 
     # Even with rerank=False in request, depth=5 forces rerank
-    req = query_module.QueryRequest(query="Complete system architecture deep dive", depth=5, rerank=False)
+    req = query_module.QueryRequest(
+        query="Complete system architecture deep dive", depth=5, rerank=False, graph_hops=2
+    )
     res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "pro", True))
 
     assert res.metadata["cognitive_depth"] == 5
     assert res.metadata["graph_hops"] == 2
+    assert res.metadata["graph_mode"] == "manual"
     assert rpc_params.get("graph_hops") == 2
     assert rerank_executed is True
     assert res.metadata["prompt_ordering"] == "lost_in_the_middle"
 
 
+@pytest.mark.asyncio
+async def test_cognitive_depth_5_graph_off_without_intent(monkeypatch):
+    """Depth 5 no longer forces graph traversal; intent-free queries stay flat."""
+    rpc_fns: list[str] = []
+
+    async def fake_rpc(fn_name, params):
+        rpc_fns.append(fn_name)
+        return [{"content": "Flat context", "document_id": str(uuid4()), "score": 0.05}]
+
+    async def fake_embed(texts):
+        return ["[" + ",".join(["0"] * 768) + "]"]
+
+    class FakeReranker:
+        async def rerank(self, query: str, candidates: list, top_n: int = 7):
+            return candidates[:top_n]
+
+    monkeypatch.setattr(query_module, "execute_rpc", fake_rpc)
+    monkeypatch.setattr(query_module, "embed_texts", fake_embed)
+    monkeypatch.setattr(query_module, "get_reranker_service", lambda: FakeReranker())
+    monkeypatch.setattr(query_module, "generate_response", AsyncMock(return_value="Deep answer"))
+
+    req = query_module.QueryRequest(query="Explain quantum mechanics", depth=5, rerank=False)
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "pro", True))
+
+    assert res.metadata["graph_hops"] == 0
+    assert res.metadata["graph_mode"] == "auto"
+    assert "dense_search_v5" in rpc_fns
+    assert not any("graph" in fn for fn in rpc_fns)
+
+
 @pytest.mark.asyncio
 async def test_compounding_qa_save_to_wiki(monkeypatch, isolated_vault):
     """save_to_wiki=True writes synthesized insight to the vault and log."""
diff --git a/tests/unit/test_dense_first.py b/tests/unit/test_dense_first.py
new file mode 100644
index 0000000..9e4b9c7
--- /dev/null
+++ b/tests/unit/test_dense_first.py
@@ -0,0 +1,134 @@
+"""Unit tests for dense-first retrieval with hybrid fallback and rerank gating."""
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from starlette.requests import Request
+
+from api.routers import query as query_module
+from api.services.security.api_key_auth import ApiKeyRecord
+
+
+def _dummy_request() -> Request:
+    return Request({"type": "http", "method": "POST", "url": "http://testserver/api/query", "headers": []})
+
+
+def _dense(rows):
+    return [{"content": f"Doc {i}", "document_id": str(uuid4()), "score": s} for i, s in enumerate(rows)]
+
+
+class _FakeReranker:
+    called = False
+
+    async def rerank(self, query: str, candidates: list, top_n: int = 5):
+        _FakeReranker.called = True
+        return candidates[:top_n]
+
+
+def _run(monkeypatch, dense_rows, hybrid_rows=None, exc_dense=False, **req_kwargs):
+    calls: list[str] = []
+    _FakeReranker.called = False
+
+    async def fake_rpc(fn_name, params):
+        calls.append(fn_name)
+        if fn_name == "dense_search_v5":
+            if exc_dense:
+                raise RuntimeError("dense down")
+            return _dense(dense_rows)
+        return hybrid_rows if hybrid_rows is not None else []
+
+    async def fake_embed(texts):
+        return ["[" + ",".join(["0"] * 768) + "]"]
+
+    monkeypatch.setattr(query_module, "execute_rpc", fake_rpc)
+    monkeypatch.setattr(query_module, "embed_texts", fake_embed)
+    monkeypatch.setattr(query_module, "get_reranker_service", lambda: _FakeReranker())
+
+    req = query_module.QueryRequest(query="What is DepthAPI?", **req_kwargs)
+    return req, calls
+
+
+@pytest.mark.asyncio
+async def test_dense_hit_skips_hybrid_and_rerank(monkeypatch):
+    req, calls = _run(monkeypatch, [0.9, 0.88, 0.85, 0.82, 0.8, 0.78], hybrid_rows=[])
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False))
+
+    assert calls == ["dense_search_v5"]
+    assert res.metadata["retrieval_mode"] == "dense"
+    assert res.metadata["rerank_applied"] is False
+    assert _FakeReranker.called is False
+    assert res.metadata["confidence"] == "high"
+    assert len(res.contexts) == 6
+
+
+@pytest.mark.asyncio
+async def test_dense_miss_falls_back_to_hybrid(monkeypatch):
+    hybrid = [{"content": "Hybrid doc", "document_id": str(uuid4()), "score": 0.03}]
+    req, calls = _run(monkeypatch, [0.4, 0.35], hybrid_rows=hybrid)
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False))
+
+    assert calls[0] == "dense_search_v5"
+    assert "hybrid_search_trusted_v5" in calls
+    assert res.metadata["retrieval_mode"] == "hybrid"
+    assert res.contexts[0]["content"] == "Hybrid doc"
+
+
+@pytest.mark.asyncio
+async def test_dense_error_falls_back_to_hybrid(monkeypatch):
+    hybrid = [{"content": "Hybrid doc", "document_id": str(uuid4()), "score": 0.03}]
+    req, calls = _run(monkeypatch, [], hybrid_rows=hybrid, exc_dense=True)
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False))
+
+    assert "hybrid_search_trusted_v5" in calls
+    assert res.metadata["retrieval_mode"] == "hybrid"
+
+
+@pytest.mark.asyncio
+async def test_marginal_dense_hit_still_reranks(monkeypatch):
+    # Top similarity 0.6 clears the dense-hit bar (0.5) but not the rerank-skip bar (0.8).
+    req, _ = _run(monkeypatch, [0.6, 0.58, 0.55, 0.52, 0.51, 0.5])
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False))
+
+    assert res.metadata["retrieval_mode"] == "dense"
+    assert res.metadata["rerank_applied"] is True
+    assert _FakeReranker.called is True
+
+
+@pytest.mark.asyncio
+async def test_dense_low_similarity_confidence(monkeypatch):
+    req, _ = _run(monkeypatch, [0.52, 0.51, 0.5, 0.5, 0.5, 0.5])
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False))
+
+    assert res.metadata["retrieval_mode"] == "dense"
+    assert res.metadata["confidence"] == "low"
+
+
+@pytest.mark.asyncio
+async def test_citations_enforced_flag_on_fallback_answer(monkeypatch):
+    req, _ = _run(monkeypatch, [0.9, 0.88, 0.85, 0.82, 0.8, 0.78])
+    res = await query_module.query(req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False))
+
+    # Fallback excerpt answers carry no [n] markers.
+    assert res.metadata["citations_enforced"] is False
+
+
+@pytest.mark.asyncio
+async def test_query_bypasses_everything_without_contexts(monkeypatch):
+    async def fake_rpc(fn_name, params):
+        return []
+
+    async def fake_embed(texts):
+        return ["[0]"]
+
+    monkeypatch.setattr(query_module, "execute_rpc", fake_rpc)
+    monkeypatch.setattr(query_module, "embed_texts", fake_embed)
+
+    req = query_module.QueryRequest(query="Nothing matches this?", rerank=False)
+    res = await query_module.query(
+        req, _dummy_request(), ApiKeyRecord(str(uuid4()), "free", False)
+    )
+
+    assert res.metadata["confidence"] == "insufficient"
+    assert res.metadata["citations_enforced"] is True
+    assert res.cached is False
diff --git a/tests/unit/test_query_cache.py b/tests/unit/test_query_cache.py
new file mode 100644
index 0000000..9a2d942
--- /dev/null
+++ b/tests/unit/test_query_cache.py
@@ -0,0 +1,136 @@
+"""Unit tests for Redis query cache and per-key quotas (FakeRedis-backed)."""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from api.routers import query as query_module
+from api.routers.query import QueryRequest, query, query_stream
+from api.services import cache as cache_mod
+from api.services.security.api_key_auth import ApiKeyRecord
+
+
+def _key():
+    return ApiKeyRecord(str(uuid4()), "free", False)
+
+
+def _mocks(monkeypatch, answer="Mocked answer."):
+    mock_rpc = AsyncMock(return_value=[{"document_id": "d1", "content": "Doc content", "score": 0.03}])
+    mock_embed = AsyncMock(return_value=[[0.01] * 768])
+    mock_gen = AsyncMock(return_value=answer)
+    monkeypatch.setattr(query_module, "execute_rpc", mock_rpc)
+    monkeypatch.setattr(query_module, "embed_texts", mock_embed)
+    monkeypatch.setattr(query_module, "generate_response", mock_gen)
+    return mock_rpc, mock_gen
+
+
+@pytest.mark.asyncio
+async def test_cache_miss_then_hit(monkeypatch, _isolated_query_cache):
+    mock_rpc, _ = _mocks(monkeypatch)
+    req = QueryRequest(query="Cache me?")
+    key = _key()
+
+    first = await query(req, AsyncMock(), key)
+    assert first.cached is False
+    assert mock_rpc.await_count >= 1
+
+    mock_rpc.reset_mock()
+    mock_rpc.side_effect = AssertionError("RPC must not run on cache hit")
+    second = await query(req, AsyncMock(), key)
+
+    assert second.cached is True
+    assert second.answer == first.answer
+    assert second.metadata["citations_enforced"] == first.metadata["citations_enforced"]
+
+
+@pytest.mark.asyncio
+async def test_bypass_cache_skips_lookup(monkeypatch, _isolated_query_cache):
+    mock_rpc, _ = _mocks(monkeypatch)
+    key = _key()
+    await query(QueryRequest(query="Bypass me?"), AsyncMock(), key)
+    assert mock_rpc.await_count >= 1
+
+    mock_rpc.reset_mock()
+    res = await query(QueryRequest(query="Bypass me?", bypass_cache=True), AsyncMock(), key)
+
+    assert res.cached is False
+    assert mock_rpc.await_count >= 1
+
+
+@pytest.mark.asyncio
+async def test_save_to_wiki_never_populates_cache(monkeypatch, _isolated_query_cache):
+    _mocks(monkeypatch)
+    with patch("api.routers.query.get_vault_manager"):
+        req = QueryRequest(query="Wiki write?", save_to_wiki=True)
+        await query(req, AsyncMock(), _key())
+
+    assert [k for k in _isolated_query_cache.store if k.startswith("depthapi:q:")] == []
+
+
+@pytest.mark.asyncio
+async def test_redis_down_serves_uncached(monkeypatch):
+    _mocks(monkeypatch)
+    monkeypatch.setattr(cache_mod, "get_client", lambda: None)
+
+    res = await query(QueryRequest(query="No redis?"), AsyncMock(), _key())
+
+    assert res.cached is False
+    assert res.answer == "Mocked answer."
+
+
+@pytest.mark.asyncio
+async def test_quota_exceeded_returns_429(monkeypatch, _isolated_query_cache):
+    _mocks(monkeypatch)
+    key = _key()
+    _isolated_query_cache.store[cache_mod._quota_redis_key(key.id)] = "999999999"
+
+    with pytest.raises(HTTPException) as exc_info:
+        await query(QueryRequest(query="Over quota?"), AsyncMock(), key)
+
+    assert exc_info.value.status_code == 429
+
+
+@pytest.mark.asyncio
+async def test_quota_consumed_on_miss(monkeypatch, _isolated_query_cache):
+    _mocks(monkeypatch)
+    key = _key()
+
+    await query(QueryRequest(query="Count me?"), AsyncMock(), key)
+
+    used = int(_isolated_query_cache.store[cache_mod._quota_redis_key(key.id)])
+    assert used > 0
+
+
+@pytest.mark.asyncio
+async def test_quota_fail_open_when_redis_raises(monkeypatch):
+    _mocks(monkeypatch)
+
+    class _Boom:
+        def __getattr__(self, _):
+            raise RuntimeError("redis down")
+
+    monkeypatch.setattr(cache_mod, "get_client", lambda: _Boom())
+
+    res = await query(QueryRequest(query="Still served?"), AsyncMock(), _key())
+    assert res.cached is False
+
+
+@pytest.mark.asyncio
+async def test_stream_replays_cache_hit(monkeypatch, _isolated_query_cache):
+    _mocks(monkeypatch, answer="Cached stream answer.")
+    key = _key()
+    await query(QueryRequest(query="Stream me?"), AsyncMock(), key)
+
+    mock_rpc = AsyncMock(side_effect=AssertionError("RPC must not run on stream hit"))
+    monkeypatch.setattr(query_module, "execute_rpc", mock_rpc)
+
+    resp = await query_stream(QueryRequest(query="Stream me?"), AsyncMock(), key)
+    body = "".join([c.decode() if isinstance(c, bytes) else c async for c in resp.body_iterator])
+
+    assert ": stream start" in body
+    assert "Cached stream answer." in body
+    assert '"cached": true' in body
+    assert body.strip().endswith("data: [DONE]")
diff --git a/uv.lock b/uv.lock
index c949002..6bb09d1 100644
--- a/uv.lock
+++ b/uv.lock
@@ -101,6 +101,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/3c/d7/8fb3044eaef08a310acfe23dae9a8e2e07d305edc29a53497e52bc76eca7/asyncpg-0.31.0-cp314-cp314t-win_amd64.whl", hash = "sha256:bd4107bb7cdd0e9e65fae66a62afd3a249663b844fa34d479f6d5b3bef9c04c3", size = 706062, upload-time = "2025-11-24T23:26:44.086Z" },
 ]
 
+[[package]]
+name = "bm25s"
+version = "0.3.11"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
+    { name = "numpy", version = "2.5.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/a4/f6/38a4eb22be0c715aa9bc290e44a48bbdf7a27cffa7f85914fd10242442e4/bm25s-0.3.11.tar.gz", hash = "sha256:8ae0f0707d279969829383c7aaeef00fb09ca343101160222571d6bfc969f889", size = 80978, upload-time = "2026-08-25T03:14:57.502Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/41/d9/bb770dabb7e0edeb2c949460a05c38df7f4f564a1693a778352fb0d213b0/bm25s-0.3.11-py3-none-any.whl", hash = "sha256:3d1d28badb299d6fc9324111e5c8276927e990b908607b39ff9bd65b102e9a24", size = 74985, upload-time = "2026-08-25T03:14:56.051Z" },
+]
+
 [[package]]
 name = "certifi"
 version = "2026.7.22"
@@ -419,7 +432,6 @@ dependencies = [
     { name = "orjson" },
     { name = "pydantic" },
     { name = "pydantic-settings" },
-    { name = "rank-bm25" },
     { name = "redis" },
     { name = "sentry-sdk", extra = ["fastapi"] },
     { name = "slowapi" },
@@ -430,6 +442,7 @@ dependencies = [
 
 [package.optional-dependencies]
 dev = [
+    { name = "bm25s" },
     { name = "coverage" },
     { name = "pandas" },
     { name = "pytest" },
@@ -441,6 +454,7 @@ dev = [
 [package.metadata]
 requires-dist = [
     { name = "asyncpg", specifier = ">=0.29.0" },
+    { name = "bm25s", marker = "extra == 'dev'", specifier = ">=0.3.0" },
     { name = "coverage", marker = "extra == 'dev'", specifier = ">=7.6.0" },
     { name = "faiss-cpu", specifier = ">=1.8.0" },
     { name = "fastapi", specifier = ">=0.109.2" },
@@ -454,7 +468,6 @@ requires-dist = [
     { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.24.0" },
     { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=5.0.0" },
     { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.12.0" },
-    { name = "rank-bm25", specifier = ">=0.2.2" },
     { name = "redis", specifier = ">=5.0.1" },
     { name = "sentry-sdk", extras = ["fastapi"], specifier = ">=2.20.0" },
     { name = "slowapi", specifier = ">=0.1.9" },
@@ -1309,19 +1322,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f1/12/de94a39c2ef588c7e6455cfbe7343d3b2dc9d6b6b2f40c4c6565744c873d/pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b", size = 149341, upload-time = "2025-09-25T21:32:56.828Z" },
 ]
 
-[[package]]
-name = "rank-bm25"
-version = "0.2.2"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
-    { name = "numpy", version = "2.5.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/fc/0a/f9579384aa017d8b4c15613f86954b92a95a93d641cc849182467cf0bb3b/rank_bm25-0.2.2.tar.gz", hash = "sha256:096ccef76f8188563419aaf384a02f0ea459503fdf77901378d4fd9d87e5e51d", size = 8347, upload-time = "2022-02-16T12:10:52.196Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/2a/21/f691fb2613100a62b3fa91e9988c991e9ca5b89ea31c0d3152a3210344f9/rank_bm25-0.2.2-py3-none-any.whl", hash = "sha256:7bd4a95571adadfc271746fa146a4bcfd89c0cf731e49c3d1ad863290adbe8ae", size = 8584, upload-time = "2022-02-16T12:10:50.626Z" },
-]
-
 [[package]]
 name = "redis"
 version = "8.1.0"

From ab1f2abd93643b31ce1e1a8229af06e52c0694b7 Mon Sep 17 00:00:00 2001
From: sanjeevafk <strucker08@gmail.com>
Date: Fri, 4 Sep 2026 19:02:55 +0530
Subject: [PATCH 2/2] fix(security): resolve CodeQL weak sensitive data hashing
 alert

- Remove tenant_id from sha256 material input to break sensitive data taint path
- Explicitly set usedforsecurity=False on hashlib.sha256 for query cache hashing
- Scope Redis query cache key as depthapi:q:{tenant_id}:{digest}
- Rename api_key_id parameter to tenant_id across cache and quota helpers
---
 api/services/cache.py | 20 +++++++++++---------
 1 file changed, 11 insertions(+), 9 deletions(-)

diff --git a/api/services/cache.py b/api/services/cache.py
index bbfc22f..15a0acd 100644
--- a/api/services/cache.py
+++ b/api/services/cache.py
@@ -51,15 +51,17 @@ def reset_client() -> None:
     _client = None
 
 
-def cache_key(api_key_id: str, query: str, collection_id: str | None, depth: int,
+def cache_key(tenant_id: str, query: str, collection_id: str | None, depth: int,
               temperature: float, rerank: bool, use_trusted: bool,
               graph_hops: int | None, llm_model: str) -> str:
+    """Generate a deterministic Redis cache key scoped by tenant/API key ID."""
     material = "|".join([
-        f"v{CACHE_VERSION}", api_key_id, query.strip(),
+        f"v{CACHE_VERSION}", query.strip(),
         collection_id or "", str(depth), f"{temperature:.2f}",
         str(rerank), str(use_trusted), str(graph_hops), llm_model,
     ])
-    return "depthapi:q:" + hashlib.sha256(material.encode("utf-8")).hexdigest()
+    digest = hashlib.sha256(material.encode("utf-8"), usedforsecurity=False).hexdigest()
+    return f"depthapi:q:{tenant_id}:{digest}"
 
 
 def get_cached(key: str) -> dict[str, Any] | None:
@@ -108,12 +110,12 @@ def count_tokens(text: str, model: str) -> int:
         return 0
 
 
-def _quota_redis_key(api_key_id: str) -> str:
+def _quota_redis_key(tenant_id: str) -> str:
     day = datetime.now(timezone.utc).strftime("%Y%m%d")
-    return f"depthapi:quota:{api_key_id}:{day}"
+    return f"depthapi:quota:{tenant_id}:{day}"
 
 
-def check_quota(api_key_id: str, is_pro: bool, estimated_tokens: int) -> None:
+def check_quota(tenant_id: str, is_pro: bool, estimated_tokens: int) -> None:
     """Raise 429 when the key would exceed its daily token budget. Fail-open."""
     limit = quota_limit(is_pro)
     if limit <= 0:
@@ -122,7 +124,7 @@ def check_quota(api_key_id: str, is_pro: bool, estimated_tokens: int) -> None:
     if client is None:
         return
     try:
-        used = int(client.get(_quota_redis_key(api_key_id)) or 0)
+        used = int(client.get(_quota_redis_key(tenant_id)) or 0)
     except Exception as exc:
         log.warning("Quota read failed, allowing request: %s", exc)
         return
@@ -130,14 +132,14 @@ def check_quota(api_key_id: str, is_pro: bool, estimated_tokens: int) -> None:
         raise HTTPException(429, "Daily token quota exceeded")
 
 
-def consume_quota(api_key_id: str, tokens: int) -> None:
+def consume_quota(tenant_id: str, tokens: int) -> None:
     if tokens <= 0:
         return
     client = get_client()
     if client is None:
         return
     try:
-        key = _quota_redis_key(api_key_id)
+        key = _quota_redis_key(tenant_id)
         client.incrby(key, tokens)
         client.expire(key, 172800)
     except Exception as exc: