src/serve.py

#!/usr/bin/env python3
"""
The live demo: speech in -> retrieve -> read -> speech out, in 14 languages.

  python src/serve.py --langs hi,ta,bn --port 8000
  python src/serve.py --no-voice            # text only, no model downloads

DESIGN NOTES THAT MATTER FOR THE SUBMISSION

* EVERY STAGE IS TIMED SEPARATELY and returned in the response. A single
  end-to-end figure hides which part of the budget the voice layer spends, and
  ASR/TTS dominate: reporting one number would make the retrieval work look slow
  when it is the cheap part (search is ~0.1-0.2 ms/query, measured).

* THE READER USES prior_weight=1.0. Measured on real bge-m3 cosine over 606
  answerable queries: reader F1 0.1835 -> 0.2146 (+17%), top-1 passage accuracy
  26.4% -> 49.3%. The oracle-reranker ceiling on the same subset is 0.302, so
  0.088 F1 of reranking headroom is still on the table and is stated as such.

* NO CONFORMAL ABSTENTION IS WIRED IN. The calibration produced tau = 0.9496
  accepting 1 of 28,000 queries, i.e. no threshold reaches alpha at any usable
  coverage. Shipping it would mean abstaining on everything. The machinery is in
  router.py and switches on the moment a scorer earns it; claiming a guarantee
  we cannot honour would be worse than having none.

* THE PAGE IS ONE SELF-CONTAINED FILE with no build step and no CDN, so the
  demo works on a pod with no outbound network.
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from src.reader import LexicalSpanReader  # noqa: E402
from src.router import Passage  # noqa: E402
from src.schema_utils import LANG_NAMES, default_root, pick_device  # noqa: E402
from src.guardrails import GuardrailConfig, Guardrails  # noqa: E402
from src.harness import AskRequest, Harness  # noqa: E402
from src.voice import Timing, VoiceStack, read_wav  # noqa: E402

# MODULE-LEVEL ON PURPOSE. `from __future__ import annotations` stringifies every
# annotation, and FastAPI resolves `req: Request` by looking the name up in the
# MODULE globals. Importing Request inside build_app() leaves it a local, the
# name never resolves, FastAPI falls back to treating `req` as a query
# parameter, and every POST returns 422 "Field required" while GET endpoints
# keep working -- which is exactly the failure that is easiest to miss by hand
# and is what tests/test_serve.py caught.
try:
    from fastapi import FastAPI, Request
    from fastapi.responses import HTMLResponse, JSONResponse
except ImportError:                                    # text-only / no fastapi
    FastAPI = Request = HTMLResponse = JSONResponse = None  # type: ignore

STATE: dict = {}


# ------------------------------------------------------------------ retrieval

class Index:
    """Per-(language, strategy) dense index held in RAM. Flat cosine over
    float16 -- at a few thousand chunks this is well under a millisecond and
    avoids a FAISS dependency the pod may not have.

    Several strategies are held at once so /api/compare can run the SAME query
    through each of them live. That is what turns the chunking study from a
    table into something a judge can poke at."""

    def __init__(self, root: Path, langs: list[str], model: str, max_len: int,
                 strategies: list[str] | None = None):
        import numpy as np
        import torch
        from src.evaluate_retrieval import Embedder

        self.np = np
        idx_dir = root / "index"
        man = json.loads((idx_dir / "manifest.json").read_text())
        self.manifest = man
        self.default_strategy = man.get("default_strategy", "FW")
        self.device = pick_device()
        self.emb = Embedder(model or man["model"], self.device, 32,
                            max_len or man["max_len"])
        self.store: dict = {}
        self.langs: dict = {}

        for key, info in man.get("indices", {}).items():
            lg, st = info["lang"], info["strategy"]
            if lg not in langs:
                continue
            if strategies and st not in strategies:
                continue
            vp, cp = idx_dir / f"{key}.vecs.npy", idx_dir / f"{key}.chunks.json"
            if not (vp.exists() and cp.exists()):
                print(f"  !! missing files for {key}")
                continue
            payload = json.loads(cp.read_text())
            self.store[(lg, st)] = {"vecs": np.load(vp).astype("float32"),
                                    "texts": payload["texts"],
                                    "meta": payload["meta"], "info": info}
            self.langs.setdefault(lg, []).append(st)
            print(f"  {lg} {st:5s}: {len(payload['texts']):,} chunks "
                  f"({info['chunks_per_doc']}/doc, {info['mean_chunk_words']} words)")

        for lg in self.langs:
            self.langs[lg].sort()

    def strategies_for(self, lang: str) -> list[str]:
        return self.langs.get(lang, [])

    def default_for(self, lang: str) -> str:
        """The strategy that SERVES this language.

        strategies_for() is sorted for stable display, so taking [0] would make
        the alphabetically-first strategy the server default -- adding DFC to the
        index for the comparison tab would silently switch what /api/ask uses.
        The manifest's declared default wins whenever it is actually loaded.
        """
        avail = self.langs.get(lang, [])
        if self.default_strategy in avail:
            return self.default_strategy
        return avail[0] if avail else self.default_strategy

    def embed_query(self, query: str):
        return self.emb.encode([query]).cpu().numpy().astype("float32")[0]

    def search(self, query: str, lang: str, k: int = 5, strategy: str | None = None,
               qv=None):
        st = strategy or self.default_for(lang)
        d = self.store.get((lang, st))
        if not d:
            return []
        if qv is None:
            qv = self.embed_query(query)
        sims = d["vecs"] @ qv
        k = min(k, len(sims))
        top = self.np.argpartition(-sims, k - 1)[:k]
        top = top[self.np.argsort(-sims[top])]
        return [Passage(d["meta"][i]["chunk_id"], d["texts"][i], float(sims[i]), lang)
                for i in top]

    def info(self, lang: str, strategy: str) -> dict:
        d = self.store.get((lang, strategy))
        return dict(d["info"]) if d else {}


# ------------------------------------------------------------------ pipeline

def answer(query: str, lang: str, k: int = 5, want_audio: bool = True,
           strategy: str | None = None, audio_wav_b64: str = ""):
    """Thin adapter over the Harness. All orchestration -- retries, timeouts,
    guardrails, error recovery -- lives in harness.py (requirement 5), so there
    is exactly one code path and the demo cannot drift from what was measured."""
    req = AskRequest(query=query, lang=lang, k=k, strategy=strategy,
                     want_audio=want_audio, audio_wav_b64=audio_wav_b64)
    out = STATE["harness"].run(req).as_dict()
    out["abstention"] = ("guardrails active — see /guardrails; conformal "
                         "abstention NOT enabled, see /about")
    return out


def compare(query: str, lang: str, k: int = 5, budget_words: int = 400):
    """Run ONE query through every chunking strategy we hold for this language.

    Two views, because the whole finding is that they disagree:

      top-k        the naive protocol. Confounded -- a strategy that emits more,
                   smaller chunks gets more shots at the same passage, and
                   corr(chunks_per_doc, nDCG@5) = -0.820 across our 7 strategies.
      equal budget as many chunks as fit in `budget_words`. Correcting for the
                   confound drops that correlation to -0.152 AND INVERTS the
                   ranking: FW is 1st on top-k and 5th on hit@400w.

    Both are shown side by side so the judge sees the correction happen on their
    own query rather than taking the table on trust.
    """
    idx, reader = STATE["index"], STATE["reader"]
    strategies = idx.strategies_for(lang)
    if not strategies:
        return {"error": f"no indices for {lang}", "rows": []}

    t0 = time.perf_counter()
    qv = idx.embed_query(query)            # embed ONCE, reuse for every strategy
    embed_ms = (time.perf_counter() - t0) * 1000

    rows = []
    for st in strategies:
        t0 = time.perf_counter()
        # Pull enough to fill the word budget even for large-chunk strategies.
        psgs = idx.search(query, lang, max(k, 12), st, qv=qv)
        search_ms = (time.perf_counter() - t0) * 1000

        used, words = [], 0
        for p in psgs:
            w = len(p.text.split())
            if words + w > budget_words and used:
                break
            used.append(p)
            words += w

        t0 = time.perf_counter()
        span = reader.read(query, psgs[:k]) if psgs else None
        read_ms = (time.perf_counter() - t0) * 1000

        info = idx.info(lang, st)
        rows.append({
            "strategy": st,
            "n_chunks": info.get("n_chunks"),
            "chunks_per_doc": info.get("chunks_per_doc"),
            "mean_chunk_words": info.get("mean_chunk_words"),
            "index_mb": info.get("index_mb"),
            "top_score": round(psgs[0].score, 4) if psgs else 0.0,
            "answer": span.text[:300] if span else "",
            "confidence": round(float(span.score), 4) if span else 0.0,
            "search_ms": round(search_ms, 2), "read_ms": round(read_ms, 1),
            "topk": [{"score": round(p.score, 4), "words": len(p.text.split()),
                      "text": p.text[:260]} for p in psgs[:k]],
            "budget": {"chunks_used": len(used), "words_used": words,
                       "mean_score": round(sum(p.score for p in used)/max(1, len(used)), 4)},
        })

    by_topk = sorted(rows, key=lambda r: -r["top_score"])
    by_budget = sorted(rows, key=lambda r: -r["budget"]["mean_score"])
    return {
        "query": query, "lang": lang, "budget_words": budget_words,
        "embed_ms": round(embed_ms, 1), "rows": rows,
        "rank_topk": [r["strategy"] for r in by_topk],
        "rank_budget": [r["strategy"] for r in by_budget],
        "ranking_changed": [r["strategy"] for r in by_topk] !=
                           [r["strategy"] for r in by_budget],
        "measured_aggregate": MEASURED,
    }


# Aggregate numbers from the full evaluation (hi/ta/bn, 400 queries/lang), shown
# next to the live single-query result so nobody mistakes one query for evidence.
MEASURED = {
    "note": "aggregate over 400 queries x 3 languages — the live panel above is ONE query",
    "top_k_protocol": {"FW": 0.7052, "FCC": 0.6834, "LCTS": 0.6705, "RC": 0.6677,
                       "DFC": 0.6490, "PGC": 0.6253, "SGC": 0.6092},
    "equal_budget_ndcg": {"FCC": 0.6686, "DFC": 0.6582, "LCTS": 0.6532, "FW": 0.6434,
                          "SGC": 0.6245, "RC": 0.6192, "PGC": 0.5890},
    "equal_budget_hit": {"DFC": 0.905, "SGC": 0.875, "FCC": 0.8675, "LCTS": 0.8258,
                         "RC": 0.7658, "FW": 0.795, "PGC": 0.7583},
    "confound": {"corr_chunks_vs_ndcg_topk": -0.820,
                 "corr_chunks_vs_ndcg_budget": -0.152},
}


# ------------------------------------------------------------------ http

def build_app(cors: str = ""):
    if FastAPI is None:
        raise SystemExit("fastapi is not installed: pip install fastapi uvicorn")

    app = FastAPI(title="Voice RAG · MSMARCO-XI", docs_url="/docs")

    # The page is served from Vercel but the API lives here, so the browser
    # treats every /api/* call as cross-origin and blocks it without this.
    origins = [o.strip() for o in cors.split(",") if o.strip()]
    if origins:
        from fastapi.middleware.cors import CORSMiddleware
        app.add_middleware(CORSMiddleware, allow_origins=origins,
                           allow_methods=["*"], allow_headers=["*"])

    @app.get("/", response_class=HTMLResponse)
    def home():
        return PAGE.replace("__LANGS__", json.dumps(
            [{"code": c, "name": LANG_NAMES.get(c, c)}
             for c in sorted(STATE["index"].langs)]))

    @app.get("/health")
    def health():
        v, idx = STATE["voice"], STATE["index"]
        # VoiceStack is loaded tts_only=True, so v.asr_ok is False by
        # construction and reporting it says "not configured" next to a
        # configured key. The ASR provider hangs off the harness; ask the
        # thing that would actually do the transcribing.
        h = STATE.get("harness")
        asr = bool(getattr(h, "asr", None) and h.asr.ok)
        return {"ok": True, "languages": sorted(idx.langs),
                "strategies": {lg: idx.strategies_for(lg) for lg in sorted(idx.langs)},
                "asr": asr, "asr_provider": getattr(getattr(h, "asr", None), "provider", ""),
                "tts": v.tts_ok, "notes": v.notes}

    @app.get("/about")
    def about():
        return {
            "reader": "lexical span reader, prior_weight=1.0",
            "measured": {
                "retrieval_ndcg@5": 0.7052, "retrieval_hit@5": 0.9008,
                "reader_f1_no_prior": 0.1835, "reader_f1_with_prior": 0.2146,
                "reader_f1_oracle_rerank": 0.3023, "extraction_ceiling_f1": 0.7147,
                "answer_extractable_verbatim_pct": 11.1,
            },
            "abstention": "NOT enabled. Conformal calibration gave tau=0.9496, "
                          "which accepts 1 of 28,000 queries — no threshold "
                          "reaches alpha=0.10 at any usable coverage.",
            "known_limits": [
                "Answers average 19.2 words and are human-written sentences; "
                "only 11.1% exist verbatim in the passage, so an extractive "
                "reader is capped at F1 0.715 before any implementation error.",
                "0.088 F1 of reranking headroom remains unclaimed.",
            ],
        }

    @app.post("/api/ask")
    async def ask(req: Request):
        b = await req.json()
        q, lang = (b.get("query") or "").strip(), b.get("lang") or "hi"
        if not q:
            return JSONResponse({"error": "empty query"}, status_code=400)
        return answer(q, lang, int(b.get("k", 5)), bool(b.get("audio", True)),
                      b.get("strategy"))

    @app.get("/guardrails")
    def guardrails_ep():
        """Requirement 6, documented and inspectable."""
        return STATE["guards"].describe()

    @app.get("/harness")
    def harness_ep():
        """Requirement 5: the declared stage graph, with each stage's timeout,
        retry policy and failure behaviour."""
        h = STATE["harness"]
        return {"budget_ms": h.budget_ms,
                "asr_provider": getattr(h.asr, "provider", None),
                "asr_ready": bool(h.asr and h.asr.ok),
                "stages": [{"name": st.name, "timeout_s": st.timeout_s,
                            "retries": st.retries, "on_error": st.on_error.value}
                           for st in h.stages()]}

    @app.post("/api/compare")
    async def compare_ep(req: Request):
        b = await req.json()
        q, lang = (b.get("query") or "").strip(), b.get("lang") or "hi"
        if not q:
            return JSONResponse({"error": "empty query"}, status_code=400)
        return compare(q, lang, int(b.get("k", 5)), int(b.get("budget_words", 400)))

    @app.post("/api/normalise")
    async def normalise_ep(req: Request):
        """Show the Indic normaliser bug on text the judge types.

        A table saying "we fixed a Unicode bug" is a claim. Watching your own
        sentence shatter into consonant fragments under the standard SQuAD
        normaliser, and stay intact under ours, is evidence."""
        import re
        import unicodedata
        from src.textnorm import normalise as fixed
        b = await req.json()
        text = (b.get("text") or "").strip()
        if not text:
            return JSONResponse({"error": "empty text"}, status_code=400)
        # The standard formulation, verbatim, as used across the QA literature.
        broken = re.sub(r"\s+", " ",
                        re.sub(r"[^\w\s]", " ",
                               unicodedata.normalize("NFKC", text).lower())).strip()
        good = fixed(text, True)
        marks = sum(1 for c in text if unicodedata.category(c)[0] == "M")
        return {
            "input": text, "input_words": len(text.split()),
            "standard_normaliser": {"regex": r"[^\w\s]", "output": broken,
                                    "tokens": len(broken.split())},
            "ours": {"method": "Unicode category: P*/S* -> space, Cf -> deleted, "
                               "L*/N*/M* kept", "output": good,
                     "tokens": len(good.split())},
            "combining_marks_in_input": marks,
            "inflation": round(len(broken.split()) / max(1, len(text.split())), 2),
            "why": "Python re \\w is str.isalnum(), which is False for Unicode "
                   "Mn/Mc -- the combining marks carrying the vowels in every "
                   "Brahmic script. They are deleted AND replaced with a space, "
                   "so each word also splits at every mark.",
        }

    @app.post("/api/voice")
    async def voice_ep(req: Request):
        """Speech in. Transcription is a harness stage (Sarvam/ElevenLabs, with
        retries), so a provider hiccup degrades exactly like any other stage."""
        b = await req.json()
        if not b.get("audio_wav_b64"):
            return JSONResponse({"error": "no audio"}, status_code=400)
        return answer("", b.get("lang") or "hi", int(b.get("k", 5)), True,
                      b.get("strategy"), b["audio_wav_b64"])

    return app


PAGE = r"""<!doctype html><html lang="en"><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Voice RAG · 14 Indic languages</title>
<style>
:root{--bg:#0f1115;--fg:#e8eaed;--mut:#9aa0a6;--acc:#4C78A8;--ok:#54A24B;--warn:#F58518;--line:#242832}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--fg);
font:15px/1.6 ui-sans-serif,system-ui,"Noto Sans",sans-serif}
.wrap{max-width:820px;margin:0 auto;padding:32px 20px 80px}
h1{font-size:22px;margin:0 0 4px}.sub{color:var(--mut);font-size:13px;margin-bottom:28px}
.row{display:flex;gap:10px;flex-wrap:wrap;align-items:center;margin-bottom:14px}
select,input,button{font:inherit;border-radius:8px;border:1px solid var(--line);
background:#171a21;color:var(--fg);padding:10px 12px}
input{flex:1;min-width:240px}button{cursor:pointer;border-color:transparent;background:var(--acc);font-weight:600}
button.ghost{background:#171a21;border-color:var(--line)}button:disabled{opacity:.5;cursor:default}
button.rec{background:#E45756}
.card{border:1px solid var(--line);border-radius:12px;padding:16px;margin-top:16px;background:#12151c}
.lbl{font-size:11px;letter-spacing:.08em;text-transform:uppercase;color:var(--mut);margin-bottom:6px}
.ans{font-size:19px;line-height:1.55}
.t{display:flex;gap:16px;flex-wrap:wrap;font-size:12px;color:var(--mut);margin-top:12px}
.t b{color:var(--fg);font-weight:600}
.psg{font-size:13px;color:var(--mut);border-top:1px solid var(--line);padding-top:10px;margin-top:10px}
.pill{display:inline-block;font-size:11px;padding:2px 8px;border-radius:99px;
background:#1d2029;color:var(--mut);margin-right:6px}
.note{font-size:12px;color:var(--warn);margin-top:10px}
a{color:var(--acc)}
.tabs{display:flex;gap:4px;border-bottom:1px solid var(--line);margin-bottom:20px}
.tab{padding:9px 14px;cursor:pointer;color:var(--mut);border-bottom:2px solid transparent;
font-size:14px}
.tab.on{color:var(--fg);border-bottom-color:var(--acc);font-weight:600}
.pane{display:none}.pane.on{display:block}
table{width:100%;border-collapse:collapse;font-size:13px;margin-top:10px}
th,td{text-align:left;padding:7px 9px;border-bottom:1px solid var(--line)}
th{color:var(--mut);font-weight:600;font-size:11px;letter-spacing:.06em;text-transform:uppercase}
td.n{text-align:right;font-variant-numeric:tabular-nums}
.win{color:var(--ok);font-weight:700}.chg{color:var(--warn);font-weight:700}
.bar{height:6px;border-radius:3px;background:var(--acc);display:inline-block;vertical-align:middle}
.mono{font-family:ui-monospace,Menlo,monospace;font-size:12.5px;word-break:break-word}
.bad{color:#E45756}.good{color:var(--ok)}
.hint{font-size:12.5px;color:var(--mut);margin:8px 0 14px}
</style>
<div class="wrap">
<h1>Voice RAG · 14 Indic languages</h1>
<div class="sub">MSMARCO-XI · bge-m3 dense retrieval · extractive reader ·
<a href="/about">measured numbers &amp; known limits</a></div>

<div class="tabs">
  <div class="tab on" data-p="ask">Ask</div>
  <div class="tab" data-p="cmp">Chunking, live</div>
  <div class="tab" data-p="nrm">The Indic bug</div>
</div>

<div class="pane on" id="p-ask">
  <div class="row">
    <select id="lang"></select>
    <input id="q" placeholder="Type a question, or hold the mic…">
    <button id="ask">Ask</button>
    <button id="mic" class="ghost">🎤 Hold</button>
  </div>
  <div id="status" class="sub"></div>
  <div id="out"></div>
</div>

<div class="pane" id="p-cmp">
  <div class="hint">The same query, run through every chunking strategy right now.
  <b>Top-k</b> is the protocol everyone uses; it is confounded, because a strategy
  emitting more, smaller chunks gets more shots at the same passage
  (corr with chunks/doc = &minus;0.820). <b>Equal budget</b> gives every strategy the
  same 400 words of context — correlation drops to &minus;0.152 and the ranking
  inverts.</div>
  <div class="row">
    <select id="clang"></select>
    <input id="cq" placeholder="Ask the same thing of every chunker…">
    <button id="cmp">Compare</button>
  </div>
  <div id="cstatus" class="sub"></div>
  <div id="cout"></div>
</div>

<div class="pane" id="p-nrm">
  <div class="hint">Every token-level metric in QA runs text through
  <span class="mono">re.sub(r"[^\w\s]", " ", s)</span>. In Python, <span class="mono">\w</span>
  is <span class="mono">str.isalnum()</span>, which is <b>False</b> for Unicode
  <span class="mono">Mn/Mc</span> — the combining marks carrying the vowels in every
  Brahmic script. Paste any Indic sentence and watch it shatter.</div>
  <div class="row">
    <input id="nt" value="ফ্ৰেংক গিফৰ্ডে তিনিগৰাকী মহিলাক বিয়া কৰাইছিল">
    <button id="nrm">Normalise</button>
  </div>
  <div id="nout"></div>
</div>
</div>
<script>
const LANGS = __LANGS__;
const $ = s => document.querySelector(s);
const sel = $("#lang");
LANGS.forEach(l => { const o = document.createElement("option");
  o.value = l.code; o.textContent = l.name + " (" + l.code + ")"; sel.appendChild(o); });

function render(d){
  if(d.error){ $("#out").innerHTML = '<div class="card">'+d.error+'</div>'; return; }
  const t = d.timing || {};
  // A gate can block before a stage runs; that stage has no timing at all.
  const ms = v => (v === undefined || v === null) ? "&mdash;" : v + " ms";
  const psg = (d.passages||[]).slice(0,3).map(p =>
    '<div class="psg"><span class="pill">'+p.score.toFixed(3)+'</span>'+
    p.text.replace(/</g,"&lt;")+'</div>').join("");
  $("#out").innerHTML = '<div class="card">'+
    (d.transcript ? '<div class="lbl">heard</div><div style="margin-bottom:12px">'+
      d.transcript.replace(/</g,"&lt;")+
      ((d.asr_dropped||[]).length
        ? '<div class="note">ignored as non-speech: '+
          d.asr_dropped.join(" ").replace(/</g,"&lt;")+'</div>' : '')+
      '</div>' : '')+
    '<div class="lbl">answer</div><div class="ans">'+(d.answer||"—")+'</div>'+
    '<div class="t">'+
      (t.transcribe_ms?'<span>ASR <b>'+t.transcribe_ms+' ms</b></span>':'')+
      '<span>retrieve <b>'+ms(t.retrieve_ms)+'</b></span>'+
      '<span>read <b>'+ms(t.read_ms)+'</b></span>'+
      (t.speak_ms?'<span>TTS <b>'+t.speak_ms+' ms</b></span>':'')+
      '<span>total <b>'+ms(t.total_ms)+'</b></span>'+
      '<span>conf <b>'+d.confidence+'</b></span>'+
    '</div>'+
    '<div class="note">abstention: '+d.abstention+'</div>'+
    '<div class="lbl" style="margin-top:14px">retrieved</div>'+psg+
  '</div>';
  if(d.audio_wav_b64){
    const a = new Audio("data:audio/wav;base64,"+d.audio_wav_b64); a.play().catch(()=>{});
  }
}
async function post(url, body){
  $("#status").textContent = "working…";
  try{
    const r = await fetch(url, {method:"POST", headers:{"Content-Type":"application/json"},
      body: JSON.stringify(body)});
    const d = await r.json(); $("#status").textContent=""; render(d);
  }catch(e){ $("#status").textContent = "request failed: "+e; }
}
$("#ask").onclick = () => { const q = $("#q").value.trim();
  if(q) post("/api/ask", {query:q, lang:sel.value}); };
$("#q").addEventListener("keydown", e => { if(e.key==="Enter") $("#ask").click(); });

// Record while held, downsample to 16 kHz mono WAV in the browser so the
// server never has to decode a container format.
let ctx, rec, chunks = [];
async function start(){
  const stream = await navigator.mediaDevices.getUserMedia({audio:true});
  ctx = new (window.AudioContext||window.webkitAudioContext)();
  // A suspended context never fires onaudioprocess, so every sample is zero
  // and a valid but silent WAV reaches the provider -- which returns an empty
  // transcript and takes the blame for a local capture failure.
  if(ctx.state === "suspended"){ try{ await ctx.resume(); }catch(e){} }
  const src = ctx.createMediaStreamSource(stream);
  const node = ctx.createScriptProcessor(4096,1,1);
  chunks = [];
  node.onaudioprocess = e => chunks.push(new Float32Array(e.inputBuffer.getChannelData(0)));
  src.connect(node); node.connect(ctx.destination);
  rec = {stream, node, src};
  $("#mic").classList.add("rec"); $("#status").textContent = "listening…";
}
function wav(samples, sr){
  const buf = new ArrayBuffer(44 + samples.length*2), v = new DataView(buf);
  const w = (o,s) => { for(let i=0;i<s.length;i++) v.setUint8(o+i, s.charCodeAt(i)); };
  w(0,"RIFF"); v.setUint32(4, 36+samples.length*2, true); w(8,"WAVEfmt ");
  v.setUint32(16,16,true); v.setUint16(20,1,true); v.setUint16(22,1,true);
  v.setUint32(24,sr,true); v.setUint32(28,sr*2,true); v.setUint16(32,2,true);
  v.setUint16(34,16,true); w(36,"data"); v.setUint32(40, samples.length*2, true);
  let o = 44; for(const s of samples){ const x = Math.max(-1,Math.min(1,s));
    v.setInt16(o, x<0 ? x*0x8000 : x*0x7FFF, true); o += 2; }
  let bin = "", bytes = new Uint8Array(buf);
  for(let i=0;i<bytes.length;i++) bin += String.fromCharCode(bytes[i]);
  return btoa(bin);
}
async function stop(){
  if(!rec) return;
  const sr = ctx.sampleRate;
  rec.node.disconnect(); rec.src.disconnect();
  rec.stream.getTracks().forEach(t=>t.stop());
  $("#mic").classList.remove("rec");
  let n = chunks.reduce((a,c)=>a+c.length,0), flat = new Float32Array(n), o = 0;
  for(const c of chunks){ flat.set(c,o); o += c.length; }
  const ratio = sr/16000, out = new Float32Array(Math.floor(n/ratio));
  for(let i=0;i<out.length;i++) out[i] = flat[Math.floor(i*ratio)];
  rec = null; ctx.close();
  if(out.length < 1600){ $("#status").textContent = "too short — hold while speaking"; return; }
  // A silent WAV is a valid WAV. Measure before sending so the message names
  // the microphone rather than the provider.
  let peak = 0;
  for(let i=0;i<out.length;i++){ const a = Math.abs(out[i]); if(a>peak) peak=a; }
  if(peak < 0.01){
    $("#status").textContent = "no sound captured (peak " + peak.toFixed(4) +
      ") — check the input device";
    return;
  }
  post("/api/voice", {audio_wav_b64: wav(out,16000), lang: sel.value});
}
// ---- tabs
document.querySelectorAll(".tab").forEach(t => t.onclick = () => {
  document.querySelectorAll(".tab").forEach(x=>x.classList.remove("on"));
  document.querySelectorAll(".pane").forEach(x=>x.classList.remove("on"));
  t.classList.add("on"); $("#p"+"-"+t.dataset.p).classList.add("on");
});

// ---- chunking comparison
const csel = $("#clang");
LANGS.forEach(l => { const o=document.createElement("option");
  o.value=l.code; o.textContent=l.name+" ("+l.code+")"; csel.appendChild(o); });

function cmpRender(d){
  if(d.error){ $("#cout").innerHTML='<div class="card">'+d.error+'</div>'; return; }
  const mx = Math.max(...d.rows.map(r=>r.top_score))||1;
  const bx = Math.max(...d.rows.map(r=>r.budget.mean_score))||1;
  const winT = d.rank_topk[0], winB = d.rank_budget[0];
  let t = '<div class="card"><div class="lbl">this query, live</div><table><tr>'+
    '<th>strategy</th><th>chunks/doc</th><th>words/chunk</th>'+
    '<th>top-1 score</th><th>equal-budget mean</th><th>chunks used</th>'+
    '<th>search</th></tr>';
  d.rows.forEach(r=>{
    t += '<tr><td><b class="'+(r.strategy===winT?'win':'')+'">'+r.strategy+'</b></td>'+
      '<td class="n">'+r.chunks_per_doc+'</td><td class="n">'+r.mean_chunk_words+'</td>'+
      '<td class="n">'+r.top_score.toFixed(4)+' <span class="bar" style="width:'+
        (44*r.top_score/mx).toFixed(0)+'px"></span></td>'+
      '<td class="n '+(r.strategy===winB?'win':'')+'">'+r.budget.mean_score.toFixed(4)+
        ' <span class="bar" style="width:'+(44*r.budget.mean_score/bx).toFixed(0)+'px"></span></td>'+
      '<td class="n">'+r.budget.chunks_used+' / '+r.budget.words_used+'w</td>'+
      '<td class="n">'+r.search_ms+' ms</td></tr>';
  });
  t += '</table>';
  t += '<div class="t"><span>top-k order <b>'+d.rank_topk.join(" › ")+'</b></span>'+
       '<span>equal budget <b>'+d.rank_budget.join(" › ")+'</b></span></div>';
  t += d.ranking_changed
    ? '<div class="note"><span class="chg">the ranking changed</span> once every '+
      'strategy got the same context budget — that is the confound, on your query.</div>'
    : '<div class="note">ranking held for this query. Aggregate over 1,200 queries '+
      'below is where it inverts.</div>';

  const m = d.measured_aggregate;
  const fmt = o => Object.entries(o).map(([k,v])=>k+" "+v.toFixed(4)).join("  ·  ");
  t += '<div class="lbl" style="margin-top:18px">aggregate — 400 queries × 3 languages</div>'+
    '<div class="psg mono">top-k nDCG@5 &nbsp;'+fmt(m.top_k_protocol)+'</div>'+
    '<div class="psg mono">equal-budget nDCG &nbsp;'+fmt(m.equal_budget_ndcg)+'</div>'+
    '<div class="psg mono">equal-budget hit &nbsp;'+fmt(m.equal_budget_hit)+'</div>'+
    '<div class="psg mono">corr(chunks/doc, nDCG): top-k <span class="bad">'+
      m.confound.corr_chunks_vs_ndcg_topk+'</span> → equal-budget <span class="good">'+
      m.confound.corr_chunks_vs_ndcg_budget+'</span></div>';

  t += '<div class="lbl" style="margin-top:18px">answers</div>';
  d.rows.forEach(r=>{ t += '<div class="psg"><span class="pill">'+r.strategy+'</span>'+
    (r.answer||"—").replace(/</g,"&lt;")+'</div>'; });
  $("#cout").innerHTML = t + '</div>';
}
async function doCompare(){
  const q = $("#cq").value.trim(); if(!q) return;
  $("#cstatus").textContent = "running every chunker…";
  try{
    const r = await fetch("/api/compare",{method:"POST",
      headers:{"Content-Type":"application/json"},
      body:JSON.stringify({query:q, lang:csel.value})});
    $("#cstatus").textContent=""; cmpRender(await r.json());
  }catch(e){ $("#cstatus").textContent = "failed: "+e; }
}
$("#cmp").onclick = doCompare;
$("#cq").addEventListener("keydown", e=>{ if(e.key==="Enter") doCompare(); });

// ---- normaliser
async function doNorm(){
  const text = $("#nt").value; if(!text.trim()) return;
  const r = await fetch("/api/normalise",{method:"POST",
    headers:{"Content-Type":"application/json"}, body:JSON.stringify({text})});
  const d = await r.json();
  if(d.error){ $("#nout").innerHTML='<div class="card">'+d.error+'</div>'; return; }
  const infl = d.inflation > 1.05;
  $("#nout").innerHTML = '<div class="card">'+
    '<div class="lbl">input · '+d.input_words+' words · '+
      d.combining_marks_in_input+' combining marks</div>'+
    '<div class="ans" style="font-size:16px">'+d.input.replace(/</g,"&lt;")+'</div>'+
    '<div class="lbl" style="margin-top:16px">standard <span class="mono">[^\\w\\s]</span>'+
      ' → <span class="'+(infl?'bad':'')+'">'+d.standard_normaliser.tokens+' tokens</span></div>'+
    '<div class="mono '+(infl?'bad':'')+'">'+d.standard_normaliser.output+'</div>'+
    '<div class="lbl" style="margin-top:16px">ours → <span class="good">'+
      d.ours.tokens+' tokens</span></div>'+
    '<div class="mono good">'+d.ours.output+'</div>'+
    (infl?'<div class="note">'+d.inflation+'× inflation — every token-F1, exact match '+
      'and overlap score computed through this was comparing consonant fragments, '+
      'not words.</div>':'<div class="note">this script is unaffected (Latin and '+
      'Arabic have no spacing combining marks) — which is exactly why the bug '+
      'survives Latin-only test suites.</div>')+
    '<div class="psg">'+d.why+'</div></div>';
}
$("#nrm").onclick = doNorm;
$("#nt").addEventListener("keydown", e=>{ if(e.key==="Enter") doNorm(); });
document.addEventListener("DOMContentLoaded", doNorm); doNorm();

const mic = $("#mic");
/* Was mousedown/mouseup on the button itself, which loses the recording in
   three ordinary ways: a touch device never fires mouse events, releasing the
   pointer anywhere off the button means mouseup never arrives and the take
   runs forever, and the keyboard cannot reach it at all. Pointer events cover
   mouse and touch; the release is caught on window; and space/enter toggle. */
mic.addEventListener("pointerdown", e => { e.preventDefault(); start(); });
addEventListener("pointerup", () => { if(rec) stop(); });
addEventListener("pointercancel", () => { if(rec) stop(); });
mic.addEventListener("keydown", e => {
  if((e.key === " " || e.key === "Enter") && !rec){ e.preventDefault(); start(); }
});
mic.addEventListener("keyup", e => {
  if((e.key === " " || e.key === "Enter") && rec){ e.preventDefault(); stop(); }
});
mic.addEventListener("touchstart", e=>{e.preventDefault();start();});
mic.addEventListener("touchend", e=>{e.preventDefault();stop();});
</script></html>"""


def parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser()
    ap.add_argument("--root", type=Path, default=None)
    ap.add_argument("--langs", default=None)
    ap.add_argument("--host", default="0.0.0.0")
    ap.add_argument("--port", type=int, default=8000)
    ap.add_argument("--k", type=int, default=5)
    ap.add_argument("--model", default=None)
    ap.add_argument("--max-len", type=int, default=0)
    ap.add_argument("--answer-mode", default="sentence", choices=["span", "sentence"],
                    help="sentence = speakable (default for serving)")
    ap.add_argument("--no-voice", action="store_true")
    ap.add_argument("--asr-provider", default=None, choices=[None, "sarvam", "elevenlabs"],
                    help="requirement 1; defaults to $VOICERAG_ASR")
    ap.add_argument("--prior-weight", type=float, default=1.0,
                    help="measured best on real bge-m3 scores; 0 disables")
    ap.add_argument("--cors", default="",
                    help="origins allowed to call /api/* — the Vercel page. "
                         "Comma-separated. Empty = same-origin only.")
    ap.add_argument("--prewarm-tts", default="",
                    help="langs to warm, or 'all'. MMS voices are lazy AND "
                         "downloaded on first use; a cold request once cost "
                         "23,511 ms inside speak alone.")
    return ap


def bootstrap(args):
    """Everything main() does except bind a port.

    Hugging Face's Docker SDK is a paid feature, so the Space runs the Gradio
    base image and owns its own uvicorn. app.py calls this and mounts the
    result; nothing below the HTTP layer knows the difference.
    """
    root = args.root.expanduser().resolve() if args.root else default_root()
    man = root / "index" / "manifest.json"
    if not man.exists():
        raise SystemExit(f"no index at {root/'index'} — run src/index_build.py first")
    # index_build.py writes indices/model/strategies -- never a "languages"
    # key. Reading one raised KeyError for every caller that omitted --langs.
    manifest = json.loads(man.read_text())
    langs = args.langs.split(",") if args.langs else sorted(
        {v["lang"] for v in manifest.get("indices", {}).values()})

    print(f"==> loading index from {root/'index'}")
    STATE["index"] = Index(root, langs, args.model, args.max_len)
    if not STATE["index"].langs:
        raise SystemExit("no languages loaded — build the index first")
    # answer_mode="sentence" for SERVING: the answer is spoken aloud, and a
    # trimmed span starts mid-clause and stops mid-number. Still fully
    # extractive, so guardrail gate 4 (verbatim substring) is unaffected.
    STATE["reader"] = LexicalSpanReader(prior_weight=args.prior_weight,
                                        answer_mode=args.answer_mode)
    print(f"==> reader prior_weight={args.prior_weight} answer_mode={args.answer_mode}")

    # REQUIREMENT 1: speech-to-text must be Sarvam or ElevenLabs. No local
    # fallback -- a stand-in that quietly works would be a compliance failure
    # dressed up as a success, so this reports NOT READY and the voice path
    # returns a clear error instead.
    from src.asr_api import CloudASR
    asr = None if args.no_voice else CloudASR(args.asr_provider)
    if asr:
        print(f"==> ASR: {asr.why}")
        if not asr.ok:
            print("    !! voice input will return an error until the key is set")

    print("==> loading TTS (MMS — the brief specifies the speech-to-TEXT provider only)")
    STATE["voice"] = VoiceStack.load(enable=not args.no_voice, tts_only=True)
    for n in STATE["voice"].notes:
        print(f"    {n}")
    if args.prewarm_tts and STATE["voice"].tts_ok:
        warm = (sorted(STATE["index"].langs) if args.prewarm_tts == "all"
                else [x.strip() for x in args.prewarm_tts.split(",") if x.strip()])
        for lg in warm:
            t0 = time.time()
            try:
                STATE["voice"].tts.speak("नमस्ते", lg)
                print(f"    warmed {lg} in {(time.time() - t0) * 1000:.0f} ms")
            except Exception as e:
                print(f"    !! {lg} has no MMS voice or failed: {str(e)[:80]}")

    # GuardrailConfig.load is the single source for gate 3: it reads the signal
    # and its threshold from the same file in the same call, and refuses a
    # threshold written on the other signal's scale.
    tp = root / "results" / "guardrail_calibration.json"
    cfg = GuardrailConfig.load(tp)
    depth = f" over top {cfg.gate_depth}" if cfg.topic_signal == "spread" else ""
    if cfg.topic_calibrated:
        print(f"==> guardrails: gate 3 = {cfg.topic_signal} < {cfg.tau_topic}{depth}")
        print(f"    {cfg.topic_provenance}")
    else:
        print(f"==> guardrails: gate 3 = {cfg.topic_signal} < {cfg.tau_topic}{depth}")
        print(f"    !! PLACEHOLDER — no {tp.name}; rebuild the index to calibrate")
    STATE["guards"] = Guardrails(cfg)
    STATE["harness"] = Harness(STATE["index"], STATE["reader"], STATE["voice"],
                               asr=asr, guards=STATE["guards"], budget_ms=200.0)

    return build_app(args.cors)


def main() -> int:
    # Imported after parsing so --help works without the serving deps
    # installed, which is how this file gets checked on a laptop.
    args = parser().parse_args()
    import uvicorn

    app = bootstrap(args)
    print(f"\n==> http://{args.host}:{args.port}  "
          f"({len(STATE['index'].langs)} languages)")
    uvicorn.run(app, host=args.host, port=args.port, log_level="warning")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())