Deploy 2b3655ac from GitHub Actions

app/agents/base.py

@@ -39,7 +39,43 @@ _MAX_RETRY_DELAY = 120.0  # seconds — cap for retry_delay parsed from API resp
 
 
 class QuotaExceededError(RuntimeError):
-    """Raised when the Gemini API quota / rate-limit is exhausted."""
+    """
+    Raised when a provider refuses the request for quota reasons.
+
+    Covers rate limits (429), exhausted free-tier grants (RESOURCE_EXHAUSTED)
+    and billing exhaustion (402 payment_required) alike: from the caller's
+    point of view all three mean "this provider will not serve us right now,
+    and retrying the same call immediately will not help".
+    """
+
+    def __init__(self, message: str, provider: str = "", agent: str = ""):
+        super().__init__(message)
+        self.provider = provider
+        self.agent = agent
+
+
+# Substrings that identify a quota/billing refusal in a provider error message.
+# 402 matters as much as 429 here: Cerebras answers an exhausted account with
+# `Error code: 402 ... payment_required_error`, which the old 429-only check
+# classified as a transient error and then slept through three pointless
+# retries before returning an empty dict.
+_QUOTA_MARKERS = (
+    "429",
+    "rate_limit",
+    "rate limit",
+    "quota",
+    "resource_exhausted",
+    "error code: 402",
+    "payment_required",
+    "insufficient_quota",
+    "billing",
+)
+
+
+def _is_quota_error(err_str: str) -> bool:
+    """True when a provider error message describes a quota/billing refusal."""
+    lowered = err_str.lower()
+    return any(marker in lowered for marker in _QUOTA_MARKERS)
 
 
 def _parse_retry_delay(err_str: str) -> Optional[float]:
@@ -210,12 +246,7 @@ class BaseAgent(ABC):
                 last_error = e
                 err_str = str(e)
 
-                is_quota = (
-                    "429" in err_str
-                    or "rate_limit" in err_str.lower()
-                    or "quota" in err_str.lower()
-                    or "RESOURCE_EXHAUSTED" in err_str
-                )
+                is_quota = _is_quota_error(err_str)
 
                 if is_quota:
                     # Try fallback provider if available and not already using it
@@ -252,8 +283,10 @@ class BaseAgent(ABC):
 
                     logger.error(f"{self.name}: All providers exhausted — {e}")
                     raise QuotaExceededError(
-                        f"Rate limit exceeded for {self.name}. "
-                        "Please try again later."
+                        f"{self.provider} quota exhausted for {self.name} "
+                        f"({err_str[:200]})",
+                        provider=self.provider,
+                        agent=self.name,
                     ) from e
 
                 # Transient non-quota error — exponential backoff

app/agents/compliance_auditor.py

@@ -3,7 +3,7 @@ Compliance Auditor Agent
 Cross-references a regulation clause against operational document evidence
 to determine compliance status (compliant / gap / missing).
 
-Uses Gemini for the nuanced cross-referencing task that requires reasoning
+Runs on Groq for the nuanced cross-referencing task that requires reasoning
 across multiple evidence chunks.
 """
 
@@ -107,7 +107,7 @@ Always cite specific evidence in your assessment. If no evidence is provided, ma
         prompt = (
             "Assess compliance for the following regulation clause against "
             "the provided operational document evidence.\n\n"
-            f"REGULATION CLAUSE{text}:\n{text}\n\n"
+            f"REGULATION CLAUSE:\n{text}\n\n"
             f"{section_hint}\n\n"
             f"EVIDENCE FROM OPERATIONAL DOCUMENTS:\n{evidence_text}\n\n"
             "Respond with a JSON object:\n"

app/agents/entity_extractor.py

@@ -26,7 +26,18 @@ class EntityExtractorAgent(BaseAgent):
 
     def __init__(self):
         # Cerebras: 1M tokens/day free, 2600+ TPS, 60K TPM — better for high-volume extraction than Groq
-        super().__init__(model_name="gpt-oss-120b", provider="cerebras")
+        super().__init__(
+            model_name="gpt-oss-120b",
+            provider="cerebras",
+            # Cerebras answers an exhausted account with 402
+            # payment_required, which took this agent out entirely while
+            # Groq was still serving. Groq hosts the same gpt-oss-120b
+            # weights, so the fallback is the same model on another host
+            # rather than a downgrade. It is only ever used after a quota
+            # refusal, so it costs nothing on the happy path.
+            fallback_provider="groq",
+            fallback_model="openai/gpt-oss-120b",
+        )
 
     @property
     def system_prompt(self) -> str:
@@ -108,6 +119,22 @@ Notes:
 """
         result = await self._generate_json(prompt)
 
+        # An exhausted _generate_json returns {}, which used to be laundered
+        # into six empty lists — a document with no entities and a document the
+        # extractor never managed to read looked identical. Mark the failure.
+        if not result:
+            return {
+                "equipment": [],
+                "chemicals": [],
+                "locations": [],
+                "personnel": [],
+                "dates": [],
+                "regulations": [],
+                "entity_count": 0,
+                "status": "error",
+                "error": "entity extractor provider returned no usable response",
+            }
+
         entities = {
             "equipment": self._deduplicate(result.get("equipment", [])),
             "chemicals": self._deduplicate(result.get("chemicals", [])),

app/agents/orchestrator.py

@@ -162,22 +162,47 @@ class AgentOrchestrator:
                 (datetime.now(timezone.utc) - t1).total_seconds() * 1000
             )
 
-            # Handle per-agent exceptions gracefully
+            # Handle per-agent exceptions gracefully.
+            #
+            # gather(return_exceptions=True) means one agent's provider dying
+            # never costs the other two: whatever classification, safety and
+            # entity work succeeded is still returned and still persisted. The
+            # failed section is replaced by a placeholder that says so.
             def _quota_error_result(agent_name: str, exc: Exception) -> dict:
                 is_quota = isinstance(exc, QuotaExceededError)
+                provider = getattr(exc, "provider", "") or "the provider"
                 logger.error(f"{agent_name} failed: {exc}")
                 return {
                     "error": str(exc),
+                    "provider": getattr(exc, "provider", ""),
                     "quota_exceeded": is_quota,
                     "status": "quota_exceeded" if is_quota else "error",
+                    "unavailable_reason": (
+                        f"{provider} quota exhausted"
+                        if is_quota
+                        else f"{provider} error"
+                    ),
                 }
 
