Fix analytics accuracy, tenant isolation, and event-loop blocking.

Correct funnel link-click aggregation, visitor fingerprint, and turn tracking; use full UUID Chroma collection names; offload CPU/DNS work to thread pool; share OpenAI client across intent/rewrite paths.

Co-authored-by: Cursor <cursoragent@cursor.com>

app/api/chat.py

@@ -212,6 +212,9 @@ async def chat(
         book_id=session_ctx.selected_book_id,
         user_message=payload.message,
         result=result,
+        # DESIGN-5 fix: session_mgr.save() has already incremented turn_count above,
+        # so this is the correct completed-turn number for this event row.
+        turn_number=session_ctx.turn_count,
     )
 
     # Phase 5A (B8 fix): Warn author by email when token budget hits 80%.
@@ -309,6 +312,8 @@ async def stream_start(
                     book_id=session_ctx.selected_book_id,
                     user_message=payload.message,
                     result=pipeline_result,
+                    # DESIGN-5 fix: same as sync path — turn_count already incremented.
+                    turn_number=session_ctx.turn_count,
                 )
 
                 if hasattr(author, "_access_grant") and author._access_grant:

app/api/ingest.py

@@ -26,14 +26,17 @@ async def upload_book(
     file: UploadFile = File(...),
     title: str = "",
     genre: str = "",
-    price: float = 0.0,
     buy_url: str = "",
     force_replace: bool = Query(False),
     current_user=Depends(get_current_author_scoped),
     db=Depends(get_db),
     redis=Depends(get_redis),
 ):
-    """Upload a book file (PDF/EPUB/DOCX/TXT) and start async ingestion."""
+    """Upload a book file (PDF/EPUB/DOCX/TXT) and start async ingestion.
+
+    MINOR-7 fix: removed 'price' param — start_file_upload() never accepted it
+    (it was silently dropped on every request).
+    """
     MAX_UPLOAD_BYTES = cfg.UPLOAD_MAX_FILE_SIZE_MB * 1024 * 1024
     contents = bytearray()
     while chunk := await file.read(64 * 1024):

app/models/analytics.py

@@ -2,7 +2,7 @@
 
 import datetime
 
-from sqlalchemy import Boolean, Date, Float, ForeignKey, Integer, String
+from sqlalchemy import Boolean, Date, Float, ForeignKey, Integer, String, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import Base, TimestampMixin, generate_uuid
@@ -43,6 +43,12 @@ class AnalyticsDaily(Base):
     """Pre-aggregated daily analytics — populated by Celery task hourly."""
 
     __tablename__ = "analytics_daily"
