fix(latency): KI-099 — bound every per-turn LLM call

KI-098 audit surfaced 5 holes where a per-turn LLM call could hang the
user-facing chat for 60-120s. This commit installs explicit budgets:

backend/orchestrator.py:
- extract_profile_updates → asyncio.wait_for(12s) + TimeoutError handler
- main brain pick.provider.chat → asyncio.wait_for(30s)
- faithfulness cross-check retry → asyncio.wait_for(20s) + skip-retry
on timeout (rare-path fallback; safer to skip than hang)

backend/providers/sarvam_llm.py:
- Monolithic timeout=60.0 → httpx.Timeout(connect=2, read=20, write=2,
pool=2) mirroring KI-084 on NIM. Constructor kept backward compatible.

backend/translator.py:
- asyncio.wait_for(20s) around both Sarvam .chat() calls. On timeout,
passthrough original text (callers already tolerate via KI-004).
Third callsite (back-translate) routes through translate_to_english
so it inherits the wrap.

backend/providers/nvidia_nim_llm.py:
- Budget guard before the synchronous probe_all() refresh inside chat().
If we've already burned 40% of total_budget_s, skip the probe and
raise immediately. This was the MAIN amplifier — a saturated chain
walking 10 candidates at 8s each = 60-80s wall time on the hot turn.
Next turn benefits from the background probe loop's 300s cadence.

Tests: 31/31 pass (routing_regression + credits_election + name_persistence).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/orchestrator.py

@@ -12,6 +12,8 @@ For each user turn:
 
 from __future__ import annotations
 
+import asyncio
+import logging
 import re
 import time
 from dataclasses import dataclass, field
