Deploy 8ca4892

app/models/pipeline.py

@@ -60,8 +60,8 @@ class PipelineState(TypedDict):
     session_id: str
     query_embedding: Optional[list[float]]   # set by cache node, reused by retrieve
     expanded_queries: Annotated[list[str], operator.add]
-    retrieved_chunks: Annotated[list[Chunk], operator.add]
-    reranked_chunks: Annotated[list[Chunk], operator.add]
+    retrieved_chunks: list[Chunk]
+    reranked_chunks: list[Chunk]
     answer: str
     sources: Annotated[list[SourceRef], operator.add]
     cached: bool

app/pipeline/nodes/gemini_fast.py

@@ -1,22 +1,7 @@
-"""
-backend/app/pipeline/nodes/gemini_fast.py
-
-Fast-path node: Gemini 2.0 Flash answers conversational / general queries
-directly from a TOON-encoded portfolio context summary, avoiding full RAG.
-
-Decision logic:
-  - Gemini answers  → state.answer is set, pipeline skips retrieve/generate.
-  - Gemini calls search_knowledge_base() → state.thinking=True, pipeline
-    goes to retrieve+generate so the user gets a cited answer.
+"""Deterministic fast-path node used only for small-talk handling.
 
-Bug A fix (Issue 2): Gemini fast-path answers now run through the same
-is_low_trust() quality gate as Groq answers. If the gate fires (hedge phrase
-detected, or suspiciously short complex answer), the answer is discarded
-and the pipeline routes to full RAG instead of returning a low-quality answer.
-
-Issue 7 fix: _is_complex() now requires BOTH a keyword match AND query length
-> 8 words, eliminating false-positive complex classifications for short
-conversational queries like "How?" or "How many projects?".
+All substantive queries are routed to full RAG so answers remain retrieval-grounded
+and citation-capable. No parametric Gemini answer generation is used here.
 """
 from __future__ import annotations
 
@@ -26,11 +11,8 @@ from typing import Any
 
 from langgraph.config import get_stream_writer
 
-from app.core.portfolio_context import is_portfolio_relevant
-from app.core.portfolio_context import KNOWN_ORGS, KNOWN_PROJECTS
 from app.models.pipeline import PipelineState
 from app.services.gemini_client import GeminiClient
-from app.core.quality import is_low_trust
 
 logger = logging.getLogger(__name__)
 
@@ -61,92 +43,6 @@ _SMALL_TALK_ANSWER = (
     "and I'll find the details for you."
 )
 
-# Words that reliably indicate the visitor wants a deep, cited answer.
-# Kept for the quality gate docstring; no longer used for routing.
-_COMPLEX_SIGNALS: frozenset[str] = frozenset({
-    "how", "why", "explain", "implement", "architecture", "deep",
-    "detail", "technical", "compare", "difference", "algorithm",
-    "code", "example", "breakdown", "analysis", "source", "cite",
-    "reference", "proof", "derive", "calculate", "optimise", "optimize",
-    # Follow-up depth signals — these phrases appear in pill-generated questions
-    # and always indicate the user wants a cited, retrieved answer not a summary.
-    "tell me more", "more detail", "more about", "what about",
-    "explain that", "go deeper", "expand", "elaborate", "dig into",
-})
-
-# Minimum token count for a query to be classified as complex.
-# Queries shorter than this are almost always conversational or simple
-# biographical lookups regardless of vocabulary. "How?" alone currently
-# triggers 70B without this gate; "How many projects?" should not.
-# Documented in copilot-instructions.md — do not lower without profiling.
-_COMPLEX_MIN_WORDS: int = 8
-
-# Named-entity detection: a capitalised word that is NOT at position 0.
-# Presence of a NE after the first word in a short query means it is asking
-# about something specific (a company, project, technology) that needs Qdrant.
-_NE_RE = re.compile(r'(?<=\s)[A-Z][a-zA-Z0-9]+')
-
-# Navigation phrases that Gemini can answer from the TOON portfolio summary.
-# Keep this tight — anything not in this set routes to RAG.
-_TRIVIAL_PHRASES: frozenset[str] = frozenset({
-    "what is this", "what's this",
-    "who are you", "what are you",
-    "what can you do", "what can you help with", "what can you help me with",
-    "what do you do",
-})
-
-_ENTITY_SPECIFIC_NOUNS: frozenset[str] = KNOWN_PROJECTS | KNOWN_ORGS
-
-
-def _is_trivial(query: str) -> bool:
-    """
-    True when the query is pure navigation — safe for Gemini fast-path.
-
-    A query is trivial ONLY when its stripped form exactly matches a known
-    navigation phrase from _TRIVIAL_PHRASES.  All other queries — including
-    short career/skills/internship questions — route to full RAG so they
-    receive citations backed by Qdrant evidence, not the stale TOON summary.
-
-    Removing the <4-word bypass (RC-10): queries like "his skills?" or
-    "any internships?" previously hit the TOON fast-path and could return
-    un-cited, outdated answers. They now always go through retrieval.
-    """
-    stripped = query.strip().rstrip("?!.")
-    return stripped.lower() in _TRIVIAL_PHRASES
-
-
-def _is_complex(query: str) -> bool:
-    """
-    O(1) heuristic — true when the query signals a need for a cited answer.
-
-    Used only to pick the LLM model tier (70B vs 8B), NOT for RAG routing.
-    Routing is decided by _is_trivial(); _is_complex() is a post-routing concern.
-
-    A query is complex only when BOTH conditions hold:
-      1. It contains a complexity-signal keyword (architecture, explain, etc.)
-      2. Its length exceeds _COMPLEX_MIN_WORDS (eliminates "How?" false positives)
-    OR it is extremely long (>20 tokens, reliably indicates detailed request).
-    """
-    tokens = set(query.lower().split())
-    token_count = len(tokens)
-    if token_count > 20:
-        return True
-    return bool(tokens & _COMPLEX_SIGNALS) and token_count > _COMPLEX_MIN_WORDS
-
-
-def _is_entity_specific_portfolio_query(query: str) -> bool:
-    tokens = re.findall(r"[a-z0-9]+", query.lower())
-    if not tokens:
-        return False
-    for token in tokens:
-        if token in _ENTITY_SPECIFIC_NOUNS:
-            return True
-    for a, b in zip(tokens, tokens[1:]):
-        if f"{a} {b}" in _ENTITY_SPECIFIC_NOUNS:
-            return True
-    return False
-
-
 def make_gemini_fast_node(gemini_client: GeminiClient) -> Any:
     """
     Returns a LangGraph-compatible async node function.
