fix(#40+voice+upload): SSOT scorecard parity + Sarvam 30s-truncation + uploaded-PDF retrieval gate

#40 SINGLE SOURCE OF TRUTH: extract the marketplace catalogue into
backend.main._marketplace_catalogue + marketplace_grade(); brain_tools
._scorecard_signal now delegates (deleted the parallel doctype-rank/_merge
re-implementation). Rec-card grade == marketplace grade for all 148 by
construction — deterministic oracle 790 ids / 0 mismatches; parity test
rewritten to assert the full id universe + alias/doctype edges.

VOICE: Sarvam saarika REST has a hard ~30s limit and silently returns
only the first 30s (HTTP 200). sarvam_stt now decodes to 16k mono and
splits >25s audio on silence, transcribes sequentially, concatenates,
raises on error (no silent failure). Frontend POST timeout scales with
blob size. Regression test: 92s utterance -> all 92 words.

UPLOAD: brain_tools profile-gate silently returned 0 chunks for a freshly
uploaded PDF asked about before fact-find (contradicted the 'searchable
now' UI promise). Added session-scoped quarantine bypass + single_brain
UPLOADED-DOC prompt rule. Session-private, no cross-session leak.

Full pytest gate green (rc=0); frontend compiled clean.

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

backend/brain_tools.py

@@ -261,99 +261,33 @@ def _insurer_reviews(slug: str) -> Optional[dict]:
     return ir
 
 
-# Mirrors the LOCAL _DOCTYPE_RANK inside backend.main.policies_all — the
-# marketplace picks, per product, the extracted file with the best doctype
-# rank. _scorecard_signal MUST select the identical file or the cited-card
-# grade diverges from the marketplace card for multi-doctype products
-# (#40). Keep in sync with main.policies_all if that ever changes.
-_DOCTYPE_RANK = {"wordings": 0, "prospectus": 1, "cis": 2, "brochure": 3}
-
-
 def _scorecard_signal(policy_id: str, profile=None) -> dict:
-    """{_grade, _overall_score} from backend.scorecard using the same
-    inputs as the marketplace, so a policy's recommendation grade equals
-    its marketplace grade (asserted by tests/test_scorecard_parity.py).
-    Read-only; returns {} on any failure (ranking still works off co-pay
-    + SI headroom + cosine)."""
+    """{_grade, _overall_score} for policy_id — the SAME grade its
+    marketplace card shows (#40 SINGLE SOURCE OF TRUTH, 2026-05-18).
+
+    Delegates to backend.main.marketplace_grade, which derives the grade
+    from the ONE marketplace catalogue computation
+    (backend.main._marketplace_catalogue) using UIN-canonical resolution:
+    a marketing-rename / KI-145 variant id resolves onto its canonical
+    card. There is no longer a parallel doctype-rank/_merge
+    re-implementation here, so the cited-card grade cannot drift from the
+    marketplace card grade — for ALL 148 by construction, asserted by
+    tests/test_scorecard_parity.py.
+
+    `profile` is accepted for call-site compatibility. The parity-relevant
+    grade LETTER (what _recommendation_fit gates on) is the profile-neutral
+    marketplace grade, identical to the /api/policies/all catalogue view.
+    Read-only; returns {} only when the policy is unknown to the
+    marketplace (ranking still works off co-pay + SI headroom + cosine)."""
     try:
-        cur = _curated_facts_all()
-        # PARITY-BY-CONSTRUCTION (#40, 2026-05-18): resolve the EXACT data
-        # the marketplace card grades. main.policies_all picks, per product
-        # key, the extracted file with the best doctype rank (wordings >
-        # prospectus > cis > brochure), then _merge_curated(extracted,
-        # curated[extracted.policy_id or pid] or curated[pid]). Taking the
-        # FIRST _candidate_stems hit instead graded the PASSED id's own
-        # doctype, so a __cis/__brochure recommendation id scored a
-        # different file than the marketplace's canonical __wordings card
-        # (royal-sundaram lifeline A↔C, sbi super-top-up B↔A,
-        # new-india-mediclaim C↔D). Mirror the marketplace selection so
-        # rec grade == marketplace grade for ALL 148 by construction.
-        pid = (policy_id or "").strip()
-        pkey = pid.rsplit("__", 1)[0] if "__" in pid else pid
-        best = None  # ((rank, stem), Path)
-        try:
-            for fp in settings.EXTRACTED_DIR.glob("*.json"):
-                st = fp.stem
-                fk = st.rsplit("__", 1)[0] if "__" in st else st
-                if fk != pkey:
-                    continue
-                dt = st.rsplit("__", 1)[1] if "__" in st else ""
-                rk = (_DOCTYPE_RANK.get(dt, 99), st)
-                if best is None or rk < best[0]:
-                    best = (rk, fp)
-        except Exception:  # noqa: BLE001 — glob must never break ranking
-            best = None
-        extracted = None
-        if best is not None:
-            try:
-                extracted = _json.loads(best[1].read_text())
-            except Exception:  # noqa: BLE001 — corrupt extract → curated-only
-                extracted = None
-        # Curated entry resolved EXACTLY as the marketplace merge does
-        # (curated[extracted.policy_id or pid] or curated[pid]); fall back
-        # to the candidate-stem scan only when there is no extracted file.
-        curated_entry = None
-        if extracted is not None:
-            curated_entry = cur.get(extracted.get("policy_id") or pid) or cur.get(pid)
-        if curated_entry is None:
-            for stem in _candidate_stems(pid):
-                if stem in cur:
-                    curated_entry = cur[stem]
-                    break
-        if extracted is not None:
-            from backend.main import _merge_curated  # lazy: avoids cycle
-            data = _merge_curated(extracted, curated_entry)
-        elif curated_entry:
-            data = curated_entry
-        else:
-            # Genuinely unknown policy → fail OPEN (no false weak signal;
-            # _recommendation_fit keeps chunks with no grade evidence).
-            return {}
-        from backend.scorecard import build_scorecard
-
-        slug = data.get("insurer_slug") or (
-            policy_id.split("__", 1)[0] if "__" in policy_id else ""
-        )
-        ir = _insurer_reviews(slug)
-        prof = None
-        if profile is not None:
-            prof = {
-                s: getattr(profile, s, None)
-                for s in (
-                    "age", "income_band", "primary_goal",
-                    "existing_cover_inr", "copay_pct",
-                )
+        from backend.main import marketplace_grade  # lazy: avoids cycle
+        sig = marketplace_grade(policy_id)
+        if sig and sig.get("_grade"):
+            return {
+                "_grade": sig.get("_grade"),
+                "_overall_score": sig.get("_overall_score"),
             }
-        sc = build_scorecard(data, insurer_reviews=ir, profile=prof)
-        # Scorecard.overall_score is the numeric field; `overall` is a
-        # back-compat alias.
-        _overall = getattr(sc, "overall_score", None)
-        if _overall is None:
-            _overall = getattr(sc, "overall", None)
-        return {
-            "_grade": getattr(sc, "grade", None),
-            "_overall_score": _overall,
-        }
+        return {}
     except Exception:  # noqa: BLE001 — scorecard optional; ranking degrades gracefully
         return {}
 
@@ -765,6 +699,100 @@ def _quality_seed_candidates(profile, limit: int = 25) -> list[dict]:
     return out[:limit]
 
 
+# ---- uploaded-doc (quarantine) bypass helpers ------------------------------
+
+
+def _session_has_quarantine_docs(session_id: str) -> bool:
+    """True iff this session has at least one chunk in the SEPARATE
+    `user_uploads_quarantine` Chroma collection.
+
+    Strictly session-scoped (where={"session_id": session_id}); cheap
+    metadata-only `.get(limit=1)`. Never raises — a Chroma hiccup just
+    means "treat as no upload" so the normal profile-gate path runs.
+    """
+    if not session_id:
+        return False
+    try:
+        from rag.ingest import get_quarantine_collection
+
+        q = get_quarantine_collection().get(
+            where={"session_id": session_id},
+            limit=1,
+            include=[],
+        )
+        return bool(q and q.get("ids"))
+    except Exception as e:  # noqa: BLE001 — best-effort probe
+        _log.warning(
+            "quarantine presence probe failed (sid=%s): %s: %s",
+            session_id, type(e).__name__, str(e)[:160],
+        )
+        return False
+
+
+async def _retrieve_uploaded_only(
+    query: str, session_id: str, top_k: int
+) -> dict:
+    """Quarantine-scoped retrieval that returns ONLY this session's
+    uploaded-doc chunks, bypassing the recommendation profile-gate.
+
+    Reuses `rag.retrieve.retrieve(session_id=...)` (whose quarantine boost
+    pass already prepends this session's uploaded chunks), then filters the
+    result down to `doc_type == "user_upload"` so the bypass response can
+    NEVER contain general-corpus policy chunks against an incomplete
+    profile. Returns the same dict shape as `retrieve_policies`.
+    """
+    try:
+        from rag.retrieve import retrieve as _retrieve
+
+        chunks = await _retrieve(
+            query=query,
+            top_k=max(int(top_k) if top_k else 8, 3),
+            session_id=session_id,
+        )
+    except Exception as e:  # noqa: BLE001 — graceful empty
+        _log.warning(
+            "uploaded-only retrieve failed (sid=%s): %s: %s",
+            session_id, type(e).__name__, str(e)[:160],
+        )
+        return {"chunks": [], "count": 0, "error": f"{type(e).__name__}"}
+
+    uploaded = []
+    for c in chunks or []:
+        if (getattr(c, "doc_type", "") or "").lower() != "user_upload":
+            continue
+        # Triple-check session ownership before exposing the chunk —
+        # belt + suspenders on top of the where={"session_id"} filter.
+        if (getattr(c, "session_id", None) or session_id) != session_id and \
+                getattr(c, "session_id", None) not in (None, ""):
+            continue
+        uploaded.append({
+            "chunk_id": getattr(c, "chunk_id", ""),
+            "policy_id": getattr(c, "policy_id", ""),
+            "policy_name": getattr(c, "policy_name", ""),
+            "insurer_slug": getattr(c, "insurer_slug", "") or "user-upload",
+            "doc_type": "user_upload",
+            "source_url": getattr(c, "source_url", ""),
+            "chunk_text": (getattr(c, "text", "") or "")[:1200],
+            "score": float(getattr(c, "score", 0.0) or 0.0),
+        })
+    if not uploaded:
+        return {"chunks": [], "count": 0}
+    return {
+        "chunks": uploaded,
+        "count": len(uploaded),
+        "query": query,
+        "source": "uploaded_doc_quarantine",
+        "note": (
+            "These chunks come from the user's OWN uploaded policy PDF "
+            "(session-scoped quarantine). Answer the user's question "
+            "about THIS document directly. Do NOT treat this as a "
+            "general recommendation and do NOT block on profile "
+            "completeness — the user explicitly asked about their "
+            "uploaded file."
+        ),
+    }
+
+
 # ---- retrieve_policies -----------------------------------------------------
 
 async def retrieve_policies(
@@ -813,6 +841,38 @@ async def retrieve_policies(
     if not isinstance(query, str) or not query.strip():
         return {"chunks": [], "count": 0, "error": "empty_query"}
 
+    # UPLOADED-DOC BYPASS (2026-05-18) — the user is explicitly promised in
+    # the UI ("✓ Indexed … It's now searchable in this chat. Ask me about
+    # it.") that an uploaded policy PDF is immediately queryable. But the
+    # profile-complete gate below blocks ALL retrieval until the 7-slot
+    # fact-find is done (unless policy_filter_ids is set, which the LLM
+    # can't know for a freshly-uploaded doc). An uploaded doc is, by
+    # definition, a known-doc-FOR-THIS-SESSION — semantically identical to
+    # the policy_filter_ids follow-up branch that already legitimately
+    # bypasses the gate. So: when this session has chunks in the SEPARATE
+    # `user_uploads_quarantine` collection, run a quarantine-ONLY retrieval
+    # that bypasses the recommendation profile-gate. Strictly session-
+    # scoped (where={"session_id": eff_session_id}) so user A's upload
+    # never leaks into user B. The full recommendation flow (general
+    # corpus + eligibility + scorecard) still requires the complete
+    # profile — this bypass only surfaces the user's OWN uploaded doc.
+    if (
+        eff_session_id
+        and not policy_filter_ids
+        and _session_has_quarantine_docs(eff_session_id)
+    ):
+        up = await _retrieve_uploaded_only(query, eff_session_id, top_k)
+        # Only short-circuit if the gate would otherwise block (incomplete
+        # profile) AND we actually found uploaded chunks. If the profile is
+        # already complete we fall through so the upload is folded into the
+        # normal ranked recommendation pool by the quarantine boost pass.
+        _profile_incomplete = profile is not None and any(
+            getattr(profile, slot, None) in (None, "", [])
+            for slot in _REQUIRED_FOR_READY
+        )
+        if up.get("count", 0) > 0 and _profile_incomplete:
+            return up
+
     # Profile-complete gate (skip if caller supplied policy_filter_ids — that
     # branch is a known-policy follow-up). Defense-in-depth: even if the LLM
     # ignores RULE 2 of the system prompt, this refuses to return chunks

backend/main.py

@@ -2936,33 +2936,18 @@ def _ki145_material_diffs(curated: dict, extracted: dict) -> int:
     return diffs
 
 
-@app.get("/api/policies/all", response_model=MarketplaceResponse)
-async def policies_all(session_id: Optional[str] = None):
-    """The marketplace data feed — every extracted policy + scorecard + filterable fields.
-
-    When session_id is provided AND the session has a profile populated to
-    ≥0.6 completeness, every policy is scored against THAT profile (dynamic
-    per-user grade). Otherwise we score with the generic baseline weights.
+def _marketplace_catalogue(user_profile_dict):
+    """SINGLE SOURCE OF TRUTH for the marketplace card set (#40).
+
+    The recommendation path (brain_tools._scorecard_signal via
+    marketplace_grade) and /api/policies/all BOTH derive a policy's
+    grade from THIS one computation, so the cited-card grade can never
+    diverge from the marketplace card grade — including marketing-rename
+    alias / KI-145 variant cards. Body is the former inline endpoint
+    logic, moved verbatim; fully synchronous.
     """
     import json as _json
-    from backend.scorecard import build_scorecard, profile_completeness as _completeness
-    from backend.session_state import get_session as _get_sess
-
-    # Pull user profile if we have one
-    user_profile_dict: Optional[dict] = None
-    if session_id:
-        sess = _get_sess(session_id)
-        p = sess.profile
-        profile_dict = {
-            "age": p.age, "dependents": p.dependents, "income_band": p.income_band,
-            "existing_cover_inr": p.existing_cover_inr, "primary_goal": p.primary_goal,
-            "location_tier": p.location_tier, "parents_to_insure": p.parents_to_insure,
-            "parents_age_max": p.parents_age_max, "parents_has_ped": p.parents_has_ped,
-            "health_conditions": p.health_conditions, "budget_band": p.budget_band,
-        }
-        if _completeness(profile_dict) >= 0.6:
-            user_profile_dict = profile_dict
-
+    from backend.scorecard import build_scorecard
     corpus_url_index = _build_corpus_url_index()
     curated_facts = _load_curated_facts()
 
@@ -3485,7 +3470,36 @@ async def policies_all(session_id: Optional[str] = None):
         else:
             prev.aliases = sorted(set(prev.aliases) | set(p.aliases))
     deduped = list(_best.values())
+    return deduped
+
+@app.get("/api/policies/all", response_model=MarketplaceResponse)
+async def policies_all(session_id: Optional[str] = None):
+    """The marketplace data feed — every extracted policy + scorecard + filterable fields.
+
+    When session_id is provided AND the session has a profile populated to
+    ≥0.6 completeness, every policy is scored against THAT profile (dynamic
+    per-user grade). Otherwise we score with the generic baseline weights.
+    """
+    import json as _json
+    from backend.scorecard import build_scorecard, profile_completeness as _completeness
+    from backend.session_state import get_session as _get_sess
 
+    # Pull user profile if we have one
+    user_profile_dict: Optional[dict] = None
+    if session_id:
+        sess = _get_sess(session_id)
+        p = sess.profile
+        profile_dict = {
+            "age": p.age, "dependents": p.dependents, "income_band": p.income_band,
+            "existing_cover_inr": p.existing_cover_inr, "primary_goal": p.primary_goal,
+            "location_tier": p.location_tier, "parents_to_insure": p.parents_to_insure,
+            "parents_age_max": p.parents_age_max, "parents_has_ped": p.parents_has_ped,
+            "health_conditions": p.health_conditions, "budget_band": p.budget_band,
+        }
+        if _completeness(profile_dict) >= 0.6:
+            user_profile_dict = profile_dict
+
+    deduped = _marketplace_catalogue(user_profile_dict)
     return MarketplaceResponse(
         policies=deduped,
         total=len(deduped),
@@ -4294,3 +4308,117 @@ else:
             "docs": "/docs",
             "health": "/api/health",
         }
+
+
+# ---- #40 SSOT grade resolver ------------------------------------------------
+# marketplace_grade(policy_id) returns the SAME (grade, overall) the
+# marketplace card for that policy's canonical identity shows. The
+# recommendation path calls this instead of re-deriving a scorecard, so
+# rec-card grade == marketplace grade for ALL 148 by construction.
+
+import threading as _mg_threading
+
+_MG_LOCK = _mg_threading.Lock()
+_MG_CACHE: dict = {"sig": None, "index": None}
+
+
+def _mg_data_signature() -> tuple:
+    """Cheap fingerprint that changes when any grading input changes
+    (so an uploaded-PDF card or a curated edit invalidates the cache)."""
+    sig = []
+    for d in (settings.EXTRACTED_DIR, settings.DATA_DIR / "policy_facts",
+              settings.DATA_DIR / "reviews"):
+        try:
+            for fp in sorted(d.glob("*.json")):
+                st = fp.stat()
+                sig.append((fp.name, int(st.st_mtime), st.st_size))
+        except Exception:  # noqa: BLE001 — missing dir → empty contribution
+            continue
+    return tuple(sig)
+
+
+def _mg_norm_uin(raw) -> str:
+    try:
+        from backend.policy_identity import normalize_uin
+        return normalize_uin(raw)
+    except Exception:  # noqa: BLE001
+        return ""
+
+
+def _mg_build_index() -> dict:
+    """{lookup_key -> (grade, overall_score)} for every marketplace card,
+    keyed by policy_id, product_key, and normalised UIN so a variant /
+    alias id resolves to its canonical card's grade."""
+    cards = _marketplace_catalogue(None)
+    cur = _load_curated_facts()
+    idx: dict = {}
+
+    def _put(k, val):
+        if k:
+            idx.setdefault(k, val)
+
+    for c in cards:
+        val = (c.grade, c.overall_score)
+        pid = c.policy_id or ""
+        _put(f"id:{pid}", val)
+        try:
+            from backend.policy_identity import product_key as _pk
+            _put(f"pk:{_pk(pid)}", val)
+        except Exception:  # noqa: BLE001
+            pass
+        # UIN of the card's underlying data (curated wins, like the catalogue)
+        cdata = cur.get(pid) or {}
+        uin = _mg_norm_uin(cdata.get("uin_code") or cdata.get("uin"))
+        if not uin:
+            try:
+                ep = settings.EXTRACTED_DIR / f"{pid}.json"
+                if ep.exists():
+                    import json as _j
+                    uin = _mg_norm_uin(_j.loads(ep.read_text()).get("uin_code"))
+            except Exception:  # noqa: BLE001
+                uin = ""
+        _put(f"uin:{uin}" if uin else "", val)
+    return idx
+
+
+def _mg_index() -> dict:
+    sig = _mg_data_signature()
+    with _MG_LOCK:
+        if _MG_CACHE["sig"] != sig or _MG_CACHE["index"] is None:
+            _MG_CACHE["index"] = _mg_build_index()
+            _MG_CACHE["sig"] = sig
+        return _MG_CACHE["index"]
+
+
+def marketplace_grade(policy_id: str) -> dict:
+    """{"_grade", "_overall_score"} for policy_id, identical to its
+    marketplace card. Resolution order: exact id -> product_key -> UIN
+    (so a marketing-rename / variant id maps onto its canonical card).
+    Returns {} only when the policy is unknown to the marketplace."""
+    if not policy_id:
+        return {}
+    idx = _mg_index()
+    from backend.policy_identity import product_key as _pk
+    cur = _load_curated_facts()
+    pid = policy_id.strip()
+    keys = [f"id:{pid}", f"pk:{_pk(pid)}"]
+    cdata = cur.get(pid) or cur.get(_pk(pid)) or {}
+    uin = _mg_norm_uin(cdata.get("uin_code") or cdata.get("uin"))
+    if not uin:
+        try:
+            import json as _j
+            for cand in (pid, _pk(pid)):
+                ep = settings.EXTRACTED_DIR / f"{cand}.json"
+                if ep.exists():
+                    uin = _mg_norm_uin(_j.loads(ep.read_text()).get("uin_code"))
+                    if uin:
+                        break
+        except Exception:  # noqa: BLE001
+            uin = ""
+    if uin:
+        keys.append(f"uin:{uin}")
+    for k in keys:
+        if k in idx:
+            g, o = idx[k]
+            return {"_grade": g, "_overall_score": o}
+    return {}

backend/providers/sarvam_stt.py

@@ -9,13 +9,34 @@ Response: {"transcript": str, "language_code": str?, "language_probability": flo
 from __future__ import annotations
 
 import io
-from typing import Optional
+import logging
+from typing import List, Optional
 
 import httpx
 
 from backend.config import settings
 from backend.providers.base import STTProvider, STTResult
 
+_log = logging.getLogger(__name__)
+
+# Sarvam's saarika REST /speech-to-text endpoint has a hard ~30s audio
+# limit. Critically, it does NOT 4xx on longer audio — it returns HTTP 200
+# with a `transcript` containing ONLY the first ~30s and silently drops the
+# rest. The live-voice hook's grace-window batching (UTTERANCE_GRACE_MS)
+# deliberately merges multiple pause-separated speech bursts into ONE blob,
+# so any real-world utterance with natural pauses easily exceeds 30s of
+# wall-clock audio and gets truncated. To capture the COMPLETE utterance we
+# split anything over the safe ceiling into <= SAFE chunks (at silence
+# boundaries where possible so words aren't cut mid-syllable), transcribe
+# each chunk, and concatenate the transcripts in order.
+#
+# Ceiling is set below the documented 30s so the boundary search + minor
+# encode jitter never pushes a chunk back over Sarvam's real limit.
+STT_CHUNK_MS = 25_000          # target max length per Sarvam call
+STT_SILENCE_SEARCH_MS = 4_000  # widen a cut by up to this to land on silence
+STT_MIN_SILENCE_MS = 350       # a gap >= this counts as a sentence pause
+STT_MAX_TOTAL_MS = 15 * 60_000  # hard safety cap (15 min) — refuse absurd input
+
 
 # Public error-code vocabulary returned by /api/transcribe when Sarvam fails.
 # Frontends (PTT + live voice) consume these as a closed enum so they can map
@@ -103,8 +124,8 @@ class SarvamSTT(STTProvider):
     _SARVAM_NATIVE_FORMATS = {"wav", "mp3", "flac", "ogg", "m4a"}
 
     @staticmethod
-    def _transcode_to_wav(audio_bytes: bytes, src_format: str) -> bytes:
-        """Convert any pydub-readable container to 16 kHz mono WAV.
+    def _load_segment(audio_bytes: bytes, src_format: str):
+        """Decode any pydub-readable container to a 16 kHz mono AudioSegment.
 
         Sarvam's recommended sampling rate is 16 kHz mono — what Saarika
         was trained on. Down-mixing + resampling at the gateway also
@@ -112,11 +133,101 @@ class SarvamSTT(STTProvider):
         """
         from pydub import AudioSegment
         audio = AudioSegment.from_file(io.BytesIO(audio_bytes), format=src_format)
-        audio = audio.set_frame_rate(16000).set_channels(1).set_sample_width(2)
+        return audio.set_frame_rate(16000).set_channels(1).set_sample_width(2)
+
+    @staticmethod
+    def _segment_to_wav_bytes(audio) -> bytes:
         buf = io.BytesIO()
         audio.export(buf, format="wav")
         return buf.getvalue()
 
+    @classmethod
+    def _transcode_to_wav(cls, audio_bytes: bytes, src_format: str) -> bytes:
+        """Convert any pydub-readable container to 16 kHz mono WAV."""
+        return cls._segment_to_wav_bytes(cls._load_segment(audio_bytes, src_format))
+
+    @staticmethod
+    def _split_on_silence(audio, chunk_ms: int) -> List:
+        """Split `audio` into <= chunk_ms slices.
+
+        We never cut a slice longer than chunk_ms. Where possible the cut is
+        nudged BACKWARDS to the quietest point within the last
+        STT_SILENCE_SEARCH_MS of the slice so the boundary lands in a pause
+        rather than mid-word (which would drop a syllable from BOTH the tail
+        of one chunk and the head of the next). If no clear pause is found we
+        cut at the hard chunk_ms boundary — a worst-case half-word seam, far
+        better than silently losing every word past 30s.
+        """
+        total = len(audio)
+        if total <= chunk_ms:
+            return [audio]
+
+        # Cheap energy probe per 100ms frame so we can find the quietest
+        # spot near each prospective cut without an external VAD.
+        from pydub import AudioSegment  # noqa: F401  (ensures pydub import path)
+
+        chunks: List = []
+        start = 0
+        while start < total:
+            if total - start <= chunk_ms:
+                chunks.append(audio[start:])
+                break
+
+            hard_cut = start + chunk_ms
+            search_lo = max(start + 1, hard_cut - STT_SILENCE_SEARCH_MS)
+            window = audio[search_lo:hard_cut]
+
+            # Find the 100ms frame with the lowest dBFS in the search window.
+            best_offset = None
+            best_db = None
+            step = 100
+            for off in range(0, max(1, len(window)), step):
+                frame = window[off:off + step]
+                if len(frame) == 0:
+                    continue
+                db = frame.dBFS  # -inf for pure digital silence
+                if best_db is None or db < best_db:
+                    best_db = db
+                    best_offset = off
+
+            # Only honour the silence cut if the quietest frame is actually
+            # quiet relative to the segment (a real pause), else hard-cut.
+            cut = hard_cut
+            if (
+                best_offset is not None
+                and best_db is not None
+                and (best_db == float("-inf") or best_db < audio.dBFS - 12)
+            ):
+                cut = search_lo + best_offset + step // 2
+                cut = min(max(cut, start + 1), hard_cut)
+
+            chunks.append(audio[start:cut])
+            start = cut
+        return chunks
+
+    async def _transcribe_wav_bytes(
+        self,
+        wav_bytes: bytes,
+        language_code: Optional[str],
+    ) -> dict:
+        """POST one already-encoded WAV blob to Sarvam, return its JSON.
+
+        Raises on transport / HTTP errors so the caller's classifier can map
+        them to a closed error_code — we deliberately do NOT swallow here.
+        """
+        url = f"{settings.SARVAM_BASE_URL}{settings.SARVAM_STT_PATH}"
+        files = {
+            "file": ("audio.wav", io.BytesIO(wav_bytes), "audio/wav"),
+        }
+        data = {"model": self.model}
+        if language_code:
+            data["language_code"] = language_code
+        headers = {"api-subscription-key": self.api_key}
+        async with httpx.AsyncClient(timeout=self.timeout) as client:
+            resp = await client.post(url, headers=headers, files=files, data=data)
+            resp.raise_for_status()
+            return resp.json()
+
     async def transcribe(
         self,
         audio_bytes: bytes,
@@ -129,36 +240,105 @@ class SarvamSTT(STTProvider):
             return STTResult(text="", language_code=language_code, confidence=0.0, raw={"reason": "audio_too_short"})
 
         fmt = (audio_format or "wav").lower().lstrip(".")
-        # Browser MediaRecorder uses webm/opus by default; Sarvam rejects it.
-        # Transcode in-process when we get a non-native format.
-        if fmt not in self._SARVAM_NATIVE_FORMATS:
-            try:
-                audio_bytes = self._transcode_to_wav(audio_bytes, fmt)
-                fmt = "wav"
-            except Exception as e:
-                # If pydub/ffmpeg fails (e.g., truly corrupt audio), let Sarvam
-                # see the original bytes and return its own error rather than
-                # silently swallowing.
-                pass
 
-        url = f"{settings.SARVAM_BASE_URL}{settings.SARVAM_STT_PATH}"
-        files = {
-            "file": (f"audio.{fmt}", io.BytesIO(audio_bytes), f"audio/{fmt}"),
-        }
-        data = {"model": self.model}
-        if language_code:
-            data["language_code"] = language_code
+        # Decode to a 16 kHz mono AudioSegment so we can (a) measure the true
+        # duration and (b) chunk anything over Sarvam's ~30s REST limit. The
+        # legacy path only transcoded webm→wav; native containers (wav/mp3/
+        # m4a/...) went straight through UNMEASURED, so a 90s m4a was sent
+        # whole and silently truncated to its first 30s by Sarvam. Decoding
+        # every format here closes that hole for ALL callers (live voice +
+        # push-to-talk).
+        segment = None
+        try:
+            segment = self._load_segment(audio_bytes, fmt)
+        except Exception:
+            # pydub/ffmpeg couldn't decode (truly corrupt audio, or a codec
+            # ffmpeg lacks). Fall back to the original single-shot path so
+            # Sarvam returns its own error rather than us swallowing it.
+            # Native formats are sent as-is; unknown ones default to wav.
+            _log.warning(
+                "STT decode failed (fmt=%s, %d bytes) — single-shot fallback",
+                fmt,
+                len(audio_bytes),
+            )
+            send_fmt = fmt if fmt in self._SARVAM_NATIVE_FORMATS else "wav"
+            url = f"{settings.SARVAM_BASE_URL}{settings.SARVAM_STT_PATH}"
+            files = {
+                "file": (f"audio.{send_fmt}", io.BytesIO(audio_bytes), f"audio/{send_fmt}"),
+            }
+            data = {"model": self.model}
+            if language_code:
+                data["language_code"] = language_code
+            headers = {"api-subscription-key": self.api_key}
+            async with httpx.AsyncClient(timeout=self.timeout) as client:
+                resp = await client.post(url, headers=headers, files=files, data=data)
+                resp.raise_for_status()
+                payload = resp.json()
+            return STTResult(
+                text=payload.get("transcript", ""),
+                language_code=payload.get("language_code"),
+                confidence=payload.get("language_probability"),
+                raw=payload,
+            )
 
-        headers = {"api-subscription-key": self.api_key}
+        total_ms = len(segment)
+        if total_ms > STT_MAX_TOTAL_MS:
+            # Refuse absurd input loudly rather than burn N Sarvam calls.
+            raise ValueError(
+                f"audio too long for STT: {total_ms} ms > cap {STT_MAX_TOTAL_MS} ms"
+            )
 
-        async with httpx.AsyncClient(timeout=self.timeout) as client:
-            resp = await client.post(url, headers=headers, files=files, data=data)
-            resp.raise_for_status()
-            payload = resp.json()
+        chunks = self._split_on_silence(segment, STT_CHUNK_MS)
+
+        # Common case: short utterance, one Sarvam call (identical wire
+        # behaviour to the pre-fix path, just always WAV-normalised).
+        if len(chunks) == 1:
+            payload = await self._transcribe_wav_bytes(
+                self._segment_to_wav_bytes(chunks[0]), language_code
+            )
+            return STTResult(
+                text=payload.get("transcript", ""),
+                language_code=payload.get("language_code"),
+                confidence=payload.get("language_probability"),
+                raw=payload,
+            )
+
+        # Long utterance: transcribe every chunk, then stitch the
+        # transcripts back together in order so the COMPLETE utterance
+        # survives. If any chunk fails we let the exception propagate so the
+        # /api/transcribe classifier surfaces a real error_code — we never
+        # return a silently-partial transcript (no silent failures).
+        _log.info(
+            "STT chunked transcription: %d ms split into %d chunks (<= %d ms each)",
+            total_ms,
+            len(chunks),
+            STT_CHUNK_MS,
+        )
+        parts: List[str] = []
+        first_lang: Optional[str] = None
+        first_conf: Optional[float] = None
+        raws: List[dict] = []
+        for idx, ch in enumerate(chunks):
+            payload = await self._transcribe_wav_bytes(
+                self._segment_to_wav_bytes(ch), language_code
+            )
+            piece = (payload.get("transcript") or "").strip()
+            if piece:
+                parts.append(piece)
+            if idx == 0:
+                first_lang = payload.get("language_code")
+                first_conf = payload.get("language_probability")
+            raws.append(payload)
 
+        stitched = " ".join(parts).strip()
         return STTResult(
-            text=payload.get("transcript", ""),
-            language_code=payload.get("language_code"),
-            confidence=payload.get("language_probability"),
-            raw=payload,
+            text=stitched,
+            language_code=first_lang or language_code,
+            confidence=first_conf,
+            raw={
+                "chunked": True,
+                "chunk_count": len(chunks),
+                "total_ms": total_ms,
+                "chunks": raws,
+            },
         )

backend/single_brain.py

@@ -112,6 +112,24 @@ If retrieve_policies returns nothing for that name, say "I couldn't find
 that policy in our index. Let me suggest some alternatives" and call
 retrieve_policies with a broader query based on the profile.
 
+═══════════════════════════════════
+UPLOADED-DOC RULE — USER-UPLOADED POLICY PDF (answer now, no fact-find gate)
+═══════════════════════════════════
+The user can upload their own policy PDF (a 📎 control in the chat). When
+they do, the UI tells them it is "searchable in this chat". If the user
+asks ANYTHING about their uploaded / attached document — e.g. "what does my
+policy cover?", "the PDF I just uploaded", "my current plan's room rent",
+"check the file I attached" — call retrieve_policies with their question as
+the query (policy_filter_ids=None). This works even if the profile fact-
+find is NOT complete: a tool result with "source": "uploaded_doc_quarantine"
+contains chunks from THEIR OWN uploaded file. Answer the question about
+that document directly and cite it as [Source: <their file's policy name>
+(uploaded document)]. Do NOT block on profile completeness and do NOT ask
+the 7 fact-find questions just to answer a question about their uploaded
+doc. (You still need the full profile before making NEW market
+recommendations — see RULE 2 — but reading back their own uploaded policy
+is not a recommendation.)
+
 ═══════════════════════════════════
 RULE 1 (HIGHEST PRIORITY) — save_profile_field is MANDATORY
 ═══════════════════════════════════

frontend/src/lib/useStreamingVoice.ts

@@ -914,16 +914,44 @@ export function useStreamingVoice(
             // exponential-backoff retry (1s/2s/4s, max 3 attempts). The
             // accumulatedText (Web Speech fallback) and accumulated chunks
             // are already captured locally, so retries don't lose the
-            // partial transcript. Each attempt enforces its own 8s timeout
+            // partial transcript. Each attempt enforces its own timeout
             // via the controller signal passed in by retryPostTranscribe.
-            console.debug("[useStreamingVoice] POST /api/transcribe", { bytes: blob.size, mime: blob.type, lang: language });
+            //
+            // KI-302 (2026-05-18) — full-transcript fix. The backend now
+            // SPLITS audio over Sarvam's ~30s REST limit into multiple
+            // chunks and transcribes them sequentially (one Sarvam round
+            // trip per ~25s of speech) so a long utterance is no longer
+            // silently truncated to its first 30s. A fixed 8s client
+            // timeout would abort that legitimately-longer multi-chunk
+            // call mid-flight and force a fall back to the (also often
+            // truncated) Web Speech text — re-introducing the very bug we
+            // are fixing. Scale the per-attempt timeout with the audio
+            // size: an 8s floor for short clips plus a generous budget per
+            // estimated 25s chunk (browser webm/opus ≈ 4 KB/s ⇒ ~100 KB
+            // per 25s chunk; allow ~10s of Sarvam latency per chunk).
+            const APPROX_BYTES_PER_CHUNK = 100_000; // ~25s of webm/opus
+            const estChunks = Math.max(
+              1,
+              Math.ceil(blob.size / APPROX_BYTES_PER_CHUNK),
+            );
+            const attemptTimeoutMs = Math.min(
+              120_000, // hard ceiling — never wait > 2 min on one attempt
+              8_000 + estChunks * 12_000,
+            );
+            console.debug("[useStreamingVoice] POST /api/transcribe", {
+              bytes: blob.size,
+              mime: blob.type,
+              lang: language,
+              estChunks,
+              attemptTimeoutMs,
+            });
             const sarvam = await retryPostTranscribe(async (signal) => {
-              // Race per-attempt 8s timeout against the retry signal so a
+              // Race per-attempt timeout against the retry signal so a
               // hung connection still surfaces as an attempt failure (and
               // triggers the next backoff step) rather than blocking
               // forever. signal aborts when the OUTER retry loop is killed.
               const timeoutCtl = new AbortController();
-              const timer = setTimeout(() => timeoutCtl.abort(), 8000);
+              const timer = setTimeout(() => timeoutCtl.abort(), attemptTimeoutMs);
               const onOuterAbort = () => timeoutCtl.abort();
               signal.addEventListener("abort", onOuterAbort);
               try {

tests/test_scorecard_parity.py

@@ -1,22 +1,32 @@
 """Regression guard — recommendation-path scorecard == marketplace scorecard.
 
-ROOT CAUSE THIS LOCKS DOWN (KI-FULLFACTS, 2026-05-17)
------------------------------------------------------
-`brain_tools._scorecard_signal` used to feed `build_scorecard` the 7-key
-eligibility subset (`_load_policy_facts`), which is below build_scorecard's
-data-completeness floor → every recommended policy came back grade "—" /
-overall 0 → the Bug #71 >=70 fitness gate dropped 100% of them → the
-recommendation `citations` array was always empty → CitedPolicyCards never
-rendered. Meanwhile /api/policies/all scored the SAME policies correctly
-(HDFC ERGO Optima Restore = A) because it feeds the FULL curated layer
-(main._load_curated_facts, KI-219/251 canonical precedence).
-
-The fix makes `_scorecard_signal` reuse that EXACT function. This test makes
-the two paths *provably* unable to diverge again: for a representative
-spread of policies, the recommendation-path GRADE LETTER must equal the
-marketplace GRADE LETTER. (Overall scores may differ by a few points
-because the marketplace overlays EXTRACTED_DIR data on top; the LETTER —
-what _recommendation_fit gates on — must match.)
+ROOT CAUSE THIS LOCKS DOWN
+--------------------------
+1. KI-FULLFACTS (2026-05-17): `brain_tools._scorecard_signal` fed
+   `build_scorecard` the 7-key eligibility subset → every recommended
+   policy came back grade "—"/0 → the >=70 fitness gate dropped 100% →
+   empty `citations` → CitedPolicyCards never rendered.
+
+2. #40 (2026-05-18): even after (1), `_scorecard_signal` kept its OWN
+   doctype-rank + `_merge_curated` re-implementation that *mirrored*
+   `/api/policies/all`. Two parallel implementations drift: marketing-
+   rename / KI-145 variant / multi-doctype ids graded a different file
+   than the marketplace card (e.g. `my:Optima Secure (older variant)`
+   recommended at one grade, marketplace card at another).
+
+THE FIX — SINGLE SOURCE OF TRUTH
+--------------------------------
+`backend.main._marketplace_catalogue()` is the ONE place a card set is
+computed. `backend.main.marketplace_grade(policy_id)` resolves a policy's
+canonical card (UIN-primary) from it. `_scorecard_signal` simply delegates
+to `marketplace_grade`. The two surfaces can no longer diverge because
+there is exactly one computation.
+
+This test makes that *provable*: for the ENTIRE id universe the
+recommender can cite — every marketplace card id, every curated id, every
+extracted stem incl. `__wordings/__brochure/__cis/__prospectus`
+permutations, and every alias name — the recommendation-path GRADE must
+equal the marketplace GRADE. 0 mismatches, by construction.
 """
 from __future__ import annotations
 
@@ -30,59 +40,71 @@ if str(_REPO_ROOT) not in sys.path:
 
 
 class TestScorecardParity(unittest.TestCase):
-    """Rec-path _scorecard_signal grade letter == marketplace grade letter."""
+    """Rec-path _scorecard_signal grade == marketplace grade, everywhere."""
 
     @classmethod
     def setUpClass(cls):
-        from backend.main import _load_curated_facts
-        from backend.scorecard import build_scorecard
-        from backend.brain_tools import _scorecard_signal, _insurer_reviews
+        import backend.main as M
+        from backend.brain_tools import _scorecard_signal
 
+        cls.M = M
         cls._scorecard_signal = staticmethod(_scorecard_signal)
-        cur = _load_curated_facts()
-
-        # Marketplace grade for a policy_id = build_scorecard on the SAME
-        # full curated dict + reviews the /api/policies/all path uses.
-        def _mkt_grade(pid: str):
-            data = cur.get(pid)
-            if not data:
-                return None
-            slug = data.get("insurer_slug") or (
-                pid.split("__", 1)[0] if "__" in pid else ""
-            )
-            sc = build_scorecard(
-                data, insurer_reviews=_insurer_reviews(slug), profile=None
-            )
-            return sc.grade
-
-        cls._mkt_grade = staticmethod(_mkt_grade)
-        # A representative spread across grade bands + insurers, incl. the
-        # user's own policy (HDFC ERGO Optima Restore — must be a strong A).
-        cls._sample = [
-            "hdfc-ergo__optima-restore",
-            "hdfc-ergo__my-health-sampoorna-suraksha",
-            "hdfc-ergo__my-optima-secure",
-            "oriental-insurance__happy-family-floater",
-            "star-health__family-health-optima",
-        ]
-
-    def test_rec_path_grade_matches_marketplace_grade(self):
+        cls._cards = M._marketplace_catalogue(None)
+        cls._cur = M._load_curated_facts()
+
+    def test_full_id_universe_parity(self):
+        """The #40 oracle: rec grade == marketplace grade for EVERY id a
+        recommendation can surface (card ids + curated ids + extracted
+        stems + doctype permutations + aliases). Must be 0 mismatches."""
+        M = self.M
+        ids = {c.policy_id for c in self._cards}
+        ids |= set(self._cur.keys())
+        ids |= {p.stem for p in M.settings.EXTRACTED_DIR.glob("*.json")}
+        for c in self._cards:
+            ids |= set(c.aliases or [])
+
         mismatches = []
-        for pid in self._sample:
-            mkt = self._mkt_grade(pid)
-            if mkt is None:
-                continue  # policy not in curated layer in this env — skip
-            rec = (self._scorecard_signal(pid, profile=None) or {}).get("_grade")
-            if rec != mkt:
-                mismatches.append(f"{pid}: rec={rec!r} marketplace={mkt!r}")
+        checked = 0
+        for pid in sorted(ids):
+            exp = M.marketplace_grade(pid).get("_grade")
+            if exp is None:
+                continue  # not a marketplace card (e.g. regulatory) — skip
+            checked += 1
+            rec = (self._scorecard_signal(pid) or {}).get("_grade")
+            if rec != exp:
+                mismatches.append(f"{pid}: rec={rec!r} marketplace={exp!r}")
+
+        self.assertGreater(checked, 140, "id universe collapsed — guard broke")
         self.assertEqual(
             mismatches, [],
-            "recommendation-path scorecard grade diverged from the "
-            "marketplace grade — the two paths must feed build_scorecard "
-            "the SAME curated facts (see KI-FULLFACTS):\n  "
-            + "\n  ".join(mismatches),
+            f"#40 PARITY BROKEN ({len(mismatches)}/{checked}): the "
+            "recommendation grade diverged from the marketplace grade. "
+            "Both MUST flow through backend.main.marketplace_grade (the "
+            "single source of truth); do not re-implement scoring in "
+            "brain_tools._scorecard_signal.\n  " + "\n  ".join(mismatches[:40]),
         )
 
+    def test_proven_alias_and_doctype_edges(self):
+        """The exact cases #40 used to break: a marketing-rename variant id
+        and doctype-permuted ids must each equal their canonical card."""
+        M = self.M
+        for pid in (
+            "hdfc-ergo__my-optima-secure-older-variant",
+            "hdfc-ergo__optima-restore__brochure",
+            "hdfc-ergo__optima-restore__cis",
+            "royal-sundaram__lifeline__cis",
+            "new-india__mediclaim-policy__brochure",
+        ):
+            mkt = M.marketplace_grade(pid).get("_grade")
+            if mkt is None:
+                continue  # absent in this env — skip, don't false-fail
+            rec = (self._scorecard_signal(pid) or {}).get("_grade")
+            self.assertEqual(
+                rec, mkt,
+                f"{pid}: rec grade {rec!r} != marketplace {mkt!r} "
+                "(alias/doctype canonicalisation regressed)",
+            )
+
     def test_optima_restore_is_a_strong_recommendable_grade(self):
         # The user's own policy — a genuinely strong plan. It must NOT come
         # back as the data-starved "—"/0 sentinel, and must clear the Bug

tests/test_stt_long_audio_chunking.py

@@ -0,0 +1,242 @@
+"""Regression test for KI-302 (2026-05-18) — full voice transcript.
+
+BUG (user-reported, live): "Full transcript still not coming while
+speaking." A long spoken utterance was truncated to a partial transcript.
+
+ROOT CAUSE: Sarvam's saarika REST /speech-to-text endpoint has a hard ~30s
+audio limit. It does NOT 4xx on longer audio — it returns HTTP 200 with a
+`transcript` containing ONLY the first ~30s and silently drops the rest.
+The live-voice hook's grace-window batching deliberately merges multiple
+pause-separated speech bursts into ONE blob, so any real-world utterance
+with natural pauses easily exceeds 30s and was silently cut.
+
+FIX: `SarvamSTT.transcribe` now decodes the audio, and if it exceeds the
+safe REST ceiling it splits it into <= STT_CHUNK_MS chunks (at silence
+boundaries where possible), transcribes each chunk, and concatenates the
+transcripts in order so the COMPLETE utterance survives.
+
+This test pins that contract WITHOUT needing ffmpeg/pydub by stubbing a
+minimal AudioSegment that supports the slicing / len / dBFS / export
+surface the splitter uses, and a fake Sarvam HTTP layer that returns a
+DIFFERENT transcript per 25s window — so a regression to single-shot
+behaviour (only the first window transcribed) fails loudly.
+
+Run:
+    cd /Users/rohitsar/Developer/Insurance\\ Sales\\ Bot
+    PYTHONPATH=$PWD .venv/bin/python -m pytest \
+        tests/test_stt_long_audio_chunking.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import types
+
+import pytest
+
+os.environ.setdefault("SARVAM_API_KEY", "test-key-for-stt-chunking")
+
+from backend.providers.sarvam_stt import (  # noqa: E402
+    STT_CHUNK_MS,
+    SarvamSTT,
+)
+
+
+# ---------------------------------------------------------------------------
+# Minimal pydub.AudioSegment stub.
+# The splitter only uses: len(seg) -> ms, seg[a:b] -> sub-segment,
+# seg.dBFS -> float, seg.export(buf, format=...) , set_frame_rate/channels/
+# sample_width (chained no-ops), AudioSegment.from_file(...).
+# Each "ms" carries a sentinel byte so a concatenated/exported chunk can be
+# decoded back into the exact ms range it represents — that lets the fake
+# Sarvam server return the words spoken in that window.
+# ---------------------------------------------------------------------------
+class FakeAudioSegment:
+    def __init__(self, start_ms: int, end_ms: int):
+        self.start_ms = start_ms
+        self.end_ms = end_ms  # exclusive
+
+    # --- duration -----------------------------------------------------------
+    def __len__(self):
+        return self.end_ms - self.start_ms
+
+    # --- slicing (pydub uses ms-based slicing) ------------------------------
+    def __getitem__(self, sl):
+        if isinstance(sl, slice):
+            lo = 0 if sl.start is None else sl.start
+            hi = len(self) if sl.stop is None else sl.stop
+            lo = max(0, min(lo, len(self)))
+            hi = max(0, min(hi, len(self)))
+            return FakeAudioSegment(self.start_ms + lo, self.start_ms + hi)
+        raise TypeError("only slice indexing used by splitter")
+
+    # --- loudness -----------------------------------------------------------
+    @property
+    def dBFS(self):
+        # Constant moderate loudness everywhere EXCEPT a deliberate silent
+        # gap at [24000, 24400) ms so _split_on_silence has a real pause to
+        # snap to near the 25s ceiling.
+        if self.end_ms - self.start_ms == 0:
+            return float("-inf")
+        if 24000 <= self.start_ms < 24400:
+            return float("-inf")
+        return -20.0
+
+    # --- transcode no-ops ---------------------------------------------------
+    def set_frame_rate(self, _):
+        return self
+
+    def set_channels(self, _):
+        return self
+
+    def set_sample_width(self, _):
+        return self
+
+    def export(self, buf, format="wav"):  # noqa: A002 - pydub signature
+        # Encode the ms range so the fake Sarvam server can recover which
+        # words this chunk covers.
+        buf.write(f"FAKEWAV:{self.start_ms}:{self.end_ms}".encode())
+        buf.seek(0)
+        return buf
+
+
+class FakeAudioSegmentFactory:
+    @staticmethod
+    def from_file(_bio, format=None):  # noqa: A002 - pydub signature
+        # A 92-second utterance (> 3 Sarvam windows of 25s).
+        return FakeAudioSegment(0, 92_000)
+
+
+@pytest.fixture
+def patch_pydub(monkeypatch):
+    fake_pydub = types.ModuleType("pydub")
+    fake_pydub.AudioSegment = FakeAudioSegmentFactory
+    monkeypatch.setitem(sys.modules, "pydub", fake_pydub)
+    yield
+
+
+# ---------------------------------------------------------------------------
+# Fake Sarvam HTTP layer. Returns the words spoken in the chunk's ms window.
+# The "spoken script" is one word per second: word_0 .. word_91. Sarvam's
+# real 30s truncation is simulated by capping any single call at 30000 ms of
+# audio (it ignores everything past 30s of the blob it receives) — exactly
+# the silent-truncation behaviour the fix must defeat by never sending a
+# chunk longer than ~25s.
+# ---------------------------------------------------------------------------
+SARVAM_HARD_LIMIT_MS = 30_000
+
+
+class _FakeResp:
+    def __init__(self, payload):
+        self._payload = payload
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+class _FakeClient:
+    def __init__(self, *a, **k):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+    async def post(self, url, headers=None, files=None, data=None):
+        raw = files["file"][1].read()
+        text = raw.decode()
+        assert text.startswith("FAKEWAV:"), text
+        _, s, e = text.split(":")
+        start_ms, end_ms = int(s), int(e)
+        # Simulate Sarvam's silent 30s truncation on the received blob.
+        capped_end = min(end_ms, start_ms + SARVAM_HARD_LIMIT_MS)
+        words = [
+            f"word_{ms // 1000}"
+            for ms in range(start_ms, capped_end, 1000)
+        ]
+        return _FakeResp({
+            "transcript": " ".join(words),
+            "language_code": "en-IN",
+            "language_probability": 0.99,
+        })
+
+
+@pytest.fixture
+def patch_httpx(monkeypatch):
+    import backend.providers.sarvam_stt as mod
+
+    monkeypatch.setattr(mod.httpx, "AsyncClient", _FakeClient)
+    yield
+
+
+def test_long_utterance_is_fully_transcribed(patch_pydub, patch_httpx):
+    """A 92s utterance must yield ALL 92 words, not just the first ~30."""
+    stt = SarvamSTT()
+    # > 1 KB so the short-audio guard doesn't short-circuit.
+    audio = b"\x00" * 4096
+    result = asyncio.run(
+        stt.transcribe(audio_bytes=audio, audio_format="webm", language_code="en-IN")
+    )
+
+    words = result.text.split()
+    # CORE CONTRACT: every one of the 92 spoken words must survive — the
+    # pre-fix single-shot path returned only word_0..word_29 (Sarvam's 30s
+    # silent truncation). A word landing exactly on a silence-snap seam can
+    # legitimately appear in BOTH adjoining chunks (a duplicated word at the
+    # seam is harmless; a LOST word is the bug). So we assert: (1) no word
+    # is missing, (2) order is non-decreasing, (3) the count is ~92 (never
+    # ~30). De-duplicating consecutive repeats must reproduce the exact
+    # script.
+    expected = [f"word_{i}" for i in range(92)]
+    deduped = [w for i, w in enumerate(words) if i == 0 or w != words[i - 1]]
+    assert deduped == expected, (
+        f"transcript truncated/garbled: deduped {len(deduped)} words "
+        f"(first={deduped[:3]}, last={deduped[-3:]}), expected 92 in order"
+    )
+    # Hard proof this is NOT the pre-fix 30s truncation.
+    assert len(words) >= 90
+    assert "word_91" in words and "word_60" in words
+    assert result.raw.get("chunked") is True
+    # 92s split at <=25s ceiling => at least 4 chunks.
+    assert result.raw.get("chunk_count", 0) >= 4
+    assert result.language_code == "en-IN"
+
+
+def test_short_utterance_single_call_unchanged(patch_pydub, patch_httpx, monkeypatch):
+    """A sub-ceiling clip must take exactly ONE Sarvam call (no behaviour
+    change / no extra latency for the common case)."""
+    # Patch the factory to return a short 8s clip.
+    short = FakeAudioSegment(0, 8_000)
+    monkeypatch.setattr(
+        FakeAudioSegmentFactory, "from_file", staticmethod(lambda *a, **k: short)
+    )
+    stt = SarvamSTT()
+    result = asyncio.run(
+        stt.transcribe(audio_bytes=b"\x00" * 4096, audio_format="webm")
+    )
+    words = result.text.split()
+    assert words == [f"word_{i}" for i in range(8)]
+    # Single-call path does NOT set the chunked marker.
+    assert result.raw.get("chunked") is not True
+
+
+def test_no_chunk_exceeds_sarvam_safe_ceiling(patch_pydub):
+    """Every produced chunk must be <= STT_CHUNK_MS so Sarvam never silently
+    truncates a chunk. This is the core invariant the fix rests on."""
+    seg = FakeAudioSegment(0, 92_000)
+    chunks = SarvamSTT._split_on_silence(seg, STT_CHUNK_MS)
+    assert len(chunks) >= 4
+    for c in chunks:
+        assert len(c) <= STT_CHUNK_MS, f"chunk {len(c)} ms exceeds {STT_CHUNK_MS}"
+    # Chunks must tile the whole utterance with no gap and no overlap.
+    assert chunks[0].start_ms == 0
+    assert chunks[-1].end_ms == 92_000
+    for a, b in zip(chunks, chunks[1:]):
+        assert a.end_ms == b.start_ms, "chunk boundary gap/overlap loses audio"