@@ -546,7 +548,20 @@ async def handle_turn(
             extracted = None  # eval mode OR fact-find turn — bypass LLM extractor
         else:
             from backend.profile_extractor import extract_profile_updates
-            extracted = await extract_profile_updates(user_text, session.profile)
+            # KI-098 — outer budget cap. extract_profile_updates calls a
+            # NIM LLM which can hang up to 120s on its default timeout when
+            # the upstream chain is in synchronous probe_all() refresh. Skip
+            # the merge on timeout rather than stall the user-facing turn.
+            try:
+                extracted = await asyncio.wait_for(
+                    extract_profile_updates(user_text, session.profile),
+                    timeout=12.0,
+                )
+            except asyncio.TimeoutError:
+                extracted = None
+                logging.warning(
+                    "extractor timeout, skipping merge (session=%s)", session_id,
+                )
         if extracted:
             for field_name, new_value in extracted.items():
                 # KI-094 — never let the LLM extractor CLEAR a filled field.
@@ -596,6 +611,13 @@ async def handle_turn(
                     "profile-chunk upsert failed (session=%s): %s: %s",
                     session_id, type(e).__name__, str(e)[:200],
                 )
+    except asyncio.TimeoutError:
+        # KI-098 — explicit timeout path (in case any awaited call inside the
+        # try-block other than extract_profile_updates raises TimeoutError).
+        extracted = None
+        logging.warning(
+            "extractor timeout, skipping merge (session=%s)", session_id,
+        )
     except Exception as e:
         # KI-006 — log profile-extraction failures (extractor LLM down,
         # malformed model output, etc.). The chat ships unaffected.
@@ -636,7 +658,15 @@ async def handle_turn(
     # responses with low voice latency. The judge model (Meta Llama-4 Maverick)
     # in faithfulness.py is from a different company, architecture, and
     # training corpus — the brain does not mark its own homework.
-    llm_result = await pick.provider.chat(messages=messages, temperature=0.2, max_tokens=1500)
+    # KI-098 — outer budget cap on the main brain call. Provider default
+    # timeouts (120s) can hang the user-facing turn when the upstream chain
+    # is in synchronous probe_all() refresh. 30s is generous enough for
+    # normal MoE responses but short enough to fall through to the chain's
+    # error path before the user gives up.
+    llm_result = await asyncio.wait_for(
+        pick.provider.chat(messages=messages, temperature=0.2, max_tokens=1500),
+        timeout=30.0,
+    )
 
     raw = llm_result.text
     reply = strip_think_tags(raw)
@@ -683,21 +713,37 @@ async def handle_turn(
         if not gate1_failure:
             try:
                 secondary = NvidiaNimLLM(model=NIM_JUDGE_MODEL)
-                second = await secondary.chat(messages=messages, temperature=0.1, max_tokens=1500)
-                second_reply = strip_think_tags(second.text)
-                # Cross-check brain was NIM_JUDGE_MODEL — pass its id so the
-                # judge for THIS retry also excludes that model+family.
-                second_verdict = await check_faithfulness(
-                    reply=second_reply, chunks=chunks, user_text=user_text, run_llm_judge=True,
-                    brain_model_used=getattr(second, "model", None) or NIM_JUDGE_MODEL,
-                )
-                if second_verdict.passed:
-                    reply = second_reply
-                    pick = BrainPick(secondary, f"crosscheck-rescued-by-maverick")
-                    verdict = second_verdict
-                else:
+                # KI-098 — outer budget cap on the cross-check retry. This is
+                # a RARE fallback for hallucination-blocked replies; skipping
+                # it on timeout is strictly safer than hanging the turn 60-120s
+                # while the upstream chain refreshes via probe_all().
+                try:
+                    second = await asyncio.wait_for(
+                        secondary.chat(messages=messages, temperature=0.1, max_tokens=1500),
+                        timeout=20.0,
+                    )
+                except asyncio.TimeoutError:
+                    logging.warning(
+                        "crosscheck retry timeout, skipping retry (session=%s)", session_id,
+                    )
                     blocked = True
                     reply = verdict.suggested_reply or "I don't have grounded evidence for that. Could you rephrase?"
+                    second = None
+                if second is not None:
+                    second_reply = strip_think_tags(second.text)
+                    # Cross-check brain was NIM_JUDGE_MODEL — pass its id so the
+                    # judge for THIS retry also excludes that model+family.
+                    second_verdict = await check_faithfulness(
+                        reply=second_reply, chunks=chunks, user_text=user_text, run_llm_judge=True,
+                        brain_model_used=getattr(second, "model", None) or NIM_JUDGE_MODEL,
+                    )
+                    if second_verdict.passed:
+                        reply = second_reply
+                        pick = BrainPick(secondary, f"crosscheck-rescued-by-maverick")
+                        verdict = second_verdict
+                    else:
+                        blocked = True
+                        reply = verdict.suggested_reply or "I don't have grounded evidence for that. Could you rephrase?"
             except Exception:
                 blocked = True
                 reply = verdict.suggested_reply or "I don't have grounded evidence for that. Could you rephrase?"

backend/providers/nvidia_nim_llm.py

@@ -662,6 +662,18 @@ class NimChainLLM(LLMProvider):
         # touched this turn. This is the pre-KI-080 graceful-degradation
         # behaviour preserved for the (rare) double-failure case.
         try:
+            # KI-099 — bound the probe-refresh cost. On a hot user-facing turn, a
+            # 60-80s probe walk through ~10 candidates is unacceptable. If we've
+            # already spent half the call's total_budget_s before reaching here,
+            # skip the probe refresh and raise immediately — the next turn will
+            # benefit from the existing probe cache or the background probe loop
+            # (300s cadence) will refresh it.
+            elapsed = time.time() - call_t0
+            if elapsed > self.total_budget_s * 0.4:
+                raise RuntimeError(
+                    f"NimChainLLM budget exhausted before probe-refresh ({elapsed:.1f}s > {self.total_budget_s * 0.4:.1f}s); "
+                    f"skipping probe_all + raising"
+                )
             await llm_health.probe_all()
             refreshed = llm_health.filter_chain(allowed_chain)
             for model in refreshed:

backend/providers/sarvam_llm.py

@@ -10,7 +10,7 @@ Auth: header `api-subscription-key: <SARVAM_API_KEY>` (Sarvam) or
 
 from __future__ import annotations
 
-from typing import Optional
+from typing import Optional, Union
 
 import httpx
 
@@ -18,6 +18,15 @@ from backend.config import settings
 from backend.providers.base import ChatMessage, LLMProvider, LLMResult
 
 
+# KI-099 — split httpx.Timeout mirrors KI-084 on NIM. Connect/write/pool
+# tight; read still generous since Sarvam can legitimately stream long.
+# Replaces the prior monolithic timeout=60.0 which let a stalled socket
+# hold a slot past any outer cancellation — amplified by Sarvam's 3x
+# per-turn call pattern (EN-translate inbound, Indic-translate outbound,
+# Hinglish back-translate check).
+_SARVAM_TIMEOUT = httpx.Timeout(connect=2.0, read=20.0, write=2.0, pool=2.0)
+
+
 class SarvamLLM(LLMProvider):
     name = "sarvam-m"
     model = settings.SARVAM_LLM_MODEL
@@ -26,11 +35,13 @@ class SarvamLLM(LLMProvider):
         self,
         api_key: Optional[str] = None,
         model: str = settings.SARVAM_LLM_MODEL,
-        timeout: float = 60.0,
+        timeout: Optional[Union[float, httpx.Timeout]] = None,
     ):
         self.api_key = api_key or settings.SARVAM_API_KEY
         self.model = model
-        self.timeout = timeout
+        # KI-099 — default to the split-Timeout; only override when caller
+        # explicitly passes a value (preserves backward-compat for tests).
+        self.timeout = timeout if timeout is not None else _SARVAM_TIMEOUT
         if not self.api_key:
             raise RuntimeError("SARVAM_API_KEY not set in .env")
 

backend/translator.py

@@ -17,10 +17,23 @@ Why this is better than either model alone:
 
 from __future__ import annotations
 
+import asyncio
+import logging
+
 from backend.providers.base import ChatMessage
 from backend.providers.sarvam_llm import SarvamLLM
 
 
+# KI-099 — outer per-call wait_for. Even with KI-099's split httpx.Timeout
+# in sarvam_llm.py, an outer asyncio.wait_for guarantees the orchestrator
+# slot can't be held by a wedged Sarvam call past this ceiling. 20s
+# matches the inner read timeout — if httpx hasn't returned by then,
+# something is genuinely wedged.
+_SARVAM_CALL_TIMEOUT = 20.0
+
+_log = logging.getLogger(__name__)
+
+
 _TRANSLATE_TO_EN_SYSTEM = """You are a precise translator from Hindi / Hinglish / code-switched Indian English to clean standard English.
 
 RULES:
@@ -52,14 +65,29 @@ async def translate_to_english(text: str, sarvam: SarvamLLM | None = None) -> st
     if not text.strip():
         return text
     sarvam = sarvam or SarvamLLM()
-    res = await sarvam.chat(
-        messages=[
-            ChatMessage(role="system", content=_TRANSLATE_TO_EN_SYSTEM),
-            ChatMessage(role="user", content=text),
-        ],
-        temperature=0.0,
-        max_tokens=400,
-    )
+    try:
+        # KI-099 — outer wait_for caps end-to-end Sarvam latency at 20s even
+        # if httpx's inner read-timeout fails to fire. On timeout we
+        # passthrough the original text — callers in orchestrator.py /
+        # translation_check.py already tolerate the original text (they
+        # log + degrade gracefully, see KI-004 path).
+        res = await asyncio.wait_for(
+            sarvam.chat(
+                messages=[
+                    ChatMessage(role="system", content=_TRANSLATE_TO_EN_SYSTEM),
+                    ChatMessage(role="user", content=text),
+                ],
+                temperature=0.0,
+                max_tokens=400,
+            ),
+            timeout=_SARVAM_CALL_TIMEOUT,
+        )
+    except asyncio.TimeoutError:
+        _log.warning(
+            "sarvam translate_to_english wait_for timed out after %.1fs — passthrough",
+            _SARVAM_CALL_TIMEOUT,
+        )
+        return text
     out = res.text.strip()
     # Strip <think> tags if Sarvam-M went into reasoning mode
     from backend.persona import strip_think_tags
@@ -79,14 +107,28 @@ async def translate_to_indic(
     if not english.strip():
         return english
     sarvam = sarvam or SarvamLLM()
-    res = await sarvam.chat(
-        messages=[
-            ChatMessage(role="system", content=_TRANSLATE_TO_INDIC_SYSTEM),
-            ChatMessage(role="user", content=english),
-        ],
-        temperature=0.2,
-        max_tokens=600,
-    )
+    try:
+        # KI-099 — see translate_to_english for rationale. On timeout we
+        # passthrough the English; orchestrator.py treats an empty/equal
+        # Indic reply as "no cascade translation" and falls back to the
+        # English reply unchanged (existing behaviour for empty reply_indic).
+        res = await asyncio.wait_for(
+            sarvam.chat(
+                messages=[
+                    ChatMessage(role="system", content=_TRANSLATE_TO_INDIC_SYSTEM),
+                    ChatMessage(role="user", content=english),
+                ],
+                temperature=0.2,
+                max_tokens=600,
+            ),
+            timeout=_SARVAM_CALL_TIMEOUT,
+        )
+    except asyncio.TimeoutError:
+        _log.warning(
+            "sarvam translate_to_indic wait_for timed out after %.1fs — passthrough",
+            _SARVAM_CALL_TIMEOUT,
+        )
+        return english
     out = res.text.strip()
     from backend.persona import strip_think_tags
     return strip_think_tags(out) or english