@@ -187,74 +83,13 @@ def make_gemini_fast_node(gemini_client: GeminiClient) -> Any:
                 "path": "gemini_fast",
             }
 
-        complexity = "complex" if _is_complex(query) else "simple"
-
-        # Force RAG for entity-specific portfolio queries (project/org names).
-        # Broad intent-only phrasing (e.g., "what tech stack does he use") first
-        # goes through Gemini fast-path and falls back to RAG if low-trust.
-        if (
-            not _is_trivial(query)
-            and is_portfolio_relevant(query)
-            and _is_entity_specific_portfolio_query(query)
-        ):
-            logger.debug("Non-trivial query — routing directly to RAG: %r", query[:60])
-            writer({"type": "status", "label": "Searching portfolio..."})
-            return {
-                "query_complexity": complexity,
-                "expanded_queries": [query],
-                "thinking": False,
-            }
-
-        # When Gemini is not configured (GEMINI_API_KEY not set), route all
-        # traffic straight to RAG — behaviour is identical to the old graph.
-        if not gemini_client.is_configured:
-            logger.debug("Gemini not configured; routing query to RAG.")
-            writer({"type": "status", "label": "Needs deeper search, checking portfolio..."})
-            return {
-                "query_complexity": complexity,
-                "expanded_queries": [query],
-                "thinking": False,
-            }
-
-        answer, tool_query = await gemini_client.fast_answer(
-            query,
-            history=state.get("conversation_history") or [],
-        )
-
-        if answer is not None:
-            # Run the same quality gate that guards Groq answers.
-            if is_low_trust(answer, [], complexity):
-                logger.debug(
-                    "Gemini fast-path answer failed quality gate — routing to RAG."
-                )
-                writer({"type": "status", "label": "Needs deeper search, checking portfolio..."})
-                return {
-                    "query_complexity": complexity,
-                    "expanded_queries": [query],
-                    "thinking": True,
-                }
-
-            # Gemini answered and passed the quality gate.
-            logger.debug("Gemini fast-path answered query (len=%d)", len(answer))
-            writer({"type": "status", "label": "Got a direct answer, writing now..."})
-            # Gemini does not stream; emit the complete answer as one token event.
-            writer({"type": "token", "text": answer})
-            return {
-                "query_complexity": complexity,
-                "answer": answer,
-                "sources": [],
-                "thinking": False,
-                "path": "gemini_fast",
-            }
-
-        # Gemini called search_knowledge_base() — route to full RAG.
-        rag_query = tool_query or query
-        logger.debug("Gemini routed to RAG (tool_query=%r)", rag_query)
+        # Substantive queries are always retrieval-grounded.
+        logger.debug("Routing query to RAG path from deterministic fast node.")
         writer({"type": "status", "label": "Needs deeper search, checking portfolio..."})
         return {
-            "query_complexity": complexity,
-            "expanded_queries": [rag_query],
-            "thinking": True,
+            "query_complexity": "simple",
+            "expanded_queries": [query],
+            "thinking": False,
         }
 
     return gemini_fast