+    # DESIGN-2 fix: add the unique constraint required by the ON CONFLICT upsert
+    # in aggregator.py. Without this, create_all creates the table without the
+    # constraint and the upsert silently fails, creating duplicate rows.
+    __table_args__ = (
+        UniqueConstraint("author_id", "date", name="uq_analytics_daily_author_date"),
+    )
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=generate_uuid)
     author_id: Mapped[str] = mapped_column(

app/services/analytics_core/aggregator.py

@@ -49,7 +49,9 @@ async def _run_daily_rollup(target_date: date) -> dict:
                 func.count(AnalyticsEvent.id).label("total_chats"),
                 func.count(func.distinct(AnalyticsEvent.visitor_fingerprint)).label("unique_visitors"),
                 func.sum(AnalyticsEvent.prompt_tokens + AnalyticsEvent.completion_tokens).label("total_tokens"),
-                func.sum(AnalyticsEvent.link_shown.cast("int")).label("total_link_clicks"),
+                # DESIGN-1 fix: was summing link_shown (button displayed) — should be
+                # link_clicked (button actually clicked). Fixes inflated conversion funnel.
+                func.sum(AnalyticsEvent.link_clicked.cast("int")).label("total_link_clicks"),
                 func.avg(AnalyticsEvent.faithfulness_score).label("avg_faithfulness"),
             ).where(
                 AnalyticsEvent.timestamp >= day_start,

app/services/analytics_core/tracker.py

@@ -51,6 +51,7 @@ async def track_turn(
     book_id: str | None,
     user_message: str,
     result,
+    turn_number: int = 0,  # DESIGN-5 fix: actual completed turn count, passed by caller
 ) -> None:
     """Log a chat turn to the analytics_events table.
 
@@ -62,9 +63,20 @@ async def track_turn(
         book_id: UUID of the selected book.
         user_message: The user's raw message (not stored — only metadata).
         result: PipelineResult from the RAG pipeline.
+        turn_number: Completed turn count for this turn (from session_ctx.turn_count).
     """
     try:
         from datetime import datetime, timezone
+        from sqlalchemy import select
+        from app.models.chat_session import ChatSession
+
+        # DESIGN-4 fix: visitor_fingerprint was hardcoded to "" which made
+        # unique-visitor aggregation (COUNT DISTINCT visitor_fingerprint) always
+        # return 1. Now fetched from the ChatSession row written at session init.
+        fp_result = await db.execute(
+            select(ChatSession.visitor_fingerprint).where(ChatSession.id == session_id)
+        )
+        visitor_fingerprint = fp_result.scalar_one_or_none() or ""
 
         # Save messages to DB
         user_msg = ChatMessage(
@@ -99,7 +111,9 @@ async def track_turn(
             author_id=author_id,
             book_id=book_id or (result.top_book_ids[0] if result.top_book_ids else None),
             timestamp=datetime.now(timezone.utc),
-            turn_number=0,  # Will be updated by aggregator
+            # DESIGN-5 fix: use the actual completed turn number passed by the caller.
+            # Aggregator never updated this field (see audit DESIGN-5).
+            turn_number=turn_number,
             intent=result.intent,
             intent_confidence=result.intent_confidence,
             faithfulness_score=result.faithfulness_score,
@@ -110,7 +124,7 @@ async def track_turn(
             response_ms=result.response_ms,
             upsell_strategy=result.upsell_strategy,
             link_shown=result.link_shown,
-            visitor_fingerprint="",
+            visitor_fingerprint=visitor_fingerprint,
         )
         db.add(event)
 

app/services/embeddings.py

@@ -47,10 +47,14 @@ def get_collection_name(author_id: str, book_id: str) -> str:
     Returns:
         Collection name string.
     """
-    # Use first 8 chars of each UUID for brevity (still unique enough with combined key)
-    short_author = author_id.replace("-", "")[:12]
-    short_book = book_id.replace("-", "")[:12]
-    return f"a{short_author}_b{short_book}"
+    # SEC-4 fix: use the full 32-char hex UUID for both segments instead of [:12]
+    # prefix. Truncation to 12 chars risked cross-tenant prefix collision between
+    # authors whose UUIDs share the same time-based prefix. Full UUID guarantees
+    # tenant isolation. Both creator (embed_and_store) and reader (vector_store.py)
+    # call this function, so one change fixes the entire naming contract.
+    full_author = author_id.replace("-", "")
+    full_book = book_id.replace("-", "")
+    return f"a{full_author}_b{full_book}"
 
 
 async def embed_and_store(

app/services/faithfulness.py

@@ -4,8 +4,13 @@ Uses cross-encoder/nli-deberta-v3-small to verify the LLM response
 is grounded in the retrieved book text. Strips hallucinated claims.
 
 Per implementation plan — Step 9 of the RAG pipeline.
+
+BUG-5 fix: model loading and predict() are synchronous CPU-bound operations.
+Both are now wrapped in asyncio.to_thread() to avoid blocking the event loop.
 """
 
+import asyncio
+
 import structlog
 from sentence_transformers import CrossEncoder
 
@@ -50,18 +55,20 @@ async def check_faithfulness(
     threshold = threshold or cfg.RAG_FAITHFULNESS_THRESHOLD
 
     try:
-        model = _get_model()
+        # BUG-5 fix: _get_model() loads a ~100MB model on first call (disk I/O),
+        # and predict() is CPU-bound inference — both block the event loop.
+        # asyncio.to_thread() offloads them to the thread pool executor.
         context = " ".join(c.text for c in chunks[:3])[:2000]
 
-        # NLI: premise = context, hypothesis = response
-        score = model.predict([(context, response)])[0]
+        def _run_inference() -> float:
+            model = _get_model()
+            score = model.predict([(context, response)])[0]
+            # CrossEncoder NLI returns [contradiction, neutral, entailment]
+            if hasattr(score, "__len__") and len(score) == 3:
+                return float(score[2])
+            return float(score)
 
-        # CrossEncoder NLI returns [contradiction, neutral, entailment]
-        # score is array of 3; take entailment (index 2)
-        if hasattr(score, "__len__") and len(score) == 3:
-            entailment_score = float(score[2])
-        else:
-            entailment_score = float(score)
+        entailment_score = await asyncio.to_thread(_run_inference)
 
         is_faithful = entailment_score >= threshold
         logger.debug(

app/services/intent.py

@@ -120,7 +120,10 @@ _JAILBREAK_SIGNALS: tuple[str, ...] = (
     "what are your instructions", "your rules", "repeat your",
     "repeat the above", "repeat everything",
     "no restrictions", "no rules", "without restrictions",
-    "hypothetically speaking", "for a story", "in a fictional world",
+    # SEC-3 fix: removed "for a story" and "in a fictional world" — too broad,
+    # causing false positives on legitimate reader questions.
+    # The precise combined patterns in guardrails.py cover real attacks.
+    "hypothetically speaking",
 )
 
 _PIRACY_SIGNALS: tuple[str, ...] = (
@@ -350,7 +353,11 @@ async def classify_intent(
     )
 
     try:
-        client = AsyncOpenAI(api_key=cfg.OPENAI_API_KEY)
+        # BUG-1 fix: use the shared singleton client (same as helpers.py) instead of
+        # creating a new AsyncOpenAI() here — which spawned a new HTTP connection pool
+        # on every single Tier-2 intent classification call.
+        from app.services.pipeline.helpers import _get_openai_client
+        client = _get_openai_client()
         response = await client.chat.completions.create(
             model=cfg.OPENAI_CHAT_MODEL,     # gpt-4o-mini — cheap and accurate
             messages=[{"role": "user", "content": prompt}],

app/services/pipeline/helpers.py

@@ -167,7 +167,10 @@ def format_history(history: list[dict]) -> str:
     if not history:
         return "This is the start of the conversation."
     lines = []
-    for msg in history[-8:]:   # B6 fix: was -6 (= 3 turns), now -8 (= 4 turns)
+    # MINOR-6 fix: was hardcoded to -8 (4 turns), ignoring RAG_SESSION_HISTORY_TURNS config.
+    # Now uses the config value so the prompt window matches the session storage window.
+    max_messages = cfg.RAG_SESSION_HISTORY_TURNS * 2
+    for msg in history[-max_messages:]:
         role = "Visitor" if msg["role"] == "user" else "You"
         content = sanitize_history_line(msg.get("content", ""))[:200]  # cap per-line
         if content:
@@ -365,7 +368,10 @@ async def check_custom_qa(
 
     if best_match:
         best_match.match_count = (best_match.match_count or 0) + 1
-        await db.commit()
+        # MINOR-5 fix: was db.commit() which permanently writes the match_count increment
+        # even if the pipeline fails afterwards. db.flush() keeps it in the current
+        # transaction so it's rolled back cleanly on any downstream exception.
+        await db.flush()
         logger.info(
             "Custom Q&A match",
             qa_id=best_match.id,

app/services/reranker.py

@@ -4,8 +4,12 @@ Uses cross-encoder/ms-marco-MiniLM-L-6-v2 (free, local) to re-rank
 retrieved chunks by relevance to the original query.
 Significantly improves precision over cosine similarity alone.
 RULE: Keep top N chunks above minimum score threshold.
+
+BUG-6 fix: reranker.predict() is CPU-bound sync — wrapped in asyncio.to_thread().
 """
 
+import asyncio
+
 import structlog
 
 from app.config import get_settings
@@ -60,7 +64,9 @@ async def rerank_chunks(
 
         # Build (query, chunk) pairs for cross-encoder
         pairs = [(query, chunk.text) for chunk in chunks]
-        scores = reranker.predict(pairs)
+
+        # BUG-6 fix: predict() is synchronous CPU-bound inference — offload to thread pool.
+        scores = await asyncio.to_thread(reranker.predict, pairs)
 
         # Apply scores
         for chunk, score in zip(chunks, scores):

app/services/rewriter.py

@@ -13,7 +13,6 @@ RULE: Always include original query in the returned list (first element).
 import re
 
 import structlog
-from openai import AsyncOpenAI
 
 from app.config import get_settings
 
@@ -257,7 +256,10 @@ async def _llm_rewrite(query: str, history: list[dict]) -> str | None:
     user_msg = f"History:\n{history_str}\n\nQuery: {query}"
 
     try:
-        client = AsyncOpenAI(api_key=cfg.OPENAI_API_KEY)
+        # BUG-2 fix: use the shared singleton client instead of creating a new
+        # AsyncOpenAI() here — was spawning a new HTTP connection pool on every call.
+        from app.services.pipeline.helpers import _get_openai_client
+        client = _get_openai_client()
         response = await client.chat.completions.create(
             model=cfg.OPENAI_CHAT_MODEL,
             messages=[

app/services/superadmin_service.py

@@ -169,7 +169,9 @@ class SuperAdminService:
         try:
             from app.core.chroma_client import get_chroma_client
             chroma = get_chroma_client()
-            author_prefix = f"a{author_id.replace('-', '')[:12]}"
+            # SEC-4 fix: use the full 32-char hex UUID to match get_collection_name().
+            # The old [:12] prefix would miss collections under the updated naming contract.
+            author_prefix = f"a{author_id.replace('-', '')}"
             for col in chroma.list_collections():
                 col_name = col.name if hasattr(col, 'name') else str(col)
                 if col_name.startswith(author_prefix):

app/services/vector_store.py

@@ -134,7 +134,10 @@ async def _get_target_collections(
         logger.error("Failed to list ChromaDB collections", error=str(e))
         return []
 
-    author_prefix = author_id.replace("-", "")[:12]
+    # SEC-4 fix: use the full 32-char hex UUID instead of [:12] prefix —
+    # the prefix could collide between authors sharing similar UUID prefixes
+    # (common with time-based UUIDs). Full UUID guarantees tenant isolation.
+    author_prefix = author_id.replace("-", "")
     author_tag = f"a{author_prefix}"
 
     targets = []

app/tasks/link_health_task.py

@@ -65,8 +65,12 @@ async def _check_url(client: httpx.AsyncClient, url: str) -> bool:
     Returns:
         True if URL is reachable (2xx/3xx), False otherwise.
     """
-    # R-130: SSRF protection — block internal/private IPs before fetching
-    if _is_internal_url(url):
+    # R-130: SSRF protection — block internal/private IPs before fetching.
+    # BUG-7 fix: _is_internal_url() calls socket.gethostbyname() which is a
+    # blocking DNS call — offload to thread pool to avoid freezing the event loop.
+    import asyncio
+    is_internal = await asyncio.to_thread(_is_internal_url, url)
+    if is_internal:
         logger.warning("Link health: blocked internal URL", url=url[:100])
         return False
     try: