app.py

# -*- coding: utf-8 -*-
"""
HUDA-Net v43-AR | Frozen Neural Decision Core + Arabic/English Query Layer | Hugging Face Spaces

Active query path:
TOP2 Retriever -> Stage4-v2 Reranker -> RRF k=60 -> Stage5 Calibrator
-> fixed threshold 0.7023535690146311 -> visible-evidence grounding.

This active app has no semantic keyword rules, manual synonym expansion,
dialect dictionary, or regex intent classification. Arabic and English queries are
accepted by the interface; the frozen decision core and confidence policy are unchanged.
"""
from __future__ import annotations

# Hugging Face ZeroGPU must be imported before Torch/CUDA initialization.
import spaces

import hashlib
import html
import importlib.util
import json
import os
import re
import threading
import time
from pathlib import Path
from typing import Any

import gradio as gr
import numpy as np
import pandas as pd
from huggingface_hub import snapshot_download

APP_VERSION = "43.0.0-AR"
UI_VERSION = "4.0-MVP"
TRANSLATION_MODEL_REPO = os.getenv("HUDANET_AR_EN_TRANSLATOR_REPO", "Helsinki-NLP/opus-mt-ar-en").strip()
MODEL_REPO = os.getenv("HUDANET_V43_MODEL_REPO", "dakheel/hudanet-v43-ar").strip()
CORPUS_REPO = os.getenv("HUDANET_V43_CORPUS_REPO", "dakheel/hudanet-v43-ar-runtime-corpus").strip()
HF_TOKEN = os.getenv("HF_TOKEN", "").strip()

EXPECTED_TOP2_SHA256 = "b1e27b53b616f98bb406dcb94a88583a2b1d659dedb0a45464a310c4200d5173"
EXPECTED_RERANKER_SHA256 = "82e301c2aef211540b628f4b5517d81de79bf5c8e0cf1f69524cf0dec40d0fbc"
EXPECTED_THRESHOLD = 0.7023535690146311
EXPECTED_RRF_K = 60
EXPECTED_CORPUS_ROWS = 2796

ROOT = Path("/tmp/hudanet_v43_ar")
MODEL_ROOT = ROOT / "model_release"
CORPUS_ROOT = ROOT / "runtime_corpus"

os.environ.setdefault("HF_HUB_DISABLE_TELEMETRY", "1")
os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

# Script detection is presentation/query-routing only, not a semantic classifier.
_ARABIC_CHAR_RE = re.compile(r"[\u0600-\u06FF]")
_LATIN_CHAR_RE = re.compile(r"[A-Za-z]")
_RUNTIME_LOCK = threading.Lock()
_CACHE_LOCK = threading.Lock()
_TRANSLATION_LOCK = threading.Lock()
_QUERY_CACHE: dict[str, dict[str, Any]] = {}
_TRANSLATION_CACHE: dict[str, str] = {}
_TRANSLATOR_TOKENIZER = None
_TRANSLATOR_MODEL = None
_CACHE_MAX = 96
_TRANSLATION_CACHE_MAX = 256


def _sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def _recursive_fingerprint(path: Path) -> str:
    path = path.resolve()
    files = sorted(p for p in path.rglob("*") if p.is_file())
    if not files:
        raise RuntimeError(f"No files found under model directory: {path}")
    h = hashlib.sha256()
    for p in files:
        rel = str(p.relative_to(path)).replace("\\", "/")
        size = p.stat().st_size
        sha = _sha256_file(p)
        h.update(rel.encode("utf-8"))
        h.update(b"\x00")
        h.update(str(size).encode("ascii"))
        h.update(b"\x00")
        h.update(sha.encode("ascii"))
        h.update(b"\n")
    return h.hexdigest()


def _load_json(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def _download_release_assets() -> tuple[Path, Path]:
    ROOT.mkdir(parents=True, exist_ok=True)
    MODEL_ROOT.mkdir(parents=True, exist_ok=True)
    CORPUS_ROOT.mkdir(parents=True, exist_ok=True)

    print(f"⬇️ Downloading frozen v43-AR model release: {MODEL_REPO}")
    snapshot_download(
        repo_id=MODEL_REPO,
        repo_type="model",
        token=HF_TOKEN or None,
        local_dir=str(MODEL_ROOT),
        allow_patterns=[
            "retriever/**",
            "reranker/**",
            "runtime/**",
            "config/**",
            "release/**",
            "reports/STAGE7_FINAL_BLIND_REPORT.json",
        ],
        max_workers=8,
    )

    if not HF_TOKEN:
        raise RuntimeError(
            "HF_TOKEN is missing. The v43 Arabic runtime corpus is private. "
            "Keep the existing read-only HF_TOKEN under Space Settings -> Secrets."
        )

    print(f"⬇️ Downloading private frozen Arabic runtime corpus: {CORPUS_REPO}")
    snapshot_download(
        repo_id=CORPUS_REPO,
        repo_type="dataset",
        token=HF_TOKEN,
        local_dir=str(CORPUS_ROOT),
        allow_patterns=[
            "v43_ar_runtime_corpus.parquet",
            "v43_ar_passage_embeddings.npy",
            "RUNTIME_CORPUS_MANIFEST.json",
        ],
        max_workers=8,
    )
    return MODEL_ROOT, CORPUS_ROOT


def _verify_release(model_root: Path, corpus_root: Path) -> tuple[dict, dict]:
    release = _load_json(model_root / "release" / "RELEASE_MANIFEST.json")
    corpus_manifest = _load_json(corpus_root / "RUNTIME_CORPUS_MANIFEST.json")

    if release.get("release") != "HUDA-Net v43-AR":
        raise RuntimeError("Unexpected model release identity.")
    if int(release.get("rrf_k", -1)) != EXPECTED_RRF_K:
        raise RuntimeError("Frozen RRF k mismatch.")
    if abs(float(release.get("abstention_threshold", -1)) - EXPECTED_THRESHOLD) > 1e-12:
        raise RuntimeError("Frozen abstention threshold mismatch.")
    if release.get("hidden_answer_pool_allowed") is not False:
        raise RuntimeError("Release manifest violates hidden-answer-pool contract.")

    print("🔐 Verifying frozen retriever fingerprint...")
    top2_hash = _recursive_fingerprint(model_root / "retriever")
    if top2_hash != EXPECTED_TOP2_SHA256 or top2_hash != release.get("top2_recursive_sha256"):
        raise RuntimeError(f"Retriever fingerprint mismatch: {top2_hash}")

    print("🔐 Verifying frozen reranker fingerprint...")
    rer_hash = _recursive_fingerprint(model_root / "reranker")
    if rer_hash != EXPECTED_RERANKER_SHA256 or rer_hash != release.get("reranker_recursive_sha256"):
        raise RuntimeError(f"Reranker fingerprint mismatch: {rer_hash}")

    corpus_file = corpus_root / "v43_ar_runtime_corpus.parquet"
    embeddings_file = corpus_root / "v43_ar_passage_embeddings.npy"
    if _sha256_file(corpus_file) != corpus_manifest.get("corpus_sha256"):
        raise RuntimeError("Runtime corpus SHA256 mismatch.")
    if _sha256_file(embeddings_file) != corpus_manifest.get("embeddings_sha256"):
        raise RuntimeError("Runtime embedding SHA256 mismatch.")
    if int(corpus_manifest.get("records", -1)) != EXPECTED_CORPUS_ROWS:
        raise RuntimeError("Runtime corpus row count manifest mismatch.")
    if corpus_manifest.get("question_labels_included") is not False:
        raise RuntimeError("Runtime corpus must not contain evaluation question labels.")
    if corpus_manifest.get("hidden_answer_field_included") is not False:
        raise RuntimeError("Runtime corpus must not contain a hidden answer field.")
    return release, corpus_manifest


def _load_frozen_runtime(model_root: Path, corpus_root: Path):
    runtime_py = model_root / "runtime" / "runtime_v43_ar.py"
    spec = importlib.util.spec_from_file_location("hudanet_v43_frozen_runtime", runtime_py)
    if spec is None or spec.loader is None:
        raise RuntimeError("Could not load frozen runtime_v43_ar.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    runtime = module.HudanetArabicV43Runtime(
        top2_model_dir=str(model_root / "retriever"),
        reranker_model_dir=str(model_root / "reranker"),
        calibrator_path=str(model_root / "config" / "RANKING_CONFIDENCE_CALIBRATOR.joblib"),
        policy_path=str(model_root / "config" / "ABSTENTION_POLICY.json"),
    )

    corpus = pd.read_parquet(corpus_root / "v43_ar_runtime_corpus.parquet").fillna("")
    embeddings = np.asarray(
        np.load(corpus_root / "v43_ar_passage_embeddings.npy", allow_pickle=False),
        dtype=np.float32,
    )

    required = {
        "record_id", "passage_ar", "book_ar", "author_ar", "title", "chapter",
        "category", "ruling", "page_number", "source_file",
    }
    missing = required - set(corpus.columns)
    if missing:
        raise RuntimeError(f"Runtime corpus missing columns: {sorted(missing)}")
    if len(corpus) != EXPECTED_CORPUS_ROWS:
        raise RuntimeError(f"Expected {EXPECTED_CORPUS_ROWS} corpus rows; found {len(corpus)}")
    if corpus["record_id"].astype(str).nunique() != EXPECTED_CORPUS_ROWS:
        raise RuntimeError("Runtime corpus record_id values are not unique.")
    if any(c in corpus.columns for c in ("question_ar_gold", "split", "leakage_family_id", "answer_ar_gold")):
        raise RuntimeError("Evaluation labels or hidden answer fields leaked into runtime corpus.")
    if embeddings.ndim != 2 or embeddings.shape[0] != len(corpus):
        raise RuntimeError(f"Embedding shape mismatch: {embeddings.shape} vs {len(corpus)} rows")
    if not np.isfinite(embeddings).all():
        raise RuntimeError("Runtime embeddings contain non-finite values.")

    runtime.corpus = corpus.reset_index(drop=True).copy()
    runtime.passage_embeddings = embeddings
    if abs(float(runtime.threshold) - EXPECTED_THRESHOLD) > 1e-12:
        raise RuntimeError("Loaded runtime threshold is not the frozen Stage5 threshold.")
    return runtime, corpus


_BOOT_STARTED = time.perf_counter()
_MODEL_ROOT, _CORPUS_ROOT = _download_release_assets()
_RELEASE_MANIFEST, _CORPUS_MANIFEST = _verify_release(_MODEL_ROOT, _CORPUS_ROOT)
RUNTIME, CORPUS = _load_frozen_runtime(_MODEL_ROOT, _CORPUS_ROOT)
BOOT_SECONDS = time.perf_counter() - _BOOT_STARTED
print(
    f"✅ HUDA-Net v43-AR frozen runtime ready | rows={len(CORPUS):,} | "
    f"rrf_k={EXPECTED_RRF_K} | threshold={EXPECTED_THRESHOLD:.6f} | {BOOT_SECONDS:.1f}s"
)


def _clean_query(value: Any) -> str:
    text = str(value or "").replace("\x00", " ").strip()
    text = re.sub(r"\s+", " ", text)
    return text[:1200]


def _is_arabic_enough(text: str) -> bool:
    return len(_ARABIC_CHAR_RE.findall(text)) >= 3



def _detect_query_language(text: str) -> str:
    """Detect Arabic vs English from the query's dominant writing script.

    This is used only for UI direction and response language. It does not alter
    retrieval semantics, ranking features, RRF, calibration, or the threshold.
    """
    value = str(text or "")
    ar = len(_ARABIC_CHAR_RE.findall(value))
    en = len(_LATIN_CHAR_RE.findall(value))
    if ar == 0 and en == 0:
        return "ar"
    if ar >= en * 1.2:
        return "ar"
    if en >= ar * 1.2:
        return "en"
    for ch in value:
        if _ARABIC_CHAR_RE.match(ch):
            return "ar"
        if _LATIN_CHAR_RE.match(ch):
            return "en"
    return "ar"


def _qdir(lang: str) -> str:
    return "ltr" if lang == "en" else "rtl"


def _qt(lang: str, ar: str, en: str, tag: str = "span", cls: str = "") -> str:
    value = en if lang == "en" else ar
    direction = _qdir(lang)
    lang_attr = "en" if lang == "en" else "ar"
    class_attr = f' class="{cls}"' if cls else ""
    return f'<{tag}{class_attr} dir="{direction}" lang="{lang_attr}">{value}</{tag}>'


def _load_translation_model():
    global _TRANSLATOR_TOKENIZER, _TRANSLATOR_MODEL
    if _TRANSLATOR_TOKENIZER is not None and _TRANSLATOR_MODEL is not None:
        return _TRANSLATOR_TOKENIZER, _TRANSLATOR_MODEL
    with _TRANSLATION_LOCK:
        if _TRANSLATOR_TOKENIZER is not None and _TRANSLATOR_MODEL is not None:
            return _TRANSLATOR_TOKENIZER, _TRANSLATOR_MODEL
        print(f"⬇️ Loading Arabic→English presentation translator: {TRANSLATION_MODEL_REPO}")
        from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
        tok = AutoTokenizer.from_pretrained(TRANSLATION_MODEL_REPO)
        model = AutoModelForSeq2SeqLM.from_pretrained(TRANSLATION_MODEL_REPO)
        model.eval()
        model.to("cpu")
        _TRANSLATOR_TOKENIZER = tok
        _TRANSLATOR_MODEL = model
        return tok, model


def _translation_cache_put(source: str, translated: str) -> None:
    with _TRANSLATION_LOCK:
        if source in _TRANSLATION_CACHE:
            _TRANSLATION_CACHE.pop(source, None)
        _TRANSLATION_CACHE[source] = translated
        while len(_TRANSLATION_CACHE) > _TRANSLATION_CACHE_MAX:
            first = next(iter(_TRANSLATION_CACHE))
            _TRANSLATION_CACHE.pop(first, None)


def _translate_ar_to_en(text: str) -> str:
    source = str(text or "").strip()
    if not source:
        return ""
    with _TRANSLATION_LOCK:
        cached = _TRANSLATION_CACHE.get(source)
    if cached:
        return cached

    tok, model = _load_translation_model()
    import torch

    # Chunk by tokenizer IDs so long source passages are translated completely
    # instead of being silently truncated by the translation model.
    ids = tok(source, add_special_tokens=False).input_ids
    chunk_size = 380
    chunks = [tok.decode(ids[i:i + chunk_size], skip_special_tokens=True) for i in range(0, len(ids), chunk_size)] or [source]
    translated_parts: list[str] = []
    with _TRANSLATION_LOCK, torch.inference_mode():
        for start in range(0, len(chunks), 6):
            batch_text = chunks[start:start + 6]
            batch = tok(batch_text, return_tensors="pt", padding=True, truncation=True, max_length=512)
            generated = model.generate(
                **batch,
                num_beams=3,
                max_new_tokens=512,
                early_stopping=True,
            )
            translated_parts.extend(tok.batch_decode(generated, skip_special_tokens=True))
    translated = " ".join(x.strip() for x in translated_parts if x and x.strip()).strip()
    if not translated:
        raise RuntimeError("Arabic-to-English translation returned empty text.")
    _translation_cache_put(source, translated)
    return translated


def _prepare_english_translations(result: dict, count: int) -> dict[str, str]:
    """Translate visible Arabic evidence for English-query presentation only.

    The frozen retrieval/reranking/calibration decision is completed before this
    function is called. These translations are never fed back into ranking.
    """
    candidates = list(result.get("candidates", []) or [])[: max(1, min(int(count), 8))]
    selected = result.get("selected_visible_evidence", {}) or {}
    items = candidates + ([selected] if selected else [])
    out: dict[str, str] = {}
    for item in items:
        rid = str(item.get("record_id", ""))
        passage = str(item.get("passage_ar", "")).strip()
        if rid and passage and rid not in out:
            out[rid] = _translate_ar_to_en(passage)
    return out


def _esc(value: Any) -> str:
    return html.escape(str(value or ""), quote=True)


def _fmt_num(value: Any, digits: int = 3) -> str:
    try:
        return f"{float(value):.{digits}f}"
    except Exception:
        return ""


def _ui(ar: str, en: str, tag: str = "span", extra_class: str = "") -> str:
    cls = ("i18n " + extra_class).strip()
    return (
        f'<{tag} class="{cls} i18n-ar" dir="rtl" lang="ar">{ar}</{tag}>'
        f'<{tag} class="{cls} i18n-en" dir="ltr" lang="en">{en}</{tag}>'
    )


def _confidence_meter(probability: float, threshold: float, answered: bool) -> str:
    pct = max(0.0, min(100.0, probability * 100.0))
    threshold_pct = max(0.0, min(100.0, threshold * 100.0))
    state = "pass" if answered else "hold"
    return (
        f'<div class="confidence-meter {state}" aria-label="Confidence meter">'
        '<div class="confidence-scale">'
        f'<span class="confidence-fill" style="width:{pct:.2f}%"></span>'
        f'<span class="threshold-pin" style="left:{threshold_pct:.2f}%" title="{threshold_pct:.1f}%"></span>'
        '</div>'
        '<div class="confidence-legend">'
        f'<span>{_ui("الثقة", "Confidence")}<strong>{pct:.1f}%</strong></span>'
        f'<span>{_ui("حد الإجابة", "Answer threshold")}<strong>{threshold_pct:.1f}%</strong></span>'
        '</div>'
        '</div>'
    )


def _candidate_card(candidate: dict, number: int, selected_id: str, answered: bool, query_lang: str, translations: dict[str, str]) -> str:
    rid = str(candidate.get("record_id", ""))
    selected = rid == selected_id
    book = _esc(candidate.get("book_ar", "") or "")
    author = _esc(candidate.get("author_ar", ""))
    title = _esc(candidate.get("title", ""))
    page = _esc(candidate.get("page_number", ""))
    passage_ar = str(candidate.get("passage_ar", "")).strip()

    source_parts = [x for x in (book, author) if x]
    source_line = " · ".join(source_parts) or "—"
    if page:
        source_line += f" · {_qt(query_lang, 'ص', 'p.')} {page}"

    badge = ""
    if selected and answered:
        badge = _qt(query_lang, "المصدر الداعم", "Supporting source", "span", "source-badge")

    if query_lang == "en":
        translated = _esc(translations.get(rid, ""))
        if not translated:
            raise RuntimeError(f"Missing English translation for visible evidence record {rid}")
        body = f'<div class="evidence-passage" dir="ltr" lang="en">{translated}</div>'
        body += (
            '<details class="source-original">'
            '<summary>Original Arabic</summary>'
            f'<div class="evidence-passage source-arabic" dir="rtl" lang="ar">{_esc(passage_ar)}</div>'
            '</details>'
        )
        heading = f"Source {number}"
    else:
        body = f'<div class="evidence-passage source-arabic" dir="rtl" lang="ar">{_esc(passage_ar)}</div>'
        heading = title or f"المصدر {number}"

    return (
        f'<article class="evidence-card" dir="{_qdir(query_lang)}" lang="{query_lang}">'
        '<div class="evidence-card-head">'
        f'<div><span class="evidence-index">{number}</span>{badge}</div>'
        f'<div class="evidence-source" dir="auto">{source_line}</div>'
        '</div>'
        f'<h3>{heading}</h3>'
        f'{body}'
        '</article>'
    )


def _render_evidence(result: dict, count: int, query_lang: str, translations: dict[str, str]) -> str:
    candidates = list(result.get("candidates", []) or [])[: max(1, min(int(count), 5))]
    selected = result.get("selected_visible_evidence", {}) or {}
    selected_id = str(selected.get("record_id", ""))
    answered = result.get("decision") == "answer"
    cards = [_candidate_card(c, i + 1, selected_id, answered, query_lang, translations) for i, c in enumerate(candidates)]
    if not cards:
        return ""
    return (
        f'<section class="evidence-section" dir="{_qdir(query_lang)}" lang="{query_lang}">'
        f'{_qt(query_lang, "المصادر", "Sources", "h2", "section-title")}'
        '<div class="evidence-list">' + "".join(cards) + '</div></section>'
    )


def _answer_html(result: dict, query_lang: str, translations: dict[str, str]) -> str:
    selected = result.get("selected_visible_evidence", {}) or {}
    passage_ar = str(selected.get("passage_ar", "")).strip()
    rid = str(selected.get("record_id", ""))
    book = str(selected.get("book_ar", "")).strip()
    author = str(selected.get("author_ar", "")).strip()
    page = str(selected.get("page_number", "")).strip()
    answered = result.get("decision") == "answer"

    if not answered:
        return (
            f'<section class="result-card result-abstain" dir="{_qdir(query_lang)}" lang="{query_lang}">'
            f'{_qt(query_lang, "لم أجد ثقة كافية للإجابة", "I do not have enough confidence to answer", "h2")}'
            f'{_qt(query_lang, "راجع المصادر أدناه أو أعد صياغة السؤال.", "Review the sources below or rephrase the question.", "p")}'
            '</section>'
        )

    if not passage_ar:
        raise RuntimeError("Answer was allowed but selected visible passage is empty.")
    if str(result.get("answer_support_record_id", "")) != rid:
        raise RuntimeError("Visible-evidence grounding invariant failed.")

    if query_lang == "en":
        answer_text = translations.get(rid, "").strip()
        if not answer_text:
            raise RuntimeError("English answer requested but selected visible evidence was not translated.")
        answer_body = f'<div class="answer-body" dir="ltr" lang="en">{_esc(answer_text)}</div>'
        original = (
            '<details class="source-original answer-original">'
            '<summary>Original Arabic</summary>'
            f'<div class="answer-body source-arabic" dir="rtl" lang="ar">{_esc(passage_ar)}</div>'
            '</details>'
        )
    else:
        answer_body = f'<div class="answer-body source-arabic" dir="rtl" lang="ar">{_esc(passage_ar)}</div>'
        original = ""

    source_parts = [x for x in (book, author) if x]
    source_line = " · ".join(source_parts)
    if page:
        source_line += (" · " if source_line else "") + (("ص" if query_lang == "ar" else "p.") + f" {page}")

    return (
        f'<section class="result-card result-answer" dir="{_qdir(query_lang)}" lang="{query_lang}">'
        f'{_qt(query_lang, "الجواب", "Answer", "h2")}'
        f'{answer_body}'
        f'<div class="answer-source" dir="auto">{_esc(source_line) if source_line else ""}</div>'
        f'{original}'
        '</section>'
    )


def _diagnostics(result: dict, elapsed: float) -> dict:
    selected = result.get("selected_visible_evidence", {}) or {}
    corpus_rows = int(len(getattr(RUNTIME, "corpus", [])))
    return {
        "version": APP_VERSION,
        "decision": result.get("decision"),
        "probability_top1_correct": round(float(result.get("probability_top1_correct", 0.0)), 6),
        "frozen_threshold": EXPECTED_THRESHOLD,
        "selected_visible_record_id": str(selected.get("record_id", "")),
        "answer_support_record_id": result.get("answer_support_record_id"),
        "hidden_answer_pool_used": bool(result.get("hidden_answer_pool_used", False)),
        "full_corpus_retrieval": corpus_rows == EXPECTED_CORPUS_ROWS,
        "retriever_scored_rows": corpus_rows,
        "reranker_candidates": len(result.get("candidates", []) or []),
        "query_time_seconds": round(elapsed, 3),
    }


def _cache_get(key: str):
    with _CACHE_LOCK:
        value = _QUERY_CACHE.get(key)
        return value.copy() if isinstance(value, dict) else None


def _cache_put(key: str, value: dict):
    with _CACHE_LOCK:
        if key in _QUERY_CACHE:
            _QUERY_CACHE.pop(key, None)
        _QUERY_CACHE[key] = value
        while len(_QUERY_CACHE) > _CACHE_MAX:
            first = next(iter(_QUERY_CACHE))
            _QUERY_CACHE.pop(first, None)


DISPLAY_EVIDENCE_COUNT = 3


@spaces.GPU(duration=60)
def answer_question(query: str):
    q = _clean_query(query)
    query_lang = _detect_query_language(q)
    query_dir = _qdir(query_lang)
    if not q:
        return (
            f'<section class="result-card result-abstain" dir="{query_dir}" lang="{query_lang}">'
            + _qt(query_lang, "اكتب سؤالًا أولًا", "Enter a question first", "h2")
            + '</section>',
            "",
            {},
        )

    corpus = getattr(RUNTIME, "corpus", None)
    embeddings = getattr(RUNTIME, "passage_embeddings", None)
    if corpus is None or embeddings is None:
        raise RuntimeError("Frozen runtime corpus/index is not loaded.")
    if len(corpus) != EXPECTED_CORPUS_ROWS or int(embeddings.shape[0]) != EXPECTED_CORPUS_ROWS:
        raise RuntimeError("Full production corpus invariant failed.")

    result = _cache_get(q)
    started = time.perf_counter()
    if result is None:
        with _RUNTIME_LOCK:
            result = RUNTIME.rank(q, top_k=20)
        _cache_put(q, result)
    elapsed = time.perf_counter() - started

    if result.get("hidden_answer_pool_used") is not False:
        raise RuntimeError("Hidden answer pool detected; refusing to render.")
    if result.get("decision") == "answer":
        selected = result.get("selected_visible_evidence", {}) or {}
        if str(result.get("answer_support_record_id", "")) != str(selected.get("record_id", "")):
            raise RuntimeError("Answer support is not the selected visible evidence.")

    translations: dict[str, str] = {}
    translation_error = ""
    if query_lang == "en":
        try:
            translations = _prepare_english_translations(result, DISPLAY_EVIDENCE_COUNT)
        except Exception as exc:
            translation_error = f"{type(exc).__name__}: {exc}"
            if result.get("decision") == "answer":
                return (
                    '<section class="result-card result-abstain" dir="ltr" lang="en"><h2>English rendering is temporarily unavailable</h2><p>Please try again shortly.</p></section>',
                    "",
                    {"version": APP_VERSION, "query_language": "en", "translation_error": translation_error, "retrieval_executed": True},
                )

    diag = _diagnostics(result, elapsed)
    diag["query_language"] = query_lang
    diag["answer_language"] = query_lang
    diag["query_direction"] = query_dir
    diag["translation_layer_used"] = query_lang == "en"
    if translation_error:
        diag["translation_error"] = translation_error

    return (
        _answer_html(result, query_lang, translations),
        _render_evidence(result, DISPLAY_EVIDENCE_COUNT, query_lang, translations),
        diag,
    )


def clear_ui():
    return "", "", "", {}


CSS = r'''
:root {
  --bg:#f7f9f8; --surface:#ffffff; --surface-2:#f2f6f4; --text:#17211d;
  --muted:#68766f; --line:#dce5e0; --brand:#0f8a68; --brand-hover:#0b7559;
  --danger:#8f3b45; --shadow:0 14px 40px rgba(20,42,34,.07);
}
html[data-huda-theme="dark"] {
  --bg:#0c1411; --surface:#111d18; --surface-2:#16251f; --text:#f3f7f5;
  --muted:#a5b3ad; --line:#293c34; --brand:#58cbaa; --brand-hover:#70d7ba;
  --danger:#ff9ca8; --shadow:0 18px 50px rgba(0,0,0,.22); color-scheme:dark;
}
html[data-huda-theme="light"] { color-scheme:light; }
body, .gradio-container { background:var(--bg)!important; color:var(--text)!important; font-family:Calibri,Aptos,"Segoe UI",Tahoma,Arial,sans-serif!important; }
.gradio-container { max-width:none!important; padding:0!important; }
#huda-shell { max-width:1040px!important; margin:0 auto!important; padding:28px 22px 48px!important; }
#frontend_shell, #answer_output, #evidence_output { margin:0!important; padding:0!important; border:0!important; background:transparent!important; }
.huda-bridge, #diagnostics_json { display:none!important; }
footer { display:none!important; }
.huda-app { color:var(--text); }
.huda-app *, #answer_output *, #evidence_output * { box-sizing:border-box; }
.huda-app button, .huda-app textarea { font:inherit; }
.huda-topbar { display:flex; align-items:center; justify-content:space-between; gap:16px; margin-bottom:54px; direction:inherit; }
.huda-brand { font-size:23px; font-weight:800; letter-spacing:-.25px; color:var(--text); }
.huda-controls { display:flex; gap:8px; }
.huda-control { min-width:78px; height:40px; padding:0 14px; border:1px solid var(--line); border-radius:10px; background:var(--surface); color:var(--text); cursor:pointer; font-weight:700; }
.huda-control:hover { border-color:var(--brand); }
.huda-main { max-width:860px; margin:0 auto; }
.huda-intro { margin-bottom:26px; }
.huda-intro h1 { margin:0; font-size:38px; line-height:1.22; letter-spacing:-.5px; color:var(--text)!important; }
.huda-intro p { margin:10px 0 0; color:var(--muted)!important; font-size:16px; line-height:1.7; }
.huda-question-wrap { background:var(--surface); border:1px solid var(--line); border-radius:18px; padding:18px; box-shadow:var(--shadow); }
.huda-question { width:100%; min-height:190px; resize:vertical; border:1px solid var(--line); border-radius:14px; background:var(--surface-2); color:var(--text)!important; padding:18px; font-size:18px; line-height:1.75; outline:none; caret-color:var(--brand); }
.huda-question::placeholder { color:var(--muted)!important; opacity:.8; }
.huda-question:focus { border-color:var(--brand); box-shadow:0 0 0 3px color-mix(in srgb,var(--brand) 18%,transparent); }
.huda-actions { display:flex; gap:10px; margin-top:12px; }
.huda-submit { flex:1; min-height:50px; border:0; border-radius:12px; background:var(--brand); color:#fff!important; font-weight:800; cursor:pointer; }
html[data-huda-theme="dark"] .huda-submit { color:#082019!important; }
.huda-submit:hover { background:var(--brand-hover); }
.huda-submit:disabled { opacity:.65; cursor:wait; }
.huda-clear { width:116px; min-height:50px; border:1px solid var(--line); border-radius:12px; background:var(--surface); color:var(--text)!important; font-weight:700; cursor:pointer; }
.huda-clear:hover { border-color:var(--brand); }
.result-card, .evidence-section { max-width:860px; margin:24px auto 0; }
.result-card { background:var(--surface); border:1px solid var(--line); border-radius:18px; padding:24px; box-shadow:var(--shadow); }
.result-card h2, .evidence-section h2 { margin:0 0 14px; color:var(--text)!important; font-size:23px; }
.result-card p { margin:6px 0 0; color:var(--muted)!important; line-height:1.7; }
.result-abstain { border-inline-start:4px solid var(--danger); }
.result-answer { border-inline-start:4px solid var(--brand); }
.answer-body { color:var(--text)!important; font-size:18px; line-height:1.9; white-space:pre-wrap; }
.answer-source { margin-top:18px; padding-top:14px; border-top:1px solid var(--line); color:var(--muted)!important; font-size:14px; }
.section-title { margin-bottom:14px!important; }
.evidence-list { display:grid; gap:12px; }
.evidence-card { background:var(--surface); border:1px solid var(--line); border-radius:16px; padding:20px; color:var(--text); }
.evidence-card-head { display:flex; justify-content:space-between; align-items:flex-start; gap:14px; margin-bottom:10px; }
.evidence-card-head > div:first-child { display:flex; align-items:center; gap:8px; }
.evidence-index { display:inline-grid; place-items:center; width:28px; height:28px; border-radius:8px; background:var(--surface-2); color:var(--brand)!important; font-weight:800; }
.source-badge { color:var(--brand)!important; font-size:13px; font-weight:800; }
.evidence-source { color:var(--muted)!important; font-size:13px; text-align:end; }
.evidence-card h3 { margin:0 0 10px; color:var(--text)!important; font-size:18px; }
.evidence-passage { color:var(--text)!important; line-height:1.85; font-size:16px; white-space:pre-wrap; }
.source-original { margin-top:14px; border-top:1px solid var(--line); padding-top:10px; }
.source-original summary { color:var(--muted)!important; cursor:pointer; font-weight:700; }
.source-original[open] summary { margin-bottom:10px; }
.source-arabic { font-family:Calibri,Aptos,"Segoe UI",Tahoma,Arial,sans-serif!important; }
.i18n[hidden] { display:none!important; }
@media (max-width:720px) {
  #huda-shell { padding:20px 14px 36px!important; }
  .huda-topbar { margin-bottom:38px; }
  .huda-brand { font-size:21px; }
  .huda-control { min-width:64px; height:38px; padding:0 10px; }
  .huda-intro h1 { font-size:31px; }
  .huda-question-wrap { padding:14px; border-radius:16px; }
  .huda-question { min-height:170px; font-size:17px; }
  .huda-actions { flex-direction:column; }
  .huda-clear { width:100%; }
  .result-card { padding:20px; }
  .evidence-card-head { flex-direction:column; }
  .evidence-source { text-align:start; }
}
'''

JS = r'''
(() => {
  const root = document.documentElement;
  const KEY_LANG='hudanet-lang-v4', KEY_THEME='hudanet-theme-v4';
  const $=(s,c=document)=>c.querySelector(s);
  const $$=(s,c=document)=>Array.from(c.querySelectorAll(s));
  const store={
    get:k=>{try{return localStorage.getItem(k)}catch(e){return null}},
    set:(k,v)=>{try{localStorage.setItem(k,v)}catch(e){}}
  };
  let busy=false;
  let resultObserver=null;

  function bridgeInput(){
    const host=document.getElementById('question_bridge');
    return host?.querySelector('textarea,input')||null;
  }
  function bridgeButton(id){
    const host=document.getElementById(id);
    if(!host)return null;
    return host.tagName==='BUTTON'?host:(host.querySelector('button')||null);
  }
  function nativeSet(el,val){
    if(!el)return;
    const proto=el.tagName==='TEXTAREA'?HTMLTextAreaElement.prototype:HTMLInputElement.prototype;
    const setter=Object.getOwnPropertyDescriptor(proto,'value')?.set;
    if(setter)setter.call(el,String(val)); else el.value=String(val);
    el.dispatchEvent(new Event('input',{bubbles:true}));
    el.dispatchEvent(new Event('change',{bubbles:true}));
  }
  function language(){return root.dataset.hudaLang==='en'?'en':'ar'}
  function theme(){return root.dataset.hudaTheme==='dark'?'dark':'light'}
  function setText(el,val){if(el&&el.textContent!==val)el.textContent=val}

  function applyI18n(){
    const lang=language();
    $$('.i18n').forEach(el=>{
      const shouldHide=!el.classList.contains('i18n-'+lang);
      if(el.hidden!==shouldHide)el.hidden=shouldHide;
    });
    const app=$('#huda-app-root');
    if(app){app.dir=lang==='ar'?'rtl':'ltr';app.lang=lang}
    const q=$('#huda_question_input');
    if(q){
      q.placeholder=lang==='ar'?'اكتب سؤالك عن الحج أو العمرة…':'Type your Hajj or Umrah question…';
      q.setAttribute('aria-label',lang==='ar'?'سؤالك عن الحج أو العمرة':'Your Hajj or Umrah question');
    }
    setText($('#huda_lang_label'),lang==='ar'?'English':'العربية');
    setText($('#huda_theme_label'),theme()==='dark'?(lang==='ar'?'فاتح':'Light'):(lang==='ar'?'داكن':'Dark'));
    if(!busy)setText($('#huda_submit_label'),lang==='ar'?'إرسال':'Submit');
    setText($('#huda_clear_label'),lang==='ar'?'مسح':'Clear');
  }
  function applyLang(lang){
    root.dataset.hudaLang=lang==='en'?'en':'ar';
    store.set(KEY_LANG,root.dataset.hudaLang);
    applyI18n();
  }
  function applyTheme(t){
    root.dataset.hudaTheme=t==='dark'?'dark':'light';
    store.set(KEY_THEME,root.dataset.hudaTheme);
    applyI18n();
  }
  function sync(){
    const q=$('#huda_question_input');
    nativeSet(bridgeInput(),q?.value||'');
  }
  function setBusy(value){
    busy=!!value;
    const button=$('#huda_submit');
    if(button)button.disabled=busy;
    setText($('#huda_submit_label'),busy?(language()==='ar'?'جارٍ البحث…':'Searching…'):(language()==='ar'?'إرسال':'Submit'));
  }
  function submit(){
    if(busy)return;
    const q=$('#huda_question_input');
    if(!q)return;
    sync();
    const b=bridgeButton('send_bridge');
    if(!b){console.error('HUDA-Net bridge submit button not found');return}
    setBusy(true);
    b.click();
    window.setTimeout(()=>{if(busy)setBusy(false)},120000);
  }
  function clearAll(){
    const q=$('#huda_question_input');
    if(q){q.value='';q.focus()}
    sync();
    setBusy(false);
    bridgeButton('clear_bridge')?.click();
  }
  function observeResults(){
    if(resultObserver)return;
    const answer=document.getElementById('answer_output');
    if(!answer)return;
    resultObserver=new MutationObserver(()=>{if(busy)setBusy(false)});
    resultObserver.observe(answer,{childList:true,subtree:true,characterData:true});
  }
  function wireEvents(){
    if(document.documentElement.dataset.hudaV4Wired==='1')return;
    document.documentElement.dataset.hudaV4Wired='1';
    document.addEventListener('click',e=>{
      const t=e.target.closest('button');
      if(!t)return;
      if(t.id==='huda_lang_toggle'){e.preventDefault();applyLang(language()==='ar'?'en':'ar')}
      else if(t.id==='huda_theme_toggle'){e.preventDefault();applyTheme(theme()==='dark'?'light':'dark')}
      else if(t.id==='huda_submit'){e.preventDefault();submit()}
      else if(t.id==='huda_clear'){e.preventDefault();clearAll()}
    },true);
    document.addEventListener('keydown',e=>{
      if(e.target?.id==='huda_question_input'&&(e.ctrlKey||e.metaKey)&&e.key==='Enter'){e.preventDefault();submit()}
    },true);
  }
  function boot(attempt=0){
    const app=$('#huda-app-root');
    const q=$('#huda_question_input');
    if(!app||!q){
      if(attempt<120)window.setTimeout(()=>boot(attempt+1),100);
      return;
    }
    const savedLang=store.get(KEY_LANG);
    const savedTheme=store.get(KEY_THEME);
    root.dataset.hudaLang=savedLang==='en'?'en':'ar';
    root.dataset.hudaTheme=savedTheme==='light'||savedTheme==='dark'?savedTheme:(window.matchMedia&&window.matchMedia('(prefers-color-scheme: dark)').matches?'dark':'light');
    applyI18n();
    wireEvents();
    observeResults();
    window.setTimeout(sync,150);
  }
  boot();
})();
'''

FRONTEND_HTML = r'''
<div class="huda-app" id="huda-app-root" dir="rtl" lang="ar">
  <header class="huda-topbar">
    <div class="huda-brand">HUDA-Net</div>
    <div class="huda-controls">
      <button type="button" class="huda-control" id="huda_lang_toggle"><span id="huda_lang_label">English</span></button>
      <button type="button" class="huda-control" id="huda_theme_toggle"><span id="huda_theme_label">فاتح</span></button>
    </div>
  </header>
  <main class="huda-main">
    <section class="huda-intro">
      <h1 class="i18n i18n-ar">اسأل HUDA-Net</h1>
      <h1 class="i18n i18n-en" hidden>Ask HUDA-Net</h1>
      <p class="i18n i18n-ar">إرشاد الحج والعمرة المستند إلى الشواهد.</p>
      <p class="i18n i18n-en" hidden>Evidence-grounded Hajj and Umrah guidance.</p>
    </section>
    <section class="huda-question-wrap">
      <textarea id="huda_question_input" class="huda-question" rows="6" maxlength="1200" dir="auto" autocomplete="off" spellcheck="true" placeholder="اكتب سؤالك عن الحج أو العمرة…"></textarea>
      <div class="huda-actions">
        <button type="button" class="huda-submit" id="huda_submit"><span id="huda_submit_label">إرسال</span></button>
        <button type="button" class="huda-clear" id="huda_clear"><span id="huda_clear_label">مسح</span></button>
      </div>
    </section>
  </main>
</div>
'''

with gr.Blocks(title="HUDA-Net") as demo:
    with gr.Column(elem_id="huda-shell"):
        gr.HTML(FRONTEND_HTML, elem_id="frontend_shell")
        answer = gr.HTML("", elem_id="answer_output")
        evidence = gr.HTML("", elem_id="evidence_output")
        diagnostics = gr.JSON(value={}, elem_id="diagnostics_json", visible=False)
        with gr.Column(elem_classes=["huda-bridge"]):
            question_bridge = gr.Textbox(value="", elem_id="question_bridge", show_label=False)
            send_bridge = gr.Button("submit", elem_id="send_bridge")
            clear_bridge = gr.Button("clear", elem_id="clear_bridge")
    outputs = [answer, evidence, diagnostics]
    send_bridge.click(answer_question, [question_bridge], outputs, show_progress="minimal", concurrency_limit=1)
    clear_bridge.click(clear_ui, None, [question_bridge, answer, evidence, diagnostics], queue=False)
try:
    demo.queue(default_concurrency_limit=1, max_size=32)
except TypeError:
    demo.queue()
demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True, share=False, ssr_mode=False, css=CSS, js=JS)