app/pipeline/nodes/generate.py

@@ -8,7 +8,6 @@ from langgraph.config import get_stream_writer
 
 from app.models.chat import SourceRef
 from app.models.pipeline import PipelineState
-from app.core.portfolio_context import SUGGESTION_HINT
 from app.services.llm_client import LLMClient
 from app.core.quality import is_low_trust
 logger = logging.getLogger(__name__)
@@ -19,7 +18,10 @@ logger = logging.getLogger(__name__)
 _THINK_COMPLETE_RE = re.compile(r"<think>[\s\S]*?</think>", re.DOTALL)
 _THINK_OPEN_RE = re.compile(r"<think>")
 _THINK_CLOSE_RE = re.compile(r"</think>")
-_GEN_HTML_TAG_RE = re.compile(r"</?[a-zA-Z][^>]*>")
+_GEN_HTML_TAG_RE = re.compile(
+    r"</?(?:div|span|p|br|ul|ol|li|strong|em|code|pre|h[1-6]|a|section|article|header|footer|main|aside|blockquote|table|thead|tbody|tr|td|th|img)[^>]*>",
+    re.IGNORECASE,
+)
 _VERSION_PARITY_RE = re.compile(r"\b(up[- ]?to[- ]?date|latest|current|in sync|same version|version)\b", re.IGNORECASE)
 _WORD_RE = re.compile(r"[a-zA-Z0-9]+")
 
@@ -163,6 +165,8 @@ def _format_history(state: "PipelineState") -> str:
     history = state.get("conversation_history") or []
     if not history:
         return ""
