fix: resolve concurrency and caching bugs in API

main.py

@@ -10,13 +10,13 @@ The heavy resources (index + embedding model) are loaded ONCE at startup via
 FastAPI's lifespan context manager and shared across all requests.
 """
 
+import asyncio
 import os
 import time
 from contextlib import asynccontextmanager
-from functools import lru_cache
 
-import requests
-from fastapi import FastAPI, HTTPException, Request
+import httpx  # API-BUG-2: async HTTP client — replaces blocking requests
+from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel, Field
@@ -50,7 +50,10 @@ async def lifespan(app: FastAPI):
     """Load the RAG index at startup; release on shutdown."""
     print("Loading RAG index from GitHub…")
     t0 = time.time()
-    index, error = load_index()
+    # API-BUG-3: load_index() makes blocking HTTP calls (requests.get) and does
+    # heavy CPU work (embedding model init).  Running it in a thread pool keeps
+    # the async event loop from freezing during the 30-60 second startup.
+    index, error = await asyncio.to_thread(load_index)
     elapsed = round(time.time() - t0, 2)
 
     if error:
@@ -118,25 +121,41 @@ class RetrieveResponse(BaseModel):
 
 
 # ---------------------------------------------------------------------------
-# Manifest caching (simple in-memory, 5-minute TTL)
+# Manifest caching — async, 5-min TTL, mutex to prevent stampede
 # ---------------------------------------------------------------------------
 _manifest_cache: dict = {"data": None, "fetched_at": 0}
 MANIFEST_TTL = 300  # seconds
+_manifest_lock = asyncio.Lock()  # API-BUG-1: one coroutine fetches at a time
 
 
-def _get_manifest() -> dict:
-    now = time.time()
-    if _manifest_cache["data"] and (now - _manifest_cache["fetched_at"]) < MANIFEST_TTL:
-        return _manifest_cache["data"]
+async def _get_manifest() -> dict:
+    """
+    Fetch and in-memory cache manifest.json from GitHub.
+
+    asyncio.Lock (API-BUG-1): when the cache expires, only ONE coroutine
+    actually calls GitHub; all other concurrent waiters block on the lock and
+    then immediately return the freshly-written result — no stampede.
 
-    url = f"https://raw.githubusercontent.com/{MATERIALS_REPO}/{MANIFEST_BRANCH}/manifest.json"
-    resp = requests.get(url, timeout=15)
-    resp.raise_for_status()
-    data = resp.json()
+    httpx.AsyncClient (API-BUG-2): non-blocking HTTP so the event loop stays
+    responsive while the GitHub call is in flight.
+    """
+    async with _manifest_lock:
+        now = time.time()
+        if _manifest_cache["data"] and (now - _manifest_cache["fetched_at"]) < MANIFEST_TTL:
+            return _manifest_cache["data"]
 
-    _manifest_cache["data"] = data
-    _manifest_cache["fetched_at"] = now
-    return data
+        url = (
+            f"https://raw.githubusercontent.com/{MATERIALS_REPO}"
+            f"/{MANIFEST_BRANCH}/manifest.json"
+        )
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            resp = await client.get(url)
+            resp.raise_for_status()
+
+        data = resp.json()
+        _manifest_cache["data"] = data
+        _manifest_cache["fetched_at"] = now
+        return data
 
 
 # ---------------------------------------------------------------------------
@@ -156,22 +175,22 @@ def health():
 
 
 @app.get("/manifest")
-def get_manifest():
+async def get_manifest():
     """
     Proxy and cache the study materials manifest.json from GitHub.
     The Cloudflare Worker also caches this in KV — this is a double layer.
     """
     try:
-        data = _get_manifest()
+        data = await _get_manifest()
         return JSONResponse(content=data)
-    except requests.HTTPError as err:
+    except httpx.HTTPStatusError as err:
         raise HTTPException(status_code=502, detail=f"GitHub fetch failed: {err}")
     except Exception as err:
         raise HTTPException(status_code=500, detail=str(err))
 
 
 @app.post("/retrieve", response_model=RetrieveResponse)
-def retrieve(body: RetrieveRequest):
+async def retrieve(body: RetrieveRequest):
     """
     Core RAG endpoint.
 
@@ -179,10 +198,17 @@ def retrieve(body: RetrieveRequest):
     2. Searches the pre-built LlamaIndex vector store
     3. Filters results by semester + subject metadata
     4. Returns top-k chunks + a formatted context string for the LLM prompt