+            def _unavailable_text(agent_label: str, exc: Exception) -> str:
+                if isinstance(exc, QuotaExceededError):
+                    provider = getattr(exc, "provider", "") or "the AI provider"
+                    return (
+                        f"{agent_label} unavailable — the {provider} account has "
+                        "no quota left for this request. Click Re-analyze once "
+                        "quota is restored."
+                    )
+                return (
+                    f"{agent_label} unavailable — the AI provider failed for this "
+                    f"request ({str(exc)[:150]}). Click Re-analyze to try again."
+                )
+
             if isinstance(safety, Exception):
                 safety = {
                     **_quota_error_result("SafetyAnalyzerAgent", safety),
                     "score": None,
                     "hazards": [],
-                    "recommendations": [],
+                    "recommendations": [_unavailable_text("Safety analysis", safety)],
                 }
 
             if isinstance(entities, Exception):
@@ -194,10 +219,31 @@ class AgentOrchestrator:
             if isinstance(summary, Exception):
                 summary = {
                     **_quota_error_result("SummarizerAgent", summary),
-                    "summary": "Analysis failed — the summarizer's provider rate-limited this request. Please try again later.",
+                    "summary": _unavailable_text("Summary", summary),
                     "key_points": [],
                 }
 
+            # Sections that produced a placeholder rather than real analysis.
+            # document_service surfaces this to the user through
+            # Document.processing_error, because the API response coerces
+            # entities to plain lists and would otherwise drop every marker.
+            degraded_sections = {
+                name: section.get("unavailable_reason")
+                or section.get("error")
+                or "unavailable"
+                for name, section in (
+                    ("safety", safety),
+                    ("entities", entities),
+                    ("summary", summary),
+                )
+                if isinstance(section, dict)
+                and (
+                    section.get("error")
+                    or section.get("quota_exceeded")
+                    or section.get("status") in ("error", "quota_exceeded")
+                )
+            }
+
             total_ms = int(
                 (datetime.now(timezone.utc) - start_time).total_seconds() * 1000
             )
@@ -219,6 +265,7 @@ class AgentOrchestrator:
                     ],
                     "analyzed_at": datetime.now(timezone.utc).isoformat(),
                     "cache_hit": False,
+                    "degraded_sections": degraded_sections,
                 },
             }
 

app/agents/summarizer.py

@@ -23,7 +23,18 @@ class SummarizerAgent(BaseAgent):
 
     def __init__(self):
         # Shift to Cerebras (gpt-oss-120b) to bypass Gemini API rate limits and avoid Groq parallel execution rate limits
-        super().__init__(model_name="gpt-oss-120b", provider="cerebras")
+        super().__init__(
+            model_name="gpt-oss-120b",
+            provider="cerebras",
+            # Cerebras answers an exhausted account with 402
+            # payment_required, which took this agent out entirely while
+            # Groq was still serving. Groq hosts the same gpt-oss-120b
+            # weights, so the fallback is the same model on another host
+            # rather than a downgrade. It is only ever used after a quota
+            # refusal, so it costs nothing on the happy path.
+            fallback_provider="groq",
+            fallback_model="openai/gpt-oss-120b",
+        )
 
     @property
     def system_prompt(self) -> str:
@@ -89,6 +100,28 @@ Respond with a JSON object:
 """
         result = await self._generate_json(prompt)
 
+        # _generate_json returns {} once it has exhausted its retries on a
+        # non-quota failure. Falling through to "Summary not available." made
+        # that indistinguishable from a successful run over an empty document,
+        # and it was stored with confidence 0.7. Say plainly that nothing was
+        # produced, and carry an error marker so the orchestrator, the analysis
+        # cache and document_service can all see it.
+        if not result:
+            message = (
+                "Summary unavailable — the summarization provider returned no "
+                "usable response after retries. Re-analyze to try again."
+            )
+            return {
+                "summary": message,
+                "key_points": [],
+                "action_items": [],
+                "document_purpose": "",
+                "confidence": 0.0,
+                "word_count": 0,
+                "status": "error",
+                "error": "summarizer provider returned no usable response",
+            }
+
         summary = result.get("summary") or "Summary not available."
         return {
             "summary": summary,

app/api/v1/chat.py

@@ -257,23 +257,28 @@ async def send_message(
         db.commit()
         db.refresh(session)
 
+    # Generate AI response with RAG
+    from app.services.chat_service import CHAT_MODEL, ChatService
+
+    chat_service = ChatService()
+
+    # Load prior turns BEFORE staging this one. The add() below is uncommitted,
+    # but autoflush would still surface it to the history query.
+    history = chat_service.load_session_history(db, session.id)
+
     # Save user message (but don't commit yet - wait for successful response)
     user_message = ChatMessage(
         session_id=session.id, role="user", content=request.content
     )
     db.add(user_message)
 
-    # Generate AI response with RAG
-    from app.services.chat_service import CHAT_MODEL, ChatService
-
-    chat_service = ChatService()
-
     try:
         ai_response, sources, tokens_used = await chat_service.generate_response(
             query=validated_query,
             user_id=user_id,
             document_ids=request.document_ids,
             db=db,
+            history=history,
         )
 
         # Calculate response time

app/api/v1/chat_stream.py

@@ -78,6 +78,12 @@ async def stream_chat(
         db.commit()
         db.refresh(session)
 
+    chat_service = ChatService()
+
+    # Load prior turns BEFORE persisting this one, or the current question is
+    # replayed to the model as its own history.
+    history = chat_service.load_session_history(db, session.id)
+
     # Save user message immediately
     user_message = ChatMessage(
         session_id=session.id,
@@ -87,8 +93,6 @@ async def stream_chat(
     db.add(user_message)
     db.commit()
 
-    chat_service = ChatService()
-
     async def event_generator():
         full_response = []
         sources = []
@@ -100,6 +104,7 @@ async def stream_chat(
                 user_id=user_id,
                 document_ids=request.document_ids,
                 db=db,
+                history=history,
             ):
                 # Forward each SSE event to client
                 yield event

app/api/v1/health.py

@@ -3,9 +3,12 @@ Health Check Endpoints
 System health and status monitoring
 """
 
+import asyncio
+import time
 from datetime import datetime
+from typing import Any, Dict
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Response
 from sqlalchemy.orm import Session
 
 from app.config import settings
@@ -14,6 +17,136 @@ from app.schemas.common import HealthResponse
 
 router = APIRouter()
 
+# ── Provider liveness ─────────────────────────────────────────────────────────
+#
+# /health used to report the AI subsystem as "configured" purely because API
+# keys were present. That is not a health check: in production Cerebras was
+# returning 402 payment_required on every call while /health still said
+# "healthy", and only /health/providers — which nothing polls — knew.
+#
+# So both endpoints now share the real ping below. The constraints that kept
+# the pings out of /health are handled rather than avoided:
+#
+#   * Cost — the result is memoised in-process for _PROVIDER_CACHE_TTL seconds,
+#     so the keepalive cron (every 30 min) and any load-balancer probe cost at
+#     most one ping per provider per minute, and bursts of probes cost nothing.
+#   * Latency — probes run concurrently under a hard timeout; a hung provider
+#     is reported as failed instead of hanging the endpoint.
+#
+# The cheap static probe used by the Dockerfile HEALTHCHECK is the plain
+# /health in app/main.py. It stays free of network calls on purpose.
+
+_PROVIDER_PING_TIMEOUT = 4.0  # seconds — per provider, hard ceiling
+_PROVIDER_CACHE_TTL = 60.0  # seconds — memoise the whole result set
+
+_provider_cache: Dict[str, Any] = {"checked_at_monotonic": None, "result": None}
+_provider_lock = asyncio.Lock()
+
+
+async def _probe(name: str, client, model: str, api_key: str) -> Dict[str, Any]:
+    """
+    Call one provider with a 1-token request and report what happened.
+
+    Never raises: a probe failure is a result, not an error.
+    """
+    if not api_key:
+        return {"model": model, "ok": False, "error": "API key not configured"}
+
+    try:
+        resp = await asyncio.wait_for(
+            client.chat.completions.create(
+                model=model,
+                messages=[{"role": "user", "content": "ping"}],
+                max_tokens=1,
+            ),
+            timeout=_PROVIDER_PING_TIMEOUT,
+        )
+        return {
+            "model": model,
+            "ok": True,
+            "finish_reason": resp.choices[0].finish_reason,
+        }
+    except asyncio.TimeoutError:
+        # A provider that cannot answer a 1-token request inside the timeout
+        # cannot serve a document analysis either. Treat it as down.
+        return {
+            "model": model,
+            "ok": False,
+            "error": f"timed out after {_PROVIDER_PING_TIMEOUT}s",
+        }
+    except Exception as e:
+        # The message carries the useful part — an unknown model id reads
+        # very differently from an auth failure, a 402 or a rate limit.
+        return {"model": model, "ok": False, "error": str(e)[:300]}
+
+
+async def check_providers(use_cache: bool = True) -> Dict[str, Any]:
+    """
+    Liveness of every LLM provider this backend actually calls.
+
+    Returns ``{"status": "ok"|"degraded", "checked_at": iso8601,
+    "providers": {name: {"model", "ok", ...}}, "cached": bool}``.
+
+    Shared by /health and /health/providers so there is exactly one definition
+    of "is the AI working".
+    """
+    from app.services.chat_service import CHAT_MODEL
+    from app.services.llm_provider import get_cerebras_client, get_groq_client
+
+    now = time.monotonic()
+    if use_cache:
+        checked_at = _provider_cache["checked_at_monotonic"]
+        if checked_at is not None and (now - checked_at) < _PROVIDER_CACHE_TTL:
+            return {**_provider_cache["result"], "cached": True}
+
+    async with _provider_lock:
+        # Re-check under the lock: while we waited, a concurrent probe may
+        # have filled the cache. Without this, a burst of simultaneous probes
+        # would each fire their own round of provider calls.
+        now = time.monotonic()
+        if use_cache:
+            checked_at = _provider_cache["checked_at_monotonic"]
+            if checked_at is not None and (now - checked_at) < _PROVIDER_CACHE_TTL:
+                return {**_provider_cache["result"], "cached": True}
+
+        targets = [
+            ("groq", get_groq_client(), CHAT_MODEL, settings.GROQ_API_KEY),
+            (
+                "cerebras",
+                get_cerebras_client(),
+                "gpt-oss-120b",
+                settings.CEREBRAS_API_KEY,
+            ),
+        ]
+
+        probed = await asyncio.gather(*(_probe(*t) for t in targets))
+        results = {name: outcome for (name, *_), outcome in zip(targets, probed)}
+
+        result = {
+            "status": "ok" if all(r["ok"] for r in results.values()) else "degraded",
+            "checked_at": datetime.utcnow().isoformat(),
+            "providers": results,
+        }
+        _provider_cache["result"] = result
+        _provider_cache["checked_at_monotonic"] = time.monotonic()
+
+    return {**result, "cached": False}
+
+
+def _summarise_providers(provider_result: Dict[str, Any]) -> str:
+    """One-line, human-readable rendering of the provider probe for services.ai."""
+    failed = [
+        f"{name} ({info.get('error', 'unknown error')})"
+        for name, info in provider_result["providers"].items()
+        if not info["ok"]
+    ]
+    if not failed:
+        return "healthy"
+    return "degraded: " + "; ".join(failed)
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────────────
+
 
 @router.get("/", response_model=HealthResponse)
 async def root():
@@ -27,19 +160,37 @@ async def root():
 
 
 @router.get("/health", response_model=HealthResponse)
-async def health_check(db: Session = Depends(get_db)):
+async def health_check(response: Response, db: Session = Depends(get_db)):
     """
     Detailed health check with service status.
-    Checks database connectivity and other services.
+
+    Checks the database, the optional Redis cache, and — for real, not by
+    inspecting environment variables — whether the LLM providers answer.
+
+    Status semantics:
+      * ``healthy``   — database up, every provider answering.
+      * ``degraded``  — database up, at least one provider failing. Still 200:
+        the keepalive cron and load-balancer probes treat non-200 as "restart
+        the Space", and a provider quota problem is not fixed by a restart.
+      * ``unhealthy`` — database unreachable. Returns 503.
+
+    Redis being ``not_configured`` never changes the overall status. It is an
+    optional cache and is deliberately unset in production.
     """
-    services = {"database": "unknown", "redis": "unknown", "ai": "unknown"}
+    services: Dict[str, Any] = {
+        "database": "unknown",
+        "redis": "unknown",
+        "ai": "unknown",
+    }
 
     # Check database
+    db_ok = False
     try:
         from sqlalchemy import text
 
         db.execute(text("SELECT 1"))
         services["database"] = "healthy"
+        db_ok = True
     except Exception as e:
         services["database"] = f"unhealthy: {str(e)}"
 
@@ -57,37 +208,20 @@ async def health_check(db: Session = Depends(get_db)):
         if r is not None:
             r.close()
 
-    # AI providers.
-    #
-    # This is deliberately a configuration check, not a liveness check. The
-    # previous version called genai.configure() and reported "healthy", which
-    # proved nothing: configure() makes no network call and validates no key,
-    # so it reported healthy with a garbage key and never touched Groq,
-    # Cerebras or Mistral at all. A model this backend could no longer call
-    # would still have shown green.
-    #
-    # It stays shallow because the keepalive workflow pings this endpoint on a
-    # schedule and Groq's free tier allows only 8K tokens/minute — burning that
-    # on health checks would starve real traffic. Use /health/providers to
-    # actually call the providers.
-    missing = [
-        name
-        for name, key in (
-            ("gemini", settings.GEMINI_API_KEY),
-            ("groq", settings.GROQ_API_KEY),
-            ("cerebras", settings.CEREBRAS_API_KEY),
-            ("mistral", settings.MISTRAL_API_KEY),
-        )
-        if not key
-    ]
-    services["ai"] = (
-        "configured" if not missing else f"missing keys: {','.join(missing)}"
-    )
-
-    # Overall status
-    overall = "healthy"
-    if services["database"] != "healthy":
+    # AI providers — a real 1-token call per provider, memoised for 60s and
+    # bounded by a timeout so this endpoint stays fast and cheap.
+    provider_result = await check_providers()
+    services["ai"] = _summarise_providers(provider_result)
+    services["ai_providers"] = provider_result["providers"]
+    services["ai_checked_at"] = provider_result["checked_at"]
+
+    if not db_ok:
+        overall = "unhealthy"
+        response.status_code = 503
+    elif provider_result["status"] != "ok":
         overall = "degraded"
+    else:
+        overall = "healthy"
 
     return HealthResponse(
         status=overall,
@@ -99,52 +233,19 @@ async def health_check(db: Session = Depends(get_db)):
 
 
 @router.get("/health/providers")
-async def provider_check():
+async def provider_check(fresh: bool = False):
     """
     Actually call each LLM provider with a minimal request.
 
-    /health only reports whether keys are present. This endpoint answers the
-    question that matters after a model migration: does the model id this
-    backend is configured with still exist and still serve us?
+    This answers the question that matters after a model migration: does the
+    model id this backend is configured with still exist and still serve us?
 
     That gap is not theoretical. Groq decommissioned llama-3.3-70b-versatile
     while it was hardcoded on both chat paths, and nothing — not the health
     check, not the test suite, which mocks every provider — could tell the
     difference between a working model and a retired one.
 
-    Not wired into /health on purpose: this spends real tokens, and the
-    keepalive workflow polls /health on a schedule. Call it by hand after a
-    deploy or a model change.
+    Shares its result with /health through a 60s in-process cache. Pass
+    ``?fresh=true`` to bypass the cache and force a live round of pings.
     """
-    from app.services.chat_service import CHAT_MODEL
-
-    results: dict = {}
-
-    async def _probe(name: str, client, model: str) -> None:
-        try:
-            resp = await client.chat.completions.create(
-                model=model,
-                messages=[{"role": "user", "content": "ping"}],
-                max_tokens=1,
-            )
-            results[name] = {
-                "model": model,
-                "ok": True,
-                "finish_reason": resp.choices[0].finish_reason,
-            }
-        except Exception as e:
-            # The message carries the useful part — an unknown model id reads
-            # very differently from an auth failure or a rate limit.
-            results[name] = {"model": model, "ok": False, "error": str(e)[:300]}
-
-    from app.services.llm_provider import get_cerebras_client, get_groq_client
-
-    await _probe("groq", get_groq_client(), CHAT_MODEL)
-    await _probe("cerebras", get_cerebras_client(), "gpt-oss-120b")
-
-    all_ok = all(r["ok"] for r in results.values())
-    return {
-        "status": "ok" if all_ok else "degraded",
-        "checked_at": datetime.utcnow().isoformat(),
-        "providers": results,
-    }
+    return await check_providers(use_cache=not fresh)

app/api/v1/search.py

@@ -31,13 +31,22 @@ genai.configure(api_key=settings.GEMINI_API_KEY)
 
 
 async def _get_query_embedding(text_input: str) -> List[float]:
-    """Generate query embedding using Gemini text-embedding-004."""
+    """Generate a query embedding using the configured Gemini embedding model.
+
+    output_dimensionality=768 is REQUIRED, not optional. The chunk_embedding
+    column is Vector(768); gemini-embedding-001 returns 3072 dimensions by
+    default. Omitting it produces a vector the pgvector cast rejects, and
+    hybrid_search swallows that failure — so the semantic arm silently
+    disappears and this endpoint degrades to lexical-only results with no
+    error anywhere. Keep this in sync with chat_service and document_service.
+    """
     try:
         result = await asyncio.to_thread(
             genai.embed_content,
             model=settings.EMBEDDING_MODEL,
             content=text_input,
             task_type="retrieval_query",
+            output_dimensionality=768,
         )
         return result["embedding"]
     except Exception as e:

app/config.py

@@ -226,9 +226,23 @@ class Settings(BaseSettings):
         "an uncapped 500-page scan would occupy a worker for 20 minutes.",
     )
 
-    # Mining AI Settings
-    SAFETY_SCORE_THRESHOLD: float = Field(default=70.0)
-    MAX_EMBEDDINGS_PER_QUERY: int = Field(default=5)
+    # Conversation Memory
+    # Prior turns are replayed to the model so follow-ups ("what about
+    # surface mines?") resolve against what was already discussed. Bounded on
+    # two axes because either one alone fails: a turn cap alone lets a few
+    # long answers blow the context window, and a char budget alone can slice
+    # the history mid-conversation in a way that strands a user turn without
+    # its answer.
+    CHAT_HISTORY_MAX_TURNS: int = Field(
+        default=6,
+        description="Max prior messages (user+assistant) replayed to the LLM. "
+        "Counted in whole user/assistant pairs, most recent first.",
+    )
+    CHAT_HISTORY_MAX_CHARS: int = Field(
+        default=6000,
+        description="Approximate char budget for replayed history. Roughly "
+        "1.5K tokens, leaving room for retrieved context and the answer.",
+    )
 
     # RAG Pipeline — Production Retrieval
     RERANK_MODEL: str = Field(

app/services/analysis_cache.py

@@ -73,6 +73,12 @@ def _is_cacheable(results: Dict[str, Any]) -> bool:
     if results.get("metadata", {}).get("failed"):
         return False
 
+    # The orchestrator lists every section it had to replace with a
+    # placeholder. One entry here means the analysis is partial, whatever the
+    # individual sections look like.
+    if results.get("metadata", {}).get("degraded_sections"):
+        return False
+
     for section in _REQUIRED_SECTIONS:
         value = results.get(section)
         if not isinstance(value, dict):

app/services/chat_service.py

@@ -108,11 +108,13 @@ class ChatService:
         user_id: str,
         document_ids: Optional[List[str]] = None,
         db: Session = None,
-    ) -> Tuple[str, List[Dict[str, Any]]]:
+        history: Optional[List[Dict[str, str]]] = None,
+    ) -> Tuple[str, List[Dict[str, Any]], Optional[Dict[str, int]]]:
         try:
-            query_embedding = await self._get_embedding(query)
+            retrieval_query = self._build_retrieval_query(query, history or [])
+            query_embedding = await self._get_embedding(retrieval_query)
             relevant_chunks = await self._retrieve_chunks(
-                query=query,
+                query=retrieval_query,
                 query_embedding=query_embedding,
                 user_id=user_id,
                 document_ids=document_ids,
@@ -127,13 +129,7 @@ class ChatService:
 
             response = await client.chat.completions.create(
                 model=CHAT_MODEL,
-                messages=[
-                    {"role": "system", "content": _SYSTEM_PROMPT},
-                    {
-                        "role": "user",
-                        "content": self._build_user_message(query, context),
-                    },
-                ],
+                messages=self._build_messages(query, context, history),
             )
             answer = response.choices[0].message.content
 
@@ -164,11 +160,13 @@ class ChatService:
         user_id: str,
         document_ids: Optional[List[str]] = None,
         db: Session = None,
+        history: Optional[List[Dict[str, str]]] = None,
     ) -> AsyncGenerator[str, None]:
         try:
-            query_embedding = await self._get_embedding(query)
+            retrieval_query = self._build_retrieval_query(query, history or [])
+            query_embedding = await self._get_embedding(retrieval_query)
             relevant_chunks = await self._retrieve_chunks(
-                query=query,
+                query=retrieval_query,
                 query_embedding=query_embedding,
                 user_id=user_id,
                 document_ids=document_ids,
@@ -186,13 +184,7 @@ class ChatService:
 
             response_stream = await client.chat.completions.create(
                 model=CHAT_MODEL,
-                messages=[
-                    {"role": "system", "content": _SYSTEM_PROMPT},
-                    {
-                        "role": "user",
-                        "content": self._build_user_message(query, context),
-                    },
-                ],
+                messages=self._build_messages(query, context, history),
                 stream=True,
             )
 
@@ -207,6 +199,142 @@ class ChatService:
             logger.error(f"Stream generation error: {e}", exc_info=True)
             yield f"event: error\ndata: {json.dumps({'message': str(e)})}\n\n"
 
+    # ── Conversation Memory ───────────────────────────────────────────────────
+
+    @staticmethod
+    def load_session_history(
+        db: Session,
+        session_id: Any,
+        max_turns: Optional[int] = None,
+        max_chars: Optional[int] = None,
+    ) -> List[Dict[str, str]]:
+        """Load recent prior turns for a session, oldest-first, ready to replay.
+
+        Call this BEFORE persisting the incoming user message. Both chat
+        endpoints add the new user row to the session early (one commits it
+        immediately), and SQLAlchemy autoflush would otherwise pull that row
+        into this query — the model would see the current question twice, once
+        as history and once as the live turn.
+
+        Bounded by turn count and an approximate char budget. Trimming walks
+        backwards from the newest message and stops on the first message that
+        would breach the budget, so the retained window is always a contiguous
+        recent slice rather than a sampling of the conversation.
+        """
+        from app.models.chat import ChatMessage
+
+        max_turns = max_turns or settings.CHAT_HISTORY_MAX_TURNS
+        max_chars = max_chars or settings.CHAT_HISTORY_MAX_CHARS
+
+        try:
+            rows = (
+                db.query(ChatMessage)
+                .filter(ChatMessage.session_id == session_id)
+                .order_by(ChatMessage.created_at.desc(), ChatMessage.id.desc())
+                .limit(max_turns)
+                .all()
+            )
+        except Exception as e:
+            # History is an enhancement, never a precondition. A failure here
+            # must degrade to a stateless answer, not break the chat.
+            logger.warning(f"Could not load session history: {e}")
+            return []
+
+        history: List[Dict[str, str]] = []
+        budget = max_chars
+        for row in rows:  # newest → oldest
+            content = (row.content or "").strip()
+            if not content or row.role not in ("user", "assistant"):
+                continue
+            if len(content) > budget:
+                break
+            budget -= len(content)
+            history.append({"role": row.role, "content": content})
+
+        history.reverse()  # back to chronological order
+
+        # Never open the replay on an assistant turn: a dangling answer with no
+        # question reads as the model talking to itself and measurably degrades
+        # follow-up quality.
+        while history and history[0]["role"] == "assistant":
+            history.pop(0)
+
+        return history
+
+    @staticmethod
+    def _build_retrieval_query(query: str, history: List[Dict[str, str]]) -> str:
+        """Expand a follow-up into something retrievable on its own.
+
+        Retrieval sees only the raw question, so "what about surface mines?"
+        embeds to almost nothing useful and the reranker has no good candidates
+        to choose from — the generation half would have full history while the
+        retrieval half stayed blind.
+
+        This is deliberately a cheap string heuristic rather than an LLM
+        condensation call: it costs nothing, adds no latency to every turn, and
+        cannot fail. A dependent-looking follow-up is prefixed with the previous
+        user turn purely for embedding and lexical matching; the prompt the
+        model actually answers is untouched.
+        """
+        if not history:
+            return query
+
+        stripped = query.strip().lower()
+        looks_dependent = len(stripped) < 60 or stripped.startswith(
+            (
+                "what about",
+                "and ",
+                "but ",
+                "how about",
+                "why",
+                "what if",
+                "does it",
+                "do they",
+                "is it",
+                "are they",
+                "which one",
+                "that ",
+                "those ",
+                "it ",
+                "they ",
+                "then ",
+                "also",
+                "same",
+            )
+        )
+        if not looks_dependent:
+            return query
+
+        prior_user = next(
+            (m["content"] for m in reversed(history) if m["role"] == "user"), None
+        )
+        if not prior_user:
+            return query
+
+        return f"{prior_user}\n{query}"
+
+    def _build_messages(
+        self,
+        query: str,
+        context: str,
+        history: Optional[List[Dict[str, str]]] = None,
+    ) -> List[Dict[str, str]]:
+        """Assemble the LLM message list: system, prior turns, then this turn.
+
+        Retrieved context rides on the final user message only. Restating it on
+        every historical turn would multiply the prompt by the history depth and
+        let stale context compete with the chunks retrieved for the live
+        question — the citation rules in the system prompt must bind against
+        the current context, not an older one.
+        """
+        messages: List[Dict[str, str]] = [{"role": "system", "content": _SYSTEM_PROMPT}]
+        if history:
+            messages.extend(history)
+        messages.append(
+            {"role": "user", "content": self._build_user_message(query, context)}
+        )
+        return messages
+
     def get_mining_suggestions(self) -> List[str]:
         """Get suggested mining-related questions."""
         return [

app/services/document_service.py

@@ -197,10 +197,40 @@ class DocumentService:
 
                 # Entities & Summary
                 document.entities = entities if isinstance(entities, dict) else {}
-                document.summary = summary.get("summary", "Summary not available.")
+                document.summary = summary.get("summary") or (
+                    "Summary unavailable — the summarizer produced no result "
+                    "for this document. Click Re-analyze to try again."
+                )
                 document.key_points = summary.get("key_points", [])
 
                 # ── Step 7: Mark completed ───────────────────────────────────
+                #
+                # One agent losing its provider (Cerebras answers an exhausted
+                # account with 402) must not throw away the work of the other
+                # three, so the document is COMPLETED with whatever succeeded.
+                # But it must not look like a clean run either: the analysis
+                # endpoint coerces entities to plain lists and drops the error
+                # markers, so the reason is recorded on processing_error, which
+                # DocumentResponse does expose.
+                degraded = (results.get("metadata") or {}).get(
+                    "degraded_sections"
+                ) or {}
+                if degraded:
+                    details = "; ".join(
+                        f"{name}: {reason}" for name, reason in degraded.items()
+                    )
+                    document.processing_error = (
+                        f"Partial AI analysis — {len(degraded)} of 4 sections "
+                        f"unavailable ({details}). Click Re-analyze to run the "
+                        "missing sections."
+                    )
+                    logger.warning(
+                        f"Document {document_id} completed with degraded "
+                        f"sections: {details}"
+                    )
+                else:
+                    document.processing_error = None
+
                 document.status = DocumentStatus.COMPLETED
                 document.processed_at = datetime.now(timezone.utc)
                 db.commit()
@@ -209,17 +239,30 @@ class DocumentService:
                 return True
 
             except QuotaExceededError as qe:
-                # Quota hit: text extraction + embeddings already succeeded.
-                # Mark COMPLETED with partial data so re-analyze is available.
+                # Only the classifier can reach here: it runs before the
+                # gather() that absorbs the other three agents' failures. Text
+                # extraction and embeddings already succeeded, so the document
+                # is searchable — mark it COMPLETED with partial data rather
+                # than FAILED, so Re-analyze is offered.
+                #
+                # The provider is named rather than assumed. This used to say
+                # "Gemini" unconditionally, which was wrong for every agent:
+                # the classifier runs on Groq and the extractor and summarizer
+                # on Cerebras.
+                provider = getattr(qe, "provider", "") or "AI"
                 logger.error(
-                    f"Gemini quota exceeded during agent analysis for {document_id}: {qe}"
+                    f"{provider} quota exceeded during agent analysis for "
+                    f"{document_id}: {qe}"
                 )
                 document.status = DocumentStatus.COMPLETED
                 document.processing_error = (
-                    "AI analysis incomplete: Gemini API quota exceeded. "
+                    f"AI analysis incomplete: {provider} quota exceeded. "
                     "Click Re-analyze to run the full analysis when quota resets."
                 )
-                document.summary = "Summary not available — Gemini quota exceeded. Re-analyze to generate."
+                document.summary = (
+                    f"Summary unavailable — {provider} quota exceeded before "
+                    "analysis could run. Click Re-analyze to generate it."
+                )
                 document.key_points = []
                 document.processed_at = datetime.now(timezone.utc)
                 db.commit()

tests/integration/conftest.py

@@ -0,0 +1,195 @@
+"""
+Integration test fixtures — real PostgreSQL, not SQLite.
+
+Why this file exists
+--------------------
+The root conftest points every test at SQLite (``test.db``). That is fine for
+unit tests, which never touch a dialect-specific column, but it does not work
+for these API tests and never really did:
+
+  * Every model uses ``sqlalchemy.dialects.postgresql.UUID`` primary keys. On
+    SQLite those bind as ``CHAR(32)`` hex, and any place the app hands the
+    driver a ``uuid.UUID`` (or a ``str`` where the other was expected) blows up
+    with ``type 'UUID' is not supported`` / ``'str' object has no attribute
+    'hex'``. That is what the five long-standing failures in
+    ``test_chat_api.py`` were.
+  * SQLite does not enforce the foreign keys to ``users.clerk_user_id`` by
+    default, so these tests were silently not exercising them.
+  * SQLite has no pgvector and no ``to_tsvector`` full-text search, so the two
+    things the retrieval path is actually built on could never be covered.
+
+Papering over the UUID binding with a type decorator would have kept the tests
+green on a database the application never runs against. CI already starts a
+``pgvector/pgvector:pg16`` service container and runs Alembic against it, so
+the real database was sitting there unused — these fixtures point at it.
+
+Running locally
+---------------
+    docker compose up -d postgres
+    pytest tests/integration -v
+
+The default URL matches the ``postgres`` service in ``docker-compose.yml``.
+Override with ``TEST_DATABASE_URL`` to use a different server. When no server
+is reachable the whole package skips with an explanatory message rather than
+failing — same convention as ``tests/eval/test_retrieval_eval.py``.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Generator
+from urllib.parse import urlsplit, urlunsplit
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.models.base import Base
+
+# Matches the `postgres` service in docker-compose.yml. `miningniti_test` keeps
+# the suite off the dev database; it is created on demand if missing.
+DEFAULT_TEST_DB_URL = (
+    "postgresql+psycopg2://postgres:postgres@localhost:5432/miningniti_test"
+)
+
+# connect_timeout keeps an unreachable-but-routable host from stalling the run
+# for the driver's default before the skip fires.
+_CONNECT_KWARGS = {"pool_pre_ping": True, "connect_args": {"connect_timeout": 5}}
+
+SKIP_REASON = (
+    "Integration tests need PostgreSQL. Start one with "
+    "`docker compose up -d postgres`, or point TEST_DATABASE_URL at a server. "
+    "Tried: {url}\nConnection error: {err}"
+)
+
+
+def _resolve_url() -> str:
+    """
+    Pick the PostgreSQL URL for the integration suite.
+
+    TEST_DATABASE_URL wins. Otherwise DATABASE_URL is used when it is a
+    PostgreSQL URL — that is the case in CI, where the workflow points it at
+    the pgvector service container. The root conftest defaults DATABASE_URL to
+    SQLite for unit tests, which is why a non-PostgreSQL value is ignored here.
+    """
+    explicit = os.environ.get("TEST_DATABASE_URL")
+    if explicit:
+        return explicit
+
+    configured = os.environ.get("DATABASE_URL", "")
+    if configured.startswith("postgresql"):
+        return configured
+
+    return DEFAULT_TEST_DB_URL
+
+
+def _with_database(url: str, database: str) -> str:
+    parts = urlsplit(url)
+    return urlunsplit(
+        (parts.scheme, parts.netloc, f"/{database}", parts.query, parts.fragment)
+    )
+
+
+def _ensure_database_exists(url: str) -> None:
+    """
+    CREATE DATABASE the target if it is missing.
+
+    Only reached for the local default; in CI the service container already
+    provisions the database and this connects, finds it, and returns.
+    """
+    target = urlsplit(url).path.lstrip("/")
+    if not target:
+        return
+
+    admin = create_engine(
+        _with_database(url, "postgres"), isolation_level="AUTOCOMMIT", **_CONNECT_KWARGS
+    )
+    try:
+        with admin.connect() as conn:
+            exists = conn.execute(
+                text("SELECT 1 FROM pg_database WHERE datname = :name"),
+                {"name": target},
+            ).scalar()
+            if not exists:
+                # Identifier cannot be bound as a parameter; target comes from
+                # our own URL, not user input.
+                conn.execute(text(f'CREATE DATABASE "{target}"'))
+    finally:
+        admin.dispose()
+
+
+@pytest.fixture(scope="session")
+def test_engine() -> Generator[Engine, None, None]:
+    """
+    Session-scoped engine against real PostgreSQL.
+
+    Overrides the SQLite engine of the same name in the root conftest for
+    everything under tests/integration/. Skips the suite when no server is
+    reachable.
+    """
+    url = _resolve_url()
+
+    try:
+        _ensure_database_exists(url)
+        engine = create_engine(url, **_CONNECT_KWARGS)
+        with engine.connect() as conn:
+            # DocumentEmbedding.embedding is a pgvector column, so the type has
+            # to exist before create_all(). In CI the Alembic migrations have
+            # already done this and create_all() below is a no-op.
+            conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+            conn.commit()
+    except Exception as exc:  # pragma: no cover - environment dependent
+        pytest.skip(SKIP_REASON.format(url=url, err=exc))
+
+    # checkfirst=True: leaves the Alembic-managed schema in CI untouched, and
+    # bootstraps a bare database locally. Deliberately no drop_all() on
+    # teardown — later CI steps (the retrieval eval) run against this same
+    # database, and every test rolls its own writes back anyway.
+    Base.metadata.create_all(bind=engine, checkfirst=True)
+
+    yield engine
+
+    engine.dispose()
+
+
+@pytest.fixture(scope="function")
+def db_session(test_engine: Engine) -> Generator[Session, None, None]:
+    """
+    Per-test session wrapped in a transaction that is always rolled back.
+
+    ``join_transaction_mode="create_savepoint"`` is what makes this work
+    against PostgreSQL: the API code under test calls ``session.commit()``, and
+    without it that commit ends the outer transaction, leaking rows into the
+    next test (on SQLite it produced the "transaction already deassociated"
+    warnings). With savepoints, the app's commits are real as far as the test
+    is concerned but the outer rollback below still undoes all of them.
+    """
+    connection = test_engine.connect()
+    transaction = connection.begin()
+
+    TestingSessionLocal = sessionmaker(
+        bind=connection,
+        autocommit=False,
+        autoflush=False,
+        join_transaction_mode="create_savepoint",
+    )
+    session = TestingSessionLocal()
+
+    # The foreign keys from documents/chat_sessions to users.clerk_user_id are
+    # enforced here, unlike on SQLite. The `client` fixture overrides
+    # get_current_user_id, which is the dependency that would normally
+    # provision this row (app/api/deps.py::_ensure_user_row), so the test has
+    # to stand it up itself.
+    from app.models.user import User
+
+    session.add(User(clerk_user_id="test_user_001", is_active=True))
+    session.commit()
+
+    try:
+        yield session
+    finally:
+        session.close()
+        if transaction.is_active:
+            transaction.rollback()
+        connection.close()

tests/integration/test_chat_api.py

@@ -108,9 +108,15 @@ class TestSendMessage:
             "app.services.chat_service.ChatService.generate_response",
             new_callable=AsyncMock,
         ) as mock_gen:
+            # ChatService.generate_response returns (answer, sources,
+            # tokens_used). The two-element mock here used to fail as
+            # "not enough values to unpack" the moment the endpoint was
+            # actually reached — it never was, because the request died on
+            # SQLite UUID binding first.
             mock_gen.return_value = (
                 "According to [Mining_site.pdf, Page 12], methane limits are below 1%.",
                 [],
+                {"input": 128, "output": 42},
             )
             response = client.post(
                 "/api/v1/chat/send",
@@ -134,7 +140,11 @@ class TestSendMessage:
             "app.services.chat_service.ChatService.generate_response",
             new_callable=AsyncMock,
         ) as mock_gen:
-            mock_gen.return_value = ("Test answer with [doc.pdf, Page 5] citation.", [])
+            mock_gen.return_value = (
+                "Test answer with [doc.pdf, Page 5] citation.",
+                [],
+                {"input": 128, "output": 42},
+            )
             response = client.post(
                 "/api/v1/chat/send",
                 json={
@@ -156,7 +166,11 @@ class TestSendMessage:
             "app.services.chat_service.ChatService.generate_response",
             new_callable=AsyncMock,
         ) as mock_gen:
-            mock_gen.return_value = ("Response text.", [])
+            mock_gen.return_value = (
+                "Response text.",
+                [],
+                {"input": 128, "output": 42},
+            )
             response = client.post(
                 "/api/v1/chat/send",
                 json={

tests/unit/test_agents.py

@@ -71,16 +71,6 @@ MOCK_SUMMARY_JSON = json.dumps(
 )
 
 
-def make_mock_model(json_response: str):
-    """Create a mock Gemini model that returns json_response."""
-    mock_response = MagicMock()
-    mock_response.text = json_response
-
-    mock_model = MagicMock()
-    mock_model.generate_content = MagicMock(return_value=mock_response)
-    return mock_model
-
-
 def make_mock_client(json_response: str):
     """Create a mock OpenAI client that returns json_response."""
     mock_choice = MagicMock()
@@ -284,7 +274,11 @@ class TestSummarizerAgent:
         from app.agents.summarizer import SummarizerAgent
 
         agent = SummarizerAgent()
-        agent.model = make_mock_model(MOCK_SUMMARY_JSON)
+        # SummarizerAgent is provider="cerebras" and goes through
+        # BaseAgent._call_openai_compat(self.client, ...). Patching `.model`
+        # here was a no-op that let this "unit" test make a live HTTPS call to
+        # api.cerebras.ai — patch the attribute the code path actually reads.
+        agent.client = make_mock_client(MOCK_SUMMARY_JSON)
 
         result = await agent.analyze(sample_mining_text)
 
@@ -300,7 +294,11 @@ class TestSummarizerAgent:
         from app.agents.summarizer import SummarizerAgent
 
         agent = SummarizerAgent()
-        agent.model = make_mock_model(MOCK_SUMMARY_JSON)
+        # SummarizerAgent is provider="cerebras" and goes through
+        # BaseAgent._call_openai_compat(self.client, ...). Patching `.model`
+        # here was a no-op that let this "unit" test make a live HTTPS call to
+        # api.cerebras.ai — patch the attribute the code path actually reads.
+        agent.client = make_mock_client(MOCK_SUMMARY_JSON)
 
         result = await agent.analyze(sample_mining_text)
         assert isinstance(result["key_points"], list)

tests/unit/test_chat_service.py

@@ -195,3 +195,209 @@ class TestGetMiningeSuggestions:
         assert len(suggestions) > 0
         for s in suggestions:
             assert isinstance(s, str)
+
+
+class TestConversationHistory:
+    """Tests for multi-turn conversation memory.
+
+    Prior turns are replayed to the model so follow-ups resolve against what
+    was already discussed. These guard the two failure modes that make history
+    worse than no history: replaying the live question as its own context, and
+    opening the replay on a dangling assistant turn.
+    """
+
+    @staticmethod
+    def _rows(*pairs):
+        """Build fake ChatMessage rows, newest first (query order)."""
+        rows = []
+        for role, content in pairs:
+            row = MagicMock()
+            row.role = role
+            row.content = content
+            rows.append(row)
+        return rows
+
+    @staticmethod
+    def _db_returning(rows):
+        db = MagicMock()
+        chain = db.query.return_value.filter.return_value.order_by.return_value
+        chain.limit.return_value.all.return_value = rows
+        return db
+
+    @pytest.mark.unit
+    def test_history_is_chronological(self):
+        """Rows arrive newest-first from the DB and must be replayed oldest-first."""
+        from app.services.chat_service import ChatService
+
+        db = self._db_returning(
+            self._rows(
+                ("assistant", "Methane must stay below 1%."),
+                ("user", "What are the methane limits?"),
+            )
+        )
+        history = ChatService.load_session_history(db, "session-1")
+
+        assert [m["role"] for m in history] == ["user", "assistant"]
+        assert history[0]["content"] == "What are the methane limits?"
+
+    @pytest.mark.unit
+    def test_history_never_starts_on_assistant_turn(self):
+        """A dangling answer with no question is dropped from the front."""
+        from app.services.chat_service import ChatService
+
+        db = self._db_returning(
+            self._rows(
+                ("assistant", "Newest answer."),
+                ("user", "Newest question."),
+                (
+                    "assistant",
+                    "Orphaned answer whose question fell outside the window.",
+                ),
+            )
+        )
+        history = ChatService.load_session_history(db, "session-1")
+
+        assert history[0]["role"] == "user"
+        assert all("Orphaned" not in m["content"] for m in history)
+
+    @pytest.mark.unit
+    def test_char_budget_trims_oldest_first(self):
+        """The budget retains the most recent turns, not the oldest."""
+        from app.services.chat_service import ChatService
+
+        db = self._db_returning(
+            self._rows(
+                ("user", "recent"),
+                ("assistant", "x" * 500),
+                ("user", "ancient"),
+            )
+        )
+        history = ChatService.load_session_history(db, "session-1", max_chars=50)
+
+        contents = [m["content"] for m in history]
+        assert "recent" in contents
+        assert "ancient" not in contents
+
+    @pytest.mark.unit
+    def test_turn_cap_is_passed_to_query(self):
+        """max_turns bounds the DB query rather than being applied after."""
+        from app.services.chat_service import ChatService
+
+        db = self._db_returning([])
+        ChatService.load_session_history(db, "session-1", max_turns=4)
+
+        limit = db.query.return_value.filter.return_value.order_by.return_value.limit
+        limit.assert_called_once_with(4)
+
+    @pytest.mark.unit
+    def test_db_failure_degrades_to_stateless(self):
+        """History is an enhancement; a DB error must not break the chat."""
+        from app.services.chat_service import ChatService
+
+        db = MagicMock()
+        db.query.side_effect = RuntimeError("connection lost")
+
+        assert ChatService.load_session_history(db, "session-1") == []
+
+    @pytest.mark.unit
+    def test_blank_and_system_rows_are_skipped(self):
+        """Empty content and non user/assistant roles never reach the model."""
+        from app.services.chat_service import ChatService
+
+        db = self._db_returning(
+            self._rows(
+                ("user", "Real question."),
+                ("system", "internal note"),
+                ("assistant", "   "),
+            )
+        )
+        history = ChatService.load_session_history(db, "session-1")
+
+        assert history == [{"role": "user", "content": "Real question."}]
+
+    @pytest.mark.unit
+    def test_messages_place_history_between_system_and_query(self):
+        """System prompt first, prior turns next, live question last."""
+        from app.services.chat_service import ChatService
+
+        history = [
+            {"role": "user", "content": "What are the methane limits?"},
+            {"role": "assistant", "content": "Below 1%."},
+        ]
+        messages = ChatService()._build_messages(
+            "What about surface mines?", "CTX", history
+        )
+
+        assert messages[0]["role"] == "system"
+        assert messages[1:3] == history
+        assert messages[-1]["role"] == "user"
+        assert "What about surface mines?" in messages[-1]["content"]
+
+    @pytest.mark.unit
+    def test_context_rides_only_on_the_live_turn(self):
+        """Retrieved context must not be restated on historical turns."""
+        from app.services.chat_service import ChatService
+
+        history = [{"role": "user", "content": "Earlier question."}]
+        messages = ChatService()._build_messages(
+            "Now what?", "UNIQUE_CTX_MARKER", history
+        )
+
+        carrying = [m for m in messages if "UNIQUE_CTX_MARKER" in m["content"]]
+        assert len(carrying) == 1
+        assert carrying[0] is messages[-1]
+
+    @pytest.mark.unit
+    def test_no_history_yields_system_plus_query_only(self):
+        """Behaviour with an empty history is unchanged from the stateless path."""
+        from app.services.chat_service import ChatService
+
+        messages = ChatService()._build_messages("A question.", "CTX", None)
+
+        assert len(messages) == 2
+        assert [m["role"] for m in messages] == ["system", "user"]
+
+
+class TestRetrievalQueryExpansion:
+    """Tests for _build_retrieval_query.
+
+    Generation gets full history, so retrieval must too — otherwise a
+    follow-up embeds to almost nothing and the reranker has no good
+    candidates to pick from.
+    """
+
+    @pytest.mark.unit
+    def test_dependent_followup_is_expanded(self):
+        """A short follow-up inherits the previous user turn for retrieval."""
+        from app.services.chat_service import ChatService
+
+        history = [
+            {"role": "user", "content": "What are the methane limits underground?"},
+            {"role": "assistant", "content": "Below 1%."},
+        ]
+        expanded = ChatService._build_retrieval_query(
+            "What about surface mines?", history
+        )
+
+        assert "methane limits underground" in expanded
+        assert "What about surface mines?" in expanded
+
+    @pytest.mark.unit
+    def test_self_contained_question_is_untouched(self):
+        """A long, standalone question must not be polluted with prior context."""
+        from app.services.chat_service import ChatService
+
+        history = [{"role": "user", "content": "What are the methane limits?"}]
+        query = (
+            "Describe in full the statutory ventilation survey obligations that "
+            "apply to an underground coal mine operator under 30 CFR 75.323."
+        )
+
+        assert ChatService._build_retrieval_query(query, history) == query
+
+    @pytest.mark.unit
+    def test_first_turn_has_nothing_to_expand_from(self):
+        """With no history the query passes through unchanged."""
+        from app.services.chat_service import ChatService
+
+        assert ChatService._build_retrieval_query("Short one?", []) == "Short one?"