+    # Keep fallback prompt context bounded for clarity and token predictability.
+    history = history[-3:]
     lines = [
         f"[T{i + 1}] Q: {t['q']} | A: {t['a']}"
         for i, t in enumerate(history)
@@ -330,54 +334,19 @@ def _build_low_trust_fallback(query: str, source_refs: list[SourceRef]) -> str:
     if not source_refs:
         return _NOT_FOUND_ANSWER
 
-    first = source_refs[0]
-    title = first.title or "the retrieved source"
-
     if _VERSION_PARITY_RE.search(query):
         return (
-            f"The retrieved sources confirm links/details for {title} [1], but they do not explicitly "
+            "The indexed sources include related details [1], but they do not explicitly "
             "confirm whether the GitHub code and live demo are currently in sync, so version parity "
-            "cannot be verified from the indexed content alone [1]."
+            "cannot be verified from indexed content alone [1]."
         )
 
     return (
-        f"Based on the retrieved evidence, the answer is grounded in {title} [1]. "
-        "If you want deeper detail, ask for a specific section, implementation part, or comparison."
+        "I couldn't find enough specific information to answer that confidently. "
+        "Try asking about a specific project, blog post, skill, or company."
     )
 
 
-def _select_chunks_for_prompt(query: str, reranked_chunks: list[dict]) -> list[dict]:
-    """
-    Prefer chunks whose source title is explicitly referenced in the query.
-
-    This prevents focused questions (e.g. one project) from receiving multi-project
-    blended context that can trigger verbose, low-quality comparison answers.
-    """
-    if not reranked_chunks:
-        return reranked_chunks
-
-    query_lower = query.lower()
-    focused: list[dict] = []
-
-    for chunk in reranked_chunks:
-        title = str(chunk["metadata"].get("source_title", "")).strip()
-        if not title:
-            continue
-        title_lower = title.lower()
-        if len(title_lower) >= 4 and title_lower in query_lower:
-            focused.append(chunk)
-            continue
-
-        title_tokens = [t for t in _WORD_RE.findall(title_lower) if len(t) >= 4]
-        if title_tokens and sum(1 for tok in title_tokens if tok in query_lower) >= min(2, len(title_tokens)):
-            focused.append(chunk)
-
-    if focused:
-        return focused[:6]
-
-    return reranked_chunks[:8]
-
-
 def make_generate_node(llm_client: LLMClient, gemini_client=None) -> Callable[[PipelineState], dict]:  # noqa: ANN001
     # Number of token chunks to buffer before deciding there is no CoT block.
     # Llama 3.1 8B may omit <think> entirely; Llama 3.3 70B always starts with one.
@@ -389,6 +358,9 @@ def make_generate_node(llm_client: LLMClient, gemini_client=None) -> Callable[[P
         query = state["query"]
         complexity = state.get("query_complexity", "simple")
         reranked_chunks = state.get("reranked_chunks", [])
+        if len(reranked_chunks) > 12:
+            logger.warning("generate: unusually large reranked chunk set (%d); truncating to 12.", len(reranked_chunks))
+            reranked_chunks = reranked_chunks[:12]
 
         # ── Enumeration path (Fix 1) ──────────────────────────────────────────────
         # enumerate_query node already set is_enumeration_query=True and populated
@@ -433,23 +405,8 @@ def make_generate_node(llm_client: LLMClient, gemini_client=None) -> Callable[[P
         # ── Not-found path ────────────────────────────────────────────────────────────
         if not reranked_chunks:
             writer({"type": "status", "label": "Could not find specific information, responding carefully..."})
-
-            # Fix 2 Rule 2: generate a specific, topical redirect suggestion using
-            # Gemini with the TOON portfolio entity list. Fires here after CRAG
-            # retries are exhausted so the user gets contextual guidance.
-            query_topic = state.get("query_topic") or "this topic"
-            not_found_answer = _NOT_FOUND_ANSWER
-            if gemini_client is not None and getattr(gemini_client, "is_configured", False):
-                suggestion = await gemini_client.generate_specific_suggestion(
-                    query=query,
-                    query_topic=query_topic,
-                    suggestion_hint=SUGGESTION_HINT,
-                )
-                if suggestion:
-                    not_found_answer = suggestion
-
-            writer({"type": "token", "text": not_found_answer})
-            return {"answer": not_found_answer, "sources": [], "path": "rag"}
+            writer({"type": "token", "text": _NOT_FOUND_ANSWER})
+            return {"answer": _NOT_FOUND_ANSWER, "sources": [], "path": "rag"}
 
         # ── Build numbered context block ────────────────────────────────────
         # Merge chunks from the same source URL first so every [N] in the prompt
@@ -457,7 +414,7 @@ def make_generate_node(llm_client: LLMClient, gemini_client=None) -> Callable[[P
         # TextOps become [1] and [2] — the LLM cites both in the same sentence,
         # which looks like self-citing hallucination even though it is technically
         # correct.  _merge_by_source preserves all text; nothing is discarded.
-        selected_chunks = _select_chunks_for_prompt(query, reranked_chunks)
+        selected_chunks = reranked_chunks[:8]
         merged_chunks = _merge_by_source(selected_chunks)
         context_parts: list[str] = []
         source_refs: list[SourceRef] = []
@@ -578,7 +535,8 @@ def make_generate_node(llm_client: LLMClient, gemini_client=None) -> Callable[[P
                 if clean_buf:
                     writer({"type": "token", "text": clean_buf})
         except (Exception, asyncio.CancelledError) as exc:
-            logger.warning("Generation stream interrupted: %s / Returning partial answer.", exc)
+            logger.warning("Generation stream interrupted (%s: %s) / Returning partial answer.", type(exc).__name__, exc)
+            writer({"type": "status", "label": "Stream interrupted; finalizing partial answer..."})
             # Flush whatever we have if the stream was cut
             if buf:
                 clean_buf = _strip_think_tags(buf)
@@ -594,15 +552,6 @@ def make_generate_node(llm_client: LLMClient, gemini_client=None) -> Callable[[P
         # It handles complete pairs, orphaned opens, and orphaned closes.
         full_answer = _strip_think_tags(raw_answer)
 
-        # ── Quality gate: Gemini editorial reformat ──────────────────────────
-        # Fires when: (a) criticism detected — always reformat, or
-        # (b) low-trust heuristic flags the draft. Zero extra cost on good responses.
-        if gemini_client is not None and (is_criticism or is_low_trust(full_answer, reranked_chunks, complexity)):
-            logger.debug("Triggering Gemini reformat (criticism=%s).", is_criticism)
-            reformatted = await gemini_client.reformat_rag_answer(query, context_block, full_answer)
-            if reformatted:
-                full_answer = reformatted
-
         full_answer = _normalise_answer_text(full_answer, max_citation_index=len(source_refs))
 
         # Final guardrail: if answer still looks low-trust after reformat + cleanup,

app/pipeline/nodes/log_eval.py

@@ -38,10 +38,15 @@ def make_log_eval_node(db_path: str, github_log=None) -> Callable[[PipelineState
         rerank_scores = json.dumps(
             [c["metadata"].get("rerank_score", 0.0) for c in state.get("reranked_chunks", [])]
         )
-        # Store chunk texts verbatim so data_prep.py can build reranker triplets
-        # without needing to re-fetch from Qdrant.
+        # Store only retrieval metadata to reduce PII exposure in interaction logs.
+        # Full text can be reconstructed on-demand by doc_id from Qdrant if needed.
         reranked_chunks_json = json.dumps(
-            [{"text": c["text"], "doc_id": c["metadata"]["doc_id"]}
+            [{
+                "doc_id": c["metadata"].get("doc_id", ""),
+                "source_title": c["metadata"].get("source_title", ""),
+                "source_type": c["metadata"].get("source_type", ""),
+                "section": c["metadata"].get("section", ""),
+            }
              for c in state.get("reranked_chunks", [])]
         )
         path = state.get("path") or "rag"
@@ -140,7 +145,12 @@ def make_log_eval_node(db_path: str, github_log=None) -> Callable[[PipelineState
                         json.dumps([c["metadata"]["doc_id"] for c in state.get("reranked_chunks", [])])
                     ),
                     "reranked_chunks_json": [
-                        {"text": c["text"], "doc_id": c["metadata"]["doc_id"]}
+                        {
+                            "doc_id": c["metadata"].get("doc_id", ""),
+                            "source_title": c["metadata"].get("source_title", ""),
+                            "source_type": c["metadata"].get("source_type", ""),
+                            "section": c["metadata"].get("section", ""),
+                        }
                         for c in state.get("reranked_chunks", [])
                     ],
                     "rerank_scores": [

app/pipeline/nodes/retrieve.py

@@ -97,6 +97,29 @@ _CAPABILITY_QUERY_HINTS: frozenset[str] = frozenset(
     }
 )
 
+_BIOGRAPHY_QUERY_HINTS: frozenset[str] = frozenset(
+    {
+        "work",
+        "experience",
+        "employment",
+        "career",
+        "internship",
+        "internships",
+        "education",
+        "degree",
+        "university",
+        "background",
+        "resume",
+        "cv",
+        "company",
+        "companies",
+        "role",
+        "roles",
+    }
+)
+
+_BIO_SOURCE_TYPES: frozenset[str] = frozenset({"resume", "cv", "bio"})
+
 _NORMALISATION_STOPWORDS: frozenset[str] = frozenset(
     {
         "tell",
@@ -242,6 +265,11 @@ def _is_capability_query(query: str) -> bool:
     return bool(tokens & _CAPABILITY_QUERY_HINTS)
 
 
+def _is_biography_query(query: str) -> bool:
+    tokens = frozenset(re.findall(r"[a-z0-9]+", query.lower()))
+    return bool(tokens & _BIOGRAPHY_QUERY_HINTS)
+
+
 def _is_informative_chunk(chunk: Chunk) -> bool:
     """True when chunk text has enough lexical content for cross-encoder reranking."""
     text = (chunk.get("contextualised_text") or chunk["text"] or "").strip()
@@ -505,6 +533,23 @@ def make_retrieve_node(
         if not rerank_candidates:
             rerank_candidates = unique_chunks
 
+        # Biography-focused queries should prioritize resume/CV evidence and avoid
+        # project/blog code passages crowding out personal background facts.
+        if _is_biography_query(retrieval_query):
+            bio_candidates = [
+                chunk
+                for chunk in rerank_candidates
+                if chunk["metadata"].get("source_type", "") in _BIO_SOURCE_TYPES
+            ]
+            if bio_candidates:
+                rerank_candidates = bio_candidates
+                writer(
+                    {
+                        "type": "status",
+                        "label": "Prioritizing resume and background sources...",
+                    }
+                )
+
         try:
             reranked = await reranker.rerank(retrieval_query, rerank_candidates, top_k=10)  # RC-5: raised from 7
         except (Exception, asyncio.CancelledError) as exc:

tests/test_generate_focus_selection.py

@@ -1,37 +1,36 @@
-from app.pipeline.nodes.generate import _select_chunks_for_prompt
-
-
-def _chunk(title: str, section: str = "Overview") -> dict:
-    return {
-        "text": f"Info about {title}",
-        "metadata": {
-            "source_title": title,
-            "section": section,
-            "source_url": "",
-            "source_type": "project",
-        },
-    }
+from app.pipeline.nodes.generate import _format_history
 
 
-def test_select_chunks_prefers_explicitly_mentioned_source_title() -> None:
-    chunks = [
-        _chunk("Sorting Demo"),
-        _chunk("EchoEcho"),
-        _chunk("Donut.asm"),
-    ]
+def test_format_history_uses_summary_when_present() -> None:
+    state = {
+        "conversation_summary": "User asked about PersonaBot architecture and caching.",
+        "conversation_history": [
+            {"q": "Q1", "a": "A1"},
+            {"q": "Q2", "a": "A2"},
+        ],
+    }
 
-    selected = _select_chunks_for_prompt(
-        "Is the source code on GitHub up-to-date with the Sorting Demo live demo?",
-        chunks,
-    )
+    rendered = _format_history(state)
 
-    assert selected
-    assert all(c["metadata"]["source_title"] == "Sorting Demo" for c in selected)
+    assert "Running conversation context:" in rendered
+    assert "PersonaBot architecture" in rendered
+    assert "Q1" not in rendered
 
 
-def test_select_chunks_falls_back_to_top_ranked_when_no_title_match() -> None:
-    chunks = [_chunk("A"), _chunk("B"), _chunk("C")]
+def test_format_history_caps_raw_history_to_three_turns() -> None:
+    state = {
+        "conversation_summary": None,
+        "conversation_history": [
+            {"q": "Q1", "a": "A1"},
+            {"q": "Q2", "a": "A2"},
+            {"q": "Q3", "a": "A3"},
+            {"q": "Q4", "a": "A4"},
+        ],
+    }
 
-    selected = _select_chunks_for_prompt("What technologies are used?", chunks)
+    rendered = _format_history(state)
 
-    assert selected == chunks
+    assert "Q1" not in rendered
+    assert "Q2" in rendered
+    assert "Q3" in rendered
+    assert "Q4" in rendered

tests/test_generate_not_found_fallback.py

@@ -8,14 +8,14 @@ _WRITER_PATCH = "app.pipeline.nodes.gemini_fast.get_stream_writer"
 
 
 @pytest.mark.asyncio
-async def test_non_portfolio_conversational_query_uses_gemini_answer() -> None:
+async def test_small_talk_returns_deterministic_fast_path_answer() -> None:
     gemini = MagicMock()
     gemini.is_configured = True
-    gemini.fast_answer = AsyncMock(return_value=("Hey there!", None))
+    gemini.fast_answer = AsyncMock(return_value=("unused", None))
 
     node = make_gemini_fast_node(gemini)
     state = {
-        "query": "I said hello.",
+        "query": "Hi",
         "is_followup": False,
         "conversation_history": [],
     }
@@ -23,9 +23,9 @@ async def test_non_portfolio_conversational_query_uses_gemini_answer() -> None:
     with patch(_WRITER_PATCH, return_value=MagicMock()):
         result = await node(state)
 
-    assert result["answer"] == "Hey there!"
+    assert "Darshan's portfolio assistant" in result["answer"]
     assert result["path"] == "gemini_fast"
-    gemini.fast_answer.assert_awaited_once()
+    gemini.fast_answer.assert_not_awaited()
 
 
 @pytest.mark.asyncio
@@ -48,22 +48,23 @@ async def test_portfolio_specific_query_forces_rag() -> None:
     assert result["expanded_queries"] == ["How does TextOps work?"]
     gemini.fast_answer.assert_not_awaited()
 
-    @pytest.mark.asyncio
-    async def test_broad_portfolio_intent_can_use_gemini_fast_path() -> None:
-        gemini = MagicMock()
-        gemini.is_configured = True
-        gemini.fast_answer = AsyncMock(return_value=("He uses a broad stack.", None))
-
-        node = make_gemini_fast_node(gemini)
-        state = {
-            "query": "What tech stack does he use?",
-            "is_followup": False,
-            "conversation_history": [],
-        }
-
-        with patch(_WRITER_PATCH, return_value=MagicMock()):
-            result = await node(state)
-
-        assert result["answer"] == "He uses a broad stack."
-        assert result["path"] == "gemini_fast"
-        gemini.fast_answer.assert_awaited_once()
+
+@pytest.mark.asyncio
+async def test_non_small_talk_routes_to_rag() -> None:
+    gemini = MagicMock()
+    gemini.is_configured = True
+    gemini.fast_answer = AsyncMock(return_value=("unused", None))
+
+    node = make_gemini_fast_node(gemini)
+    state = {
+        "query": "What tech stack does he use?",
+        "is_followup": False,
+        "conversation_history": [],
+    }
+
+    with patch(_WRITER_PATCH, return_value=MagicMock()):
+        result = await node(state)
+
+    assert "answer" not in result
+    assert result["expanded_queries"] == ["What tech stack does he use?"]
+    gemini.fast_answer.assert_not_awaited()

tests/test_generate_quality_fallback.py

@@ -36,5 +36,5 @@ def test_low_trust_fallback_general_query_is_concise() -> None:
         sources,
     )
 
-    assert "Sorting Demo" in answer
-    assert "[1]" in answer
+    assert "specific project" in answer.lower()
+    assert "sorting demo" not in answer.lower()

tests/test_log_eval_privacy.py

@@ -0,0 +1,45 @@
+import json
+import sqlite3
+
+import pytest
+
+from app.pipeline.nodes.log_eval import make_log_eval_node
+
+
+@pytest.mark.asyncio
+async def test_log_eval_stores_chunk_metadata_without_text(tmp_path) -> None:
+    db_path = str(tmp_path / "interactions.db")
+    node = make_log_eval_node(db_path)
+
+    state = {
+        "session_id": "s1",
+        "query": "What work experience does Darshan have?",
+        "answer": "He worked at VK Live.",
+        "reranked_chunks": [
+            {
+                "text": "Phone +44 7818 975908 and email someone@example.com",
+                "metadata": {
+                    "doc_id": "resume-rag",
+                    "source_title": "Resume",
+                    "source_type": "resume",
+                    "section": "Work Experience",
+                    "rerank_score": 0.9,
+                },
+            }
+        ],
+        "latency_ms": 123,
+        "cached": False,
+        "path": "rag",
+        "is_enumeration_query": False,
+    }
+
+    await node(state)
+
+    with sqlite3.connect(db_path) as conn:
+        row = conn.execute("SELECT reranked_chunks_json FROM interactions LIMIT 1").fetchone()
+
+    assert row is not None
+    payload = json.loads(row[0])
+    assert payload and payload[0]["doc_id"] == "resume-rag"
+    assert payload[0]["source_title"] == "Resume"
+    assert "text" not in payload[0]