+
+    CPU-bound embedding + vector search runs in a thread pool via
+    asyncio.to_thread — the event loop is never blocked (API-BUG-3).
     """
     index = _state.get("index")
 
-    chunks = retrieve_chunks(
+    # API-BUG-3: retrieve_chunks is synchronous (HuggingFace embedding model
+    # + LlamaIndex vector search).  Run it in a thread so FastAPI can handle
+    # other requests concurrently while this one is working.
+    chunks = await asyncio.to_thread(
+        retrieve_chunks,
         index=index,
         query=body.query,
         semester=body.semester,

rag.py

@@ -11,8 +11,10 @@ Handles everything related to the LlamaIndex vector index:
 
 import io
 import os
+import shutil          # API-BUG-7: temp-dir cleanup after index is in memory
 import tempfile
 from pathlib import Path
+from typing import TypedDict  # API-BUG-4: explicit typed return from retrieve_chunks
 
 import requests
 
@@ -27,6 +29,20 @@ try:
 except ImportError:
     LLAMA_INDEX_AVAILABLE = False
 
+# API-BUG-5: MetadataFilters let the vector store do scope-filtering internally,
+# avoiding the over-fetch window that could miss relevant chunks.
+try:
+    from llama_index.core.vector_stores.types import MetadataFilter, MetadataFilters
+
+    METADATA_FILTERS_AVAILABLE = True
+except ImportError:
+    try:
+        from llama_index.core.vector_stores import MetadataFilter, MetadataFilters
+
+        METADATA_FILTERS_AVAILABLE = True
+    except ImportError:
+        METADATA_FILTERS_AVAILABLE = False
+
 try:
     import PyPDF2
 
@@ -53,7 +69,17 @@ DEFAULT_TOP_K = 5
 
 
 # ---------------------------------------------------------------------------
-# Index loading (called once at FastAPI startup)
+# Typed chunk return (API-BUG-4)
+# ---------------------------------------------------------------------------
+class ChunkDict(TypedDict):
+    text: str
+    score: float | None
+    filename: str | None
+    subject: str | None
+
+
+# ---------------------------------------------------------------------------
+# Index loading (called once at FastAPI startup via asyncio.to_thread)
 # ---------------------------------------------------------------------------
 
 def load_index():
@@ -62,6 +88,10 @@ def load_index():
     VectorStoreIndex ready for querying.
 
     Returns (index, error_msg). index is None if loading failed.
+
+    API-BUG-7: The temp directory is always removed in the finally block.
+    After load_index_from_storage() returns, all data is in memory — the
+    files on disk are no longer needed.
     """
     if not LLAMA_INDEX_AVAILABLE:
         return None, "llama-index-core is not installed."
@@ -71,33 +101,33 @@ def load_index():
     )
     index_dir = tempfile.mkdtemp(prefix="plexi_index_")
 
-    for filename in INDEX_FILES:
-        url = f"{index_base_url}/{filename}"
-        try:
-            resp = requests.get(url, timeout=30)
-            resp.raise_for_status()
-            with open(os.path.join(index_dir, filename), "wb") as fh:
-                fh.write(resp.content)
-        except Exception as err:
-            return None, f"Failed to download index file '{filename}': {err}"
-
     try:
-        embed_model = HuggingFaceEmbedding(model_name=EMBED_MODEL_ID)
-        Settings.embed_model = embed_model
-        Settings.llm = None
-
-        storage_ctx = StorageContext.from_defaults(persist_dir=index_dir)
-        index = load_index_from_storage(storage_ctx)
-        return index, None
-    except Exception as err:
-        return None, f"Failed to load index from storage: {err}"
+        # Download each index shard from GitHub
+        for filename in INDEX_FILES:
+            url = f"{index_base_url}/{filename}"
+            try:
+                resp = requests.get(url, timeout=30)
+                resp.raise_for_status()
+                with open(os.path.join(index_dir, filename), "wb") as fh:
+                    fh.write(resp.content)
+            except Exception as err:
+                return None, f"Failed to download index file '{filename}': {err}"
+
+        # Build the in-memory index from the downloaded shards
+        try:
+            embed_model = HuggingFaceEmbedding(model_name=EMBED_MODEL_ID)
+            Settings.embed_model = embed_model
+            Settings.llm = None
 
+            storage_ctx = StorageContext.from_defaults(persist_dir=index_dir)
+            index = load_index_from_storage(storage_ctx)
+            return index, None
+        except Exception as err:
+            return None, f"Failed to load index from storage: {err}"
 
-def load_embed_model():
-    """Load and return the HuggingFace embedding model (for health checks)."""
-    if not LLAMA_INDEX_AVAILABLE:
-        return None
-    return HuggingFaceEmbedding(model_name=EMBED_MODEL_ID)
+    finally:
+        # API-BUG-7: always wipe the temp dir — data is now in RAM.
+        shutil.rmtree(index_dir, ignore_errors=True)
 
 
 # ---------------------------------------------------------------------------
@@ -119,32 +149,49 @@ def retrieve_chunks(
     semester: str,
     subject: str,
     top_k: int = DEFAULT_TOP_K,
-) -> list[dict]:
+) -> list[ChunkDict]:
     """
     Embed the query, retrieve top-k chunks from the index scoped to the
     given semester + subject.
 
-    Returns a list of dicts:
-        { text, score, filename, subject }
+    API-BUG-4: Returns list[ChunkDict] (TypedDict) instead of list[dict] so
+    type errors surface at the source rather than silently at Pydantic
+    serialization time.
+
+    API-BUG-5: Primary path uses MetadataFilters so the vector store does the
+    scope-gating internally — no risk of the over-fetch window failing to reach
+    chunks that belong to the active subject.  Falls back to the generous
+    over-fetch + manual filter approach when MetadataFilters are unavailable
+    (e.g., older llama-index-core builds).
     """
     if index is None:
         return []
 
     try:
-        # Fetch more than needed so we have room to filter by scope
-        retriever = index.as_retriever(similarity_top_k=max(top_k * 5, 10))
-        nodes = retriever.retrieve(query)
-
-        scoped = [n for n in nodes if _matches_scope(n, semester, subject)]
+        if METADATA_FILTERS_AVAILABLE:
+            # Primary: vector store filters by metadata at query time.
+            filters = MetadataFilters(
+                filters=[
+                    MetadataFilter(key="semester", value=semester),
+                    MetadataFilter(key="subject", value=subject),
+                ]
+            )
+            retriever = index.as_retriever(similarity_top_k=top_k, filters=filters)
+            nodes = retriever.retrieve(query)
+        else:
+            # Fallback: over-fetch (generous 10× window) + manual scope filter.
+            retriever = index.as_retriever(similarity_top_k=max(top_k * 10, 50))
+            nodes = retriever.retrieve(query)
+            nodes = [n for n in nodes if _matches_scope(n, semester, subject)]
 
         return [
-            {
-                "text": node.node.get_content(),
-                "score": round(float(node.score), 4) if node.score is not None else None,
-                "filename": (getattr(node.node, "metadata", {}) or {}).get("filename"),
-                "subject": (getattr(node.node, "metadata", {}) or {}).get("subject"),
-            }
-            for node in scoped[:top_k]
+            ChunkDict(
+                text=node.node.get_content(),
+                score=round(float(node.score), 4) if node.score is not None else None,
+                filename=(getattr(node.node, "metadata", {}) or {}).get("filename"),
+                subject=(getattr(node.node, "metadata", {}) or {}).get("subject"),
+            )
+            for node in nodes[:top_k]
         ]
     except Exception as err:
         print(f"Retrieval error: {err}")
@@ -155,7 +202,7 @@ def retrieve_chunks(
 # Context formatting (for system prompt injection)
 # ---------------------------------------------------------------------------
 
-def format_context(chunks: list[dict]) -> str:
+def format_context(chunks: list[ChunkDict]) -> str:
     """Format retrieved chunks as a numbered block for the LLM system prompt."""
     if not chunks:
         return "(No relevant context retrieved for this query.)"
@@ -170,7 +217,9 @@ def format_context(chunks: list[dict]) -> str:
 
 
 # ---------------------------------------------------------------------------
-# PDF text extraction (used for full-context fallback loading)
+# PDF text extraction
+# NOTE: read_pdf_text() is currently unused by the API routes — retained for
+# future use (CQ-4). load_embed_model() removed — was dead code (API-BUG-6).
 # ---------------------------------------------------------------------------
 
 def read_pdf_text(pdf_bytes: bytes) -> str:

requirements.txt

@@ -2,6 +2,7 @@ fastapi>=0.115.0,<1.0.0
 uvicorn[standard]>=0.30.0,<1.0.0
 pydantic>=2.0.0,<3.0.0
 requests>=2.31.0,<3.0.0
+httpx>=0.27.0,<1.0.0
 python-dotenv>=1.0.0
 PyPDF2>=3.0.0,<4.0.0
 llama-index-core>=0.11.0,<0.13.0