app.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""EchoCache - a semantic cache and cost-observability layer for LLM APIs.

Design rules that this file never breaks:
  * No torch / transformers / sklearn / faiss. Only numpy, pandas, gradio,
    huggingface_hub and the standard library.
  * Similarity, normalization and safety screening are 100% local. The optional
    embedding API only *re-ranks*; if it is disabled, rate limited (429), out of
    credit (402) or slow, every core feature keeps working unchanged.
  * No public function may raise. Failures come back as
    {"ok": false, "error": {"code": ..., "message": ...}}.
  * Import time does no heavy work: no model download, no big allocation.
  * Memory is bounded by MAX_ENTRIES and MAX_TEXT_CHARS, with LRU eviction.
"""

from __future__ import annotations

import csv
import functools
import hashlib
import html as html_lib
import io
import json
import math
import os
import re
import tempfile
import threading
import time
import unicodedata
import zlib
from collections import Counter, OrderedDict, deque
from enum import Enum
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple

import numpy as np
import pandas as pd
import gradio as gr

APP_NAME = "EchoCache"
APP_VERSION = "1.0.0"
START_TIME = time.time()

# =============================================================================
# (A) Configuration
# =============================================================================


def _env_str(name: str, default: str = "") -> str:
    """Guarantees: returns a stripped string, never raises, never returns None."""
    try:
        raw = os.environ.get(name)
        return default if raw is None else str(raw).strip()
    except Exception:
        return default


def _env_int(name: str, default: int, lo: int = 0, hi: int = 10 ** 9) -> int:
    """Guarantees: returns an int clamped into [lo, hi]; bad values fall back to default."""
    try:
        val = int(float(_env_str(name, "").replace(",", "") or default))
    except Exception:
        val = default
    return max(lo, min(hi, val))


def _env_float(name: str, default: float, lo: float = -1e9, hi: float = 1e9) -> float:
    """Guarantees: returns a float clamped into [lo, hi]; bad values fall back to default."""
    try:
        val = float(_env_str(name, "") or default)
        if math.isnan(val) or math.isinf(val):
            val = default
    except Exception:
        val = default
    return max(lo, min(hi, val))


def _safe_int(value: Any, default: int) -> int:
    """Guarantees: an int for ANY input - NaN, inf, None, "abc" and objects all fall back to default."""
    try:
        if value is None or isinstance(value, bool):
            return default
        number = float(value)
        if math.isnan(number) or math.isinf(number):
            return default
        return int(number)
    except Exception:
        return default


def _safe_float(value: Any, default: float) -> float:
    """Guarantees: a finite float for ANY input; NaN, inf and junk fall back to default."""
    try:
        if value is None or isinstance(value, bool):
            return default
        number = float(value)
        if math.isnan(number) or math.isinf(number):
            return default
        return number
    except Exception:
        return default


def _env_bool(name: str, default: bool = False) -> bool:
    """Guarantees: returns a bool for any input ('1/true/yes/on' are true)."""
    raw = _env_str(name, "").lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "y", "on")


# --- required by the spec ----------------------------------------------------
HF_TOKEN = _env_str("HF_TOKEN") or _env_str("HUGGINGFACE_HUB_TOKEN") or _env_str("HUGGINGFACEHUB_API_TOKEN")
EMBED_MODEL = _env_str("EMBED_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
MAX_ENTRIES = _env_int("MAX_ENTRIES", 5000, lo=1, hi=200000)
MAX_TEXT_CHARS = _env_int("MAX_TEXT_CHARS", 8000, lo=64, hi=200000)
DEFAULT_THRESHOLD = _env_float("DEFAULT_THRESHOLD", 0.92, lo=0.0, hi=1.0)
BORDERLINE_MARGIN = _env_float("BORDERLINE_MARGIN", 0.03, lo=0.0, hi=0.5)
MAX_TTL_SEC = 365 * 24 * 3600     # hard ceiling for any TTL, including imported ones
DEFAULT_TTL_SEC = _env_int("DEFAULT_TTL_SEC", 86400, lo=1, hi=MAX_TTL_SEC)
PRICE_IN_PER_1K = _env_float("PRICE_IN_PER_1K", 0.0, lo=0.0, hi=1000.0)
PRICE_OUT_PER_1K = _env_float("PRICE_OUT_PER_1K", 0.0, lo=0.0, hi=1000.0)

# --- tunables (all optional, all have safe defaults) -------------------------
VECTOR_DIM = _env_int("VECTOR_DIM", 4096, lo=256, hi=65536)
NGRAM_N = _env_int("NGRAM_N", 3, lo=2, hi=6)
SIMHASH_BITS = 64
SIMHASH_MAX_DIST = _env_int("SIMHASH_MAX_DIST", 12, lo=0, hi=64)
# Recall guard: when the SimHash band returns nothing and the partition is small,
# fall back to a full scan. Cheap for small indexes, never used for big ones.
FULL_SCAN_LIMIT = _env_int("FULL_SCAN_LIMIT", 512, lo=0, hi=20000)
CANDIDATE_LIMIT = _env_int("CANDIDATE_LIMIT", 64, lo=1, hi=4096)
GUARD_MAX_CHECK = _env_int("GUARD_MAX_CHECK", 5, lo=1, hi=64)
EMBED_WEIGHT = _env_float("EMBED_WEIGHT", 0.35, lo=0.0, hi=1.0)  # local-leaning by default
EMBED_TIMEOUT = _env_float("EMBED_TIMEOUT", 6.0, lo=0.5, hi=60.0)
EMBED_TOPK = _env_int("EMBED_TOPK", 5, lo=1, hi=32)
EMBED_CACHE_SIZE = _env_int("EMBED_CACHE_SIZE", 2048, lo=16, hi=100000)
EMBED_FAIL_THRESHOLD = _env_int("EMBED_FAIL_THRESHOLD", 3, lo=1, hi=100)
EMBED_COOLDOWN_SEC = _env_int("EMBED_COOLDOWN_SEC", 600, lo=10, hi=86400)
EVENT_BUFFER = _env_int("EVENT_BUFFER", 5000, lo=100, hi=200000)
LOG_PROMPT_PREFIX = _env_int("LOG_PROMPT_PREFIX", 0, lo=0, hi=200)  # 0 = never log prompt text
SWEEP_MAX_ROWS = _env_int("SWEEP_MAX_ROWS", 2000, lo=10, hi=100000)
SWEEP_TIME_BUDGET_SEC = _env_float("SWEEP_TIME_BUDGET_SEC", 90.0, lo=5.0, hi=900.0)
EXPIRE_SWEEP_INTERVAL = _env_float("EXPIRE_SWEEP_INTERVAL", 30.0, lo=1.0, hi=3600.0)
ENTROPY_THRESHOLD = _env_float("ENTROPY_THRESHOLD", 4.0, lo=0.0, hi=8.0)
ENTROPY_MIN_LEN = _env_int("ENTROPY_MIN_LEN", 24, lo=8, hi=512)
MAX_EMAILS = _env_int("MAX_EMAILS", 1, lo=0, hi=100)   # reject when count > MAX_EMAILS
MAX_PHONES = _env_int("MAX_PHONES", 1, lo=0, hi=100)   # reject when count > MAX_PHONES
MAX_TENANTS = _env_int("MAX_TENANTS", 256, lo=1, hi=10000)
# MAX_ENTRIES is PER TENANT; this is the hard ceiling across every partition.
# Memory is dominated by the vector matrix: entries x VECTOR_DIM x 4 bytes
# (20000 x 4096 x 4B = 327 MB), which is what keeps a 16 GB free Space safe.
MAX_TOTAL_ENTRIES = _env_int("MAX_TOTAL_ENTRIES", 20000, lo=1, hi=2000000)
IMPORT_MAX_BYTES = _env_int("IMPORT_MAX_BYTES", 64 * 1024 * 1024, lo=1024, hi=512 * 1024 * 1024)
# Comma separated guard checks to disable, e.g. GUARD_DISABLE="proper_noun,question_type"
GUARD_DISABLE = {x.strip() for x in _env_str("GUARD_DISABLE", "").split(",") if x.strip()}

EMBED_ENABLED = bool(HF_TOKEN)
MODE = "local+embedding" if EMBED_ENABLED else "local-only"

_TMP_DIR = os.path.join(tempfile.gettempdir(), "echocache")
try:
    os.makedirs(_TMP_DIR, exist_ok=True)
except Exception:
    _TMP_DIR = tempfile.gettempdir()


def config_snapshot() -> Dict[str, Any]:
    """Guarantees: returns a JSON-serializable copy of the effective configuration, never secrets."""
    return {
        "app": APP_NAME,
        "version": APP_VERSION,
        "mode": MODE,
        "hf_token_present": bool(HF_TOKEN),
        "embed_model": EMBED_MODEL if EMBED_ENABLED else None,
        "embed_weight": EMBED_WEIGHT if EMBED_ENABLED else 0.0,
        "max_entries": MAX_ENTRIES,
        "max_text_chars": MAX_TEXT_CHARS,
        "default_threshold": DEFAULT_THRESHOLD,
        "borderline_margin": BORDERLINE_MARGIN,
        "default_ttl_sec": DEFAULT_TTL_SEC,
        "price_in_per_1k": PRICE_IN_PER_1K,
        "price_out_per_1k": PRICE_OUT_PER_1K,
        "vector_dim": VECTOR_DIM,
        "ngram_n": NGRAM_N,
        "simhash_max_dist": SIMHASH_MAX_DIST,
        "candidate_limit": CANDIDATE_LIMIT,
        "guard_disabled_checks": sorted(GUARD_DISABLE),
        "log_prompt_prefix": LOG_PROMPT_PREFIX,
        "max_tenants": MAX_TENANTS,
        "max_total_entries": MAX_TOTAL_ENTRIES,
        "gradio_version": getattr(gr, "__version__", "unknown"),
    }


# =============================================================================
# Result envelopes + the "nothing ever raises" decorator
# =============================================================================


class ErrorCode(str, Enum):
    """Stable machine-readable error codes shared by the UI and the HTTP API."""

    INTERNAL = "internal_error"
    BAD_INPUT = "bad_input"
    EMPTY_PROMPT = "empty_prompt"
    EMPTY_RESPONSE = "empty_response"
    EMPTY_PREFIX = "empty_prefix"
    TENANT_LIMIT = "tenant_limit_reached"
    NOT_FOUND = "not_found"
    PAYLOAD_TOO_LARGE = "payload_too_large"
    BAD_PAYLOAD = "bad_payload"
    NO_FILE = "no_file"


def ok(**payload: Any) -> Dict[str, Any]:
    """Guarantees: returns a dict whose "ok" key is always True."""
    out: Dict[str, Any] = {"ok": True}
    out.update(payload)
    return out


def err(code: Any, message: str, **extra: Any) -> Dict[str, Any]:
    """Guarantees: returns {"ok": False, "error": {...}} with a string code, always JSON-safe."""
    code_str = code.value if isinstance(code, Enum) else str(code)
    body: Dict[str, Any] = {"code": code_str, "message": str(message)[:800]}
    for key, value in extra.items():
        try:
            json.dumps(value)
            body[key] = value
        except Exception:
            body[key] = repr(value)[:200]
    return {"ok": False, "error": body}


def guarded(fn: Callable[..., Any]) -> Callable[..., Any]:
    """Guarantees: the wrapped callable never raises; any exception becomes an error envelope."""

    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        """Guarantees: calls the wrapped function and converts any exception into an error envelope."""
        try:
            return fn(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 - deliberate catch-all, see module docstring
            return err(
                ErrorCode.INTERNAL,
                "{0}: {1}".format(type(exc).__name__, exc),
                where=getattr(fn, "__name__", "unknown"),
            )

    return wrapper


def _jsonable(obj: Any) -> Any:
    """Guarantees: returns a structure json.dumps can serialize, degrading to repr() strings."""
    try:
        json.dumps(obj)
        return obj
    except Exception:
        pass
    if isinstance(obj, dict):
        return {str(k): _jsonable(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple, set)):
        return [_jsonable(v) for v in obj]
    if isinstance(obj, (np.integer,)):
        return int(obj)
    if isinstance(obj, (np.floating,)):
        return float(obj)
    if isinstance(obj, np.ndarray):
        return obj.tolist()
    return repr(obj)[:500]


# =============================================================================
# (B) Normalizer
# =============================================================================

# CJK punctuation is NOT folded by NFKC, so it is mapped explicitly. Keeping the
# table tiny and explicit matters: over-aggressive folding creates false merges.
_PUNCT_MAP = {
    "。": ".",   # 。
    "、": ",",   # 、
    "，": ",",   # ，
    "．": ".",   # ．
    "・": " ",   # ・
    "「": '"', "」": '"',   # 「 」
    "『": '"', "』": '"',   # 『 』
    "｢": '"', "｣": '"',
    "‘": "'", "’": "'",
    "“": '"', "”": '"',
    "–": "-", "—": "-", "‐": "-", "‑": "-", "‒": "-",
    "―": "-", "ー": "ー",  # keep the katakana prolonged mark as-is
    "…": "...",
    " ": " ", "　": " ", "​": "", "": "",
}
_PUNCT_TABLE = {ord(k): v for k, v in _PUNCT_MAP.items()}

# Boilerplate removal. Each pattern is deliberately anchored so that it can only
# eat greetings / sign-offs / disclaimers - never the substance of a request.
_BOILERPLATE_PATTERNS: List[re.Pattern] = [
    # English greetings at the start of a line: "Hi team," / "Dear support:"
    re.compile(r"^[ \t]*(?:hi|hello|hey|dear|good\s+(?:morning|afternoon|evening))\b[^\n]{0,40}?[,:!.]\s*", re.I | re.M),
    # English sign-offs and everything that follows (signature blocks).
    # These MUST stay anchored to the start of a line: an unanchored
    # "cheers[\s\S]*$" would turn "Translate cheers into French" into
    # "translate", colliding with every other "Translate ... " prompt and
    # serving a wrong answer through the exact-match stage, which never
    # consults MismatchGuard.
    re.compile(r"^[ \t]*(?:best|kind|warm)\s+regards\b[\s\S]*$", re.I | re.M),
    re.compile(r"^[ \t]*(?:sincerely|cheers|thanks\s+again)\b[\s\S]*$", re.I | re.M),
    re.compile(r"^[ \t]*--[ \t]*$[\s\S]*", re.M),
    # Trailing politeness on its own line.
    re.compile(r"\b(?:thanks?|thank you)(?:\s+(?:so|very)\s+much)?(?:\s+in\s+advance)?[!.,]*[ \t]*$", re.I | re.M),
    re.compile(r"^[ \t]*(?:please\s+)?let\s+me\s+know\s+if[^\n]{0,80}$", re.I | re.M),
    # Leading polite-request frames. "Could you please tell me how do I reset my
    # password?" is the SAME request as "How do I reset my password?" - without
    # this the two score 0.73 and a legitimate reuse is lost.
    re.compile(r"^[ \t]*(?:could|can|would|will)\s+you\s+(?:kindly\s+|please\s+)?"
               r"(?:tell\s+me|let\s+me\s+know|explain|clarify|confirm|advise)\b[,:]?\s*", re.I | re.M),
    re.compile(r"^[ \t]*i(?:'m|\s+am)?\s*(?:was\s+)?wondering\b(?:\s+if|\s+whether)?[,:]?\s*", re.I | re.M),
    re.compile(r"^[ \t]*(?:just\s+a\s+)?quick\s+question[:,]?\s*", re.I | re.M),
    re.compile(r"^[ \t]*(?:please\s+)?let\s+me\s+know\s+(?=where|how|what|when|why|who|which)", re.I | re.M),
    re.compile(r"^[ \t]*(?:i\s+)?(?:have|had)\s+a\s+quick\s+question[:,.]?\s*", re.I | re.M),
    re.compile(r"お手数(?:ですが|ですけれども|をおかけしますが|をおかけしますけれども|ながら)[、,]?"),
    re.compile(r"(?:大変)?恐縮ですが[、,]?"),
    re.compile(r"(?:大変)?(?:申し訳(?:ありません|ございません)|すみません)が[、,]?"),
    # Confidentiality / AI disclaimers.
    re.compile(r"this\s+(?:e-?mail|message)\s+(?:and\s+any\s+attachments\s+)?(?:is|are)\s+confidential[\s\S]*$", re.I),
    re.compile(r"\bas\s+an\s+ai\s+(?:language\s+)?model\b[^.\n]*[.\n]?", re.I),
    re.compile(r"\bi'?m\s+sorry,?\s+but\s+as\s+an\s+ai\b[^.\n]*[.\n]?", re.I),
    # Japanese greetings / closings / hedges.
    re.compile(r"(?:いつも)?(?:大変)?お世話になって(?:おります|います)[。\.、,]?"),
    re.compile(r"(?:何卒|どうぞ)?よろしくお願い(?:いた|致)?します[。\.！!]?"),
    re.compile(r"よろしくお願いいたします[。\.！!]?"),
    re.compile(r"恐れ入りま(?:すが|す)[、,]?"),
    re.compile(r"お忙しいところ(?:恐縮ですが|失礼(?:いた)?します)[、,]?"),
    re.compile(r"ご確認のほど(?:よろしくお願いします)?[。\.]?"),
    re.compile(r"^[ \t]*(?:こんにちは|こんばんは|おはようございます|はじめまして)[、,。\.！!]?", re.M),
    re.compile(r"(?:どうも)?ありがとうございま(?:す|した)[。\.！!]?"),
    re.compile(r"^[ \t]*以上[、,。\.]?[ \t]*$", re.M),
    # A line that is nothing but an addressee ("山田様" / "サポート担当者様").
    re.compile(r"^[ \t]*[^\n]{1,16}(?:様|御中)[、,]?[ \t]*$", re.M),
]

_WS_RE = re.compile(r"[ \t　]+")
_MULTINL_RE = re.compile(r"\n{2,}")
_REPEAT_PUNCT_RE = re.compile(r"([?!.,])\1+")


def _visible_len(text: str) -> int:
    """Guarantees: the number of non-whitespace characters, the unit the strip budget is measured in."""
    return len("".join(text.split()))


def strip_boilerplate(text: str) -> str:
    """Guarantees: removes only greeting/sign-off/disclaimer spans, never more than half the content per rule."""
    out = text
    for pattern in _BOILERPLATE_PATTERNS:
        try:
            candidate = pattern.sub(" ", out)
        except Exception:
            continue
        if candidate == out:        # no match: skip the (relatively costly) budget check
            continue
        before = _visible_len(out)
        after = _visible_len(candidate)
        # Fail-safe: a single boilerplate rule that swallows more than half of
        # what is left is almost certainly matching content, not a sign-off.
        # Skipping it can only cost a cache miss; applying it could collapse
        # two different prompts onto one key and serve a wrong answer.
        if before > 0 and after * 2 < before:
            continue
        out = candidate
    return out


def normalize(text: Any) -> str:
    """Guarantees: deterministic, offline, idempotent canonical form; equal meaning-surface -> equal string."""
    if text is None:
        return ""
    try:
        raw = text if isinstance(text, str) else str(text)
    except Exception:
        return ""
    if not raw:
        return ""
    raw = raw[: MAX_TEXT_CHARS * 2]
    try:
        out = unicodedata.normalize("NFKC", raw)
    except Exception:
        out = raw
    out = out.replace("\r\n", "\n").replace("\r", "\n")
    out = strip_boilerplate(out)
    out = out.translate(_PUNCT_TABLE)
    out = out.lower()
    # Collapse whitespace: per-line trim first, then fold blank lines away.
    lines = [_WS_RE.sub(" ", line).strip() for line in out.split("\n")]
    out = "\n".join(line for line in lines if line)
    out = _MULTINL_RE.sub("\n", out)
    out = _WS_RE.sub(" ", out)
    # Fold repeated and trailing sentence punctuation so "...ください。", "...ください"
    # and "password??" collapse onto one key. A single "?" is deliberately kept:
    # it distinguishes a question from a statement.
    out = _REPEAT_PUNCT_RE.sub(r"\1", out)
    out = out.strip().rstrip(" .!,;:")
    return out.strip()[:MAX_TEXT_CHARS]


def light_normalize(text: Any) -> str:
    """Guarantees: NFKC + whitespace folding only - case, digits and negations are preserved for MismatchGuard."""
    if text is None:
        return ""
    try:
        raw = text if isinstance(text, str) else str(text)
    except Exception:
        return ""
    try:
        out = unicodedata.normalize("NFKC", raw)
    except Exception:
        out = raw
    out = out.replace("\r\n", "\n").replace("\r", "\n")
    out = out.translate({ord("　"): " ", ord(" "): " ", ord("​"): None})
    out = _WS_RE.sub(" ", out)
    return out.strip()[:MAX_TEXT_CHARS]


def exact_key(text: Any) -> str:
    """Guarantees: the sha256 of normalize(text) - the canonical exact-match key of stage 1."""
    return hashlib.sha256(normalize(text).encode("utf-8", "replace")).hexdigest()


# =============================================================================
# (C) Vectorizer - numpy only, no sklearn, no tokenizer, no word boundaries
# =============================================================================

_POPCOUNT8 = np.array([bin(i).count("1") for i in range(256)], dtype=np.uint8)
_MASK64 = 0xFFFFFFFFFFFFFFFF
_M1 = 0x5555555555555555
_M2 = 0x3333333333333333
_M4 = 0x0F0F0F0F0F0F0F0F
_H01 = 0x0101010101010101


def _ngram_counts(text: str, n: int = NGRAM_N) -> Counter:
    """Guarantees: character n-gram multiset that works for CJK and any unspaced script."""
    if not text:
        return Counter()
    padded = " " + text + " "
    n = max(1, min(8, int(n)))
    if len(padded) <= n:
        return Counter([padded])
    return Counter(padded[i : i + n] for i in range(len(padded) - n + 1))


def char_ngram_hash(text: str, n: int = 3, dim: int = 4096) -> np.ndarray:
    """Guarantees: a deterministic L2-normalized float32 vector of length dim, tokenizer-free and offline."""
    dim = max(16, int(dim))
    vec = np.zeros(dim, dtype=np.float32)
    counts = _ngram_counts(text or "", n)
    if not counts:
        return vec
    idxs = np.empty(len(counts), dtype=np.int64)
    vals = np.empty(len(counts), dtype=np.float32)
    for i, (gram, count) in enumerate(counts.items()):
        # crc32 is stable across processes and Python versions (unlike hash()),
        # which is what makes export/import of an index reproducible.
        h = zlib.crc32(gram.encode("utf-8", "replace")) & 0xFFFFFFFF
        idxs[i] = h % dim
        sign = 1.0 if (h >> 17) & 1 else -1.0
        # sub-linear term weighting damps the effect of a repeated n-gram
        vals[i] = sign * (1.0 + math.log(count))
    np.add.at(vec, idxs, vals)
    norm = float(np.linalg.norm(vec))
    if norm > 0.0:
        vec /= np.float32(norm)
    return vec


def simhash(text: str, bits: int = 64) -> int:
    """Guarantees: a stable b-bit fingerprint where near-duplicate texts differ in few bits."""
    bits = 64 if bits not in (32, 64, 128) else int(bits)
    nbytes = bits // 8
    counts = _ngram_counts(text or "", NGRAM_N)
    if not counts:
        return 0
    digests = bytearray()
    weights = np.empty(len(counts), dtype=np.int32)
    for i, (gram, count) in enumerate(counts.items()):
        digests += hashlib.blake2b(gram.encode("utf-8", "replace"), digest_size=nbytes).digest()
        weights[i] = count
    arr = np.frombuffer(bytes(digests), dtype=np.uint8).reshape(-1, nbytes)
    bit_matrix = np.unpackbits(arr, axis=1).astype(np.int32)  # (m, bits), MSB first
    acc = ((bit_matrix * 2 - 1) * weights[:, None]).sum(axis=0)
    packed = np.packbits((acc > 0).astype(np.uint8))
    return int.from_bytes(packed.tobytes(), "big")


def hamming(a: int, b: int) -> int:
    """Guarantees: exact Hamming distance of two 64-bit fingerprints using pure bit arithmetic."""
    x = (int(a) ^ int(b)) & _MASK64
    x = x - ((x >> 1) & _M1)
    x = (x & _M2) + ((x >> 2) & _M2)
    x = (x + (x >> 4)) & _M4
    return int(((x * _H01) & _MASK64) >> 56)


def hamming_array(fingerprints: np.ndarray, query: int) -> np.ndarray:
    """Guarantees: vectorized Hamming distances (uint8 popcount table) for a whole partition at once."""
    if fingerprints.size == 0:
        return np.zeros(0, dtype=np.int32)
    xor = np.bitwise_xor(np.ascontiguousarray(fingerprints), np.uint64(int(query) & _MASK64))
    return _POPCOUNT8[xor.view(np.uint8).reshape(-1, 8)].sum(axis=1).astype(np.int32)


def cosine(a: np.ndarray, b: np.ndarray) -> float:
    """Guarantees: cosine similarity in [-1, 1]; returns 0.0 for degenerate vectors instead of NaN."""
    try:
        na = float(np.linalg.norm(a))
        nb = float(np.linalg.norm(b))
        if na <= 0.0 or nb <= 0.0:
            return 0.0
        return float(np.clip(float(np.dot(a, b)) / (na * nb), -1.0, 1.0))
    except Exception:
        return 0.0


def featurize(text: str, dim: int = None, n: int = None) -> Dict[str, Any]:
    """Guarantees: every local feature (normalized text, key, vector, fingerprint) computed in one offline pass."""
    norm = normalize(text)
    return {
        "norm": norm,
        "key": hashlib.sha256(norm.encode("utf-8", "replace")).hexdigest(),
        "vector": char_ngram_hash(norm, NGRAM_N if n is None else n, VECTOR_DIM if dim is None else dim),
        "simhash": simhash(norm, SIMHASH_BITS),
    }


# =============================================================================
# (E) SafetyFilter - never cache a secret
# =============================================================================


class ReasonCode(str, Enum):
    """Stable reason codes returned identically by the UI and by the JSON API."""

    OK = "ok"
    EMPTY_PROMPT = "empty_prompt"
    EMPTY_RESPONSE = "empty_response"
    CREDIT_CARD = "credit_card_luhn"
    CREDENTIAL_PREFIX = "credential_prefix"
    JWT = "jwt_structure"
    HIGH_ENTROPY = "high_entropy_secret"
    CONTACT_PII = "contact_pii_excess"
    PRIVATE_KEY = "private_key_block"


REASON_HELP: Dict[str, str] = {
    ReasonCode.OK.value: "Safe to cache.",
    ReasonCode.EMPTY_PROMPT.value: "Prompt was empty after normalization.",
    ReasonCode.EMPTY_RESPONSE.value: "Response was empty.",
    ReasonCode.CREDIT_CARD.value: "A digit run passed the Luhn checksum (probable card number).",
    ReasonCode.CREDENTIAL_PREFIX.value: "A known credential prefix was found (sk-, ghp_, AKIA, xoxb-, AIza, hf_...).",
    ReasonCode.JWT.value: "A three-part JWT structure was found.",
    ReasonCode.HIGH_ENTROPY.value: "A long alphanumeric run exceeded the Shannon-entropy threshold.",
    ReasonCode.CONTACT_PII.value: "Too many e-mail addresses or phone numbers.",
    ReasonCode.PRIVATE_KEY.value: "A PEM private-key block was found.",
}

_CARD_CANDIDATE_RE = re.compile(r"(?<![0-9])(?:[0-9][ \-]?){12,18}[0-9](?![0-9])")
_CREDENTIAL_RES = [
    re.compile(r"\bsk-[A-Za-z0-9_\-]{16,}"),
    re.compile(r"\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{20,}"),
    re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}"),
    re.compile(r"\bAKIA[0-9A-Z]{12,}"),
    re.compile(r"\bASIA[0-9A-Z]{12,}"),
    re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{10,}"),
    re.compile(r"\bAIza[0-9A-Za-z_\-]{30,}"),
    re.compile(r"\bya29\.[A-Za-z0-9_\-]{20,}"),
    re.compile(r"\bhf_[A-Za-z0-9]{30,}"),
    re.compile(r"\bglpat-[A-Za-z0-9_\-]{18,}"),
    re.compile(r"\bSG\.[A-Za-z0-9_\-]{20,}\.[A-Za-z0-9_\-]{20,}"),
    re.compile(r"\bAC[0-9a-fA-F]{32}\b"),  # Twilio account sid
]
_PRIVATE_KEY_RE = re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")
_JWT_RE = re.compile(r"\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}")
_ENTROPY_TOKEN_RE = re.compile(r"[A-Za-z0-9+/=_\-]{%d,}" % ENTROPY_MIN_LEN)
_EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b")
_PHONE_RE = re.compile(r"(?<![0-9])(?:\+?\d{1,3}[ \-.]?)?(?:\(\d{2,4}\)[ \-.]?|\d{2,4}[ \-.])\d{2,4}[ \-.]\d{3,4}(?![0-9])")
_HEXY_RE = re.compile(r"^[0-9a-fA-F]+$")


def luhn_ok(digits: str) -> bool:
    """Guarantees: True only for a 13-19 digit string whose Luhn checksum is valid."""
    digits = "".join(ch for ch in digits if ch.isdigit())
    if not (13 <= len(digits) <= 19):
        return False
    total = 0
    parity = len(digits) % 2
    for i, ch in enumerate(digits):
        d = ord(ch) - 48
        if i % 2 == parity:
            d *= 2
            if d > 9:
                d -= 9
        total += d
    return total % 10 == 0


def shannon_entropy(text: str) -> float:
    """Guarantees: Shannon entropy in bits/char computed locally, 0.0 for empty input."""
    if not text:
        return 0.0
    counts = Counter(text)
    n = float(len(text))
    return float(-sum((c / n) * math.log2(c / n) for c in counts.values()))


_SEGMENT_SPLIT_RE = re.compile(r"[-_/=.+:@]")
_CAMEL_SPLIT_RE = re.compile(r"[A-Z]+(?![a-z])|[A-Z][a-z]+|[a-z]+|\d+")


def _looks_like_identifier(token: str) -> bool:
    """Guarantees: True for URL slugs, snake_case names and CamelCase identifiers - never for a random key.

    Shannon entropy alone cannot separate the two populations: measured over 340
    random 24-48 char keys the minimum was 3.97 bits/char, while ordinary
    identifiers such as "com/blog/how-to-reset-your-password-2024" (4.16) and
    "UserAuthenticationServiceImpl2024Handler" (4.36) sit in the same band.
    Structure is what separates them - a real key has no word-shaped segments.
    With this filter the screen catches 97.4% of random keys and stops refusing
    ordinary URLs and filenames.
    """
    segments = [s for s in _SEGMENT_SPLIT_RE.split(token) if s]
    if len(segments) >= 2:
        words = sum(1 for s in segments if s.isalpha() and len(s) >= 3)
        if words >= max(2, int(0.6 * len(segments))):
            return True
    parts = _CAMEL_SPLIT_RE.findall(token)
    wordy = [p for p in parts if p.isalpha() and len(p) >= 3]
    return len(wordy) >= 3 and sum(len(p) for p in wordy) >= 0.7 * len(token)


def _has_high_entropy_secret(text: str) -> bool:
    """Guarantees: True only for long mixed-class runs whose entropy exceeds the configured threshold."""
    for match in _ENTROPY_TOKEN_RE.finditer(text):
        token = match.group(0)
        if len(token) > 512:
            token = token[:512]
        has_digit = any(ch.isdigit() for ch in token)
        has_alpha = any(ch.isalpha() for ch in token)
        if not (has_digit and has_alpha):
            continue
        if _looks_like_identifier(token):
            continue
        entropy = shannon_entropy(token)
        # Long hex strings (hashes, request ids) cap out near 4.0 bits/char, so a
        # slightly lower bar is used for them; base64-ish secrets go well above 4.
        limit = ENTROPY_THRESHOLD if not _HEXY_RE.match(token) else min(ENTROPY_THRESHOLD, 3.75)
        if entropy >= limit and len(set(token)) >= 12:
            return True
    return False


def should_cache(prompt: str, response: str) -> Tuple[bool, str]:
    """Guarantees: never raises and returns (False, reason_code) whenever a secret-shaped pattern is present."""
    try:
        prompt = prompt or ""
        response = response or ""
        blob = prompt + "\n" + response
        if not normalize(prompt):
            return False, ReasonCode.EMPTY_PROMPT.value
        if not response.strip():
            return False, ReasonCode.EMPTY_RESPONSE.value
        if _PRIVATE_KEY_RE.search(blob):
            return False, ReasonCode.PRIVATE_KEY.value
        if _JWT_RE.search(blob):
            return False, ReasonCode.JWT.value
        for pattern in _CREDENTIAL_RES:
            if pattern.search(blob):
                return False, ReasonCode.CREDENTIAL_PREFIX.value
        for match in _CARD_CANDIDATE_RE.finditer(blob):
            if luhn_ok(match.group(0)):
                return False, ReasonCode.CREDIT_CARD.value
        if _has_high_entropy_secret(blob):
            return False, ReasonCode.HIGH_ENTROPY.value
        if len(_EMAIL_RE.findall(blob)) > MAX_EMAILS:
            return False, ReasonCode.CONTACT_PII.value
        if len(_PHONE_RE.findall(blob)) > MAX_PHONES:
            return False, ReasonCode.CONTACT_PII.value
        return True, ReasonCode.OK.value
    except Exception:
        # Fail closed: if the screen itself fails we refuse to cache.
        return False, ReasonCode.HIGH_ENTROPY.value


# =============================================================================
# (F) MismatchGuard - the quality core
# =============================================================================
#
# WHY THIS EXISTS
# ---------------
# A character n-gram cosine of 0.95+ means "these two strings look alike", which
# is NOT the same as "these two questions have the same answer". Serving a cached
# answer for a different question is not a cache miss, it is a *wrong answer* -
# a quality incident that is far more expensive than the tokens it saved.
# Every check below exists because of a concrete near-duplicate pair that a pure
# similarity score happily accepts. Each check is intentionally conservative:
# when in doubt we return a miss (cheap) rather than a wrong hit (expensive).

_NEG_EN_RE = re.compile(
    r"\b(?:not|n't|never|no|none|cannot|can't|cant|won't|wont|don't|dont|doesn't|doesnt|didn't|didnt|"
    r"isn't|isnt|aren't|arent|wasn't|wasnt|shouldn't|shouldnt|without|unable|fails?|failed|except|"
    r"exclude[sd]?|excluding|omit(?:s|ted)?|"
    r"disable[sd]?|deny|denied|forbidden|prohibited)\b",
    re.I,
)
_NEG_JA_RE = re.compile(
    r"(?:ない|無い|ありません|ません|できない|出来ない|しない|不可|不能|禁止|以外|除く|除外|なし|無し|"
    r"いいえ|未対応|非対応|不要|不可能)"
)
_NUM_RE = re.compile(r"(?<![A-Za-z0-9])\d+(?:[.,]\d+)*")
_UNIT_RE = re.compile(
    r"(?<![A-Za-z0-9])(\d+(?:[.,]\d+)*)\s*"
    r"(%|％|gb|mb|kb|tb|gib|mib|kg|mg|km|cm|mm|ms|sec|secs|seconds?|mins?|minutes?|hrs?|hours?|days?|weeks?|"
    r"months?|years?|usd|jpy|eur|dollars?|yen|円|ドル|件|個|人|回|年|月|日|時間|分|秒|倍|台|名)",
    re.I,
)
_YEAR_RE = re.compile(r"\b(?:19|20)\d{2}\b")
_PROPER_RE = re.compile(r"\b[A-Z][A-Za-z0-9]*(?:[-'][A-Za-z0-9]+)*\b")
_MODEL_RE = re.compile(r"\b(?:[A-Za-z]{1,12}[-_]?\d{1,6}[A-Za-z0-9]{0,8}|v\d+(?:\.\d+)+)\b")
_SENT_SPLIT_RE = re.compile(r"[.!?\n]+")

# Capitalization at the start of a sentence is grammar, not evidence of a proper
# noun, so sentence-initial words are excluded; these stopwords cover the rest.
_CAP_STOPWORDS = {
    "i", "a", "an", "the", "this", "that", "these", "those", "there", "here",
    "what", "how", "why", "when", "where", "who", "which", "whose", "whom",
    "is", "are", "am", "was", "were", "be", "been", "being", "do", "does", "did",
    "can", "could", "would", "should", "shall", "will", "may", "might", "must",
    "if", "in", "on", "at", "for", "to", "of", "with", "from", "about", "after",
    "before", "and", "but", "or", "not", "no", "yes", "so", "as", "by", "my",
    "your", "our", "their", "its", "it", "we", "you", "he", "she", "they",
    "please", "hi", "hello", "hey", "dear", "thanks", "thank", "have", "has",
    "had", "get", "got", "need", "want", "let", "make", "does", "why", "also",
}

_TIME_WORDS_EN = [
    "today", "yesterday", "tomorrow", "tonight", "now", "currently", "current",
    "latest", "newest", "recent", "recently", "this year", "last year", "next year",
    "this month", "last month", "next month", "this week", "last week", "next week",
    "this quarter", "last quarter", "ytd", "q1", "q2", "q3", "q4",
    "january", "february", "march", "april", "june", "july", "august",
    "september", "october", "november", "december",
]
# Bare "may" is deliberately absent: the modal verb ("May I reset...") is far
# more common than the month and would veto ordinary paraphrases.
# Word boundaries are mandatory here - a substring test makes "know" contain
# "now", "snowfall" contain "now" and "acknowledge" contain "now", which
# rejects perfectly good reuse with a phantom temporal_mismatch.
_TIME_RE_EN = re.compile(
    r"\b(?:" + "|".join(re.escape(w) for w in sorted(_TIME_WORDS_EN, key=len, reverse=True)) + r")\b",
    re.I,
)
_TIME_WORDS_JA = [
    "今日", "本日", "昨日", "明日", "今年", "本年", "昨年", "去年", "来年",
    "今月", "先月", "来月", "今週", "先週", "来週", "現在", "最新", "直近",
    "当時", "以前", "将来", "今期", "前期", "来期", "年度",
]

_QUESTION_PATTERNS: List[Tuple[str, re.Pattern]] = [
    ("how_much", re.compile(r"\bhow\s+(?:much|many)\b|いくら|何円|何個|何件|どのくらい|どれくらい|どれぐらい", re.I)),
    ("what", re.compile(r"\bwhat\b|\bwhich\b|何(?:です|を|が|は|の)|どれ|どちら", re.I)),
    ("how", re.compile(r"\bhow\b|どうやって|どのように|方法|手順|やり方", re.I)),
    ("why", re.compile(r"\bwhy\b|なぜ|どうして|理由|原因", re.I)),
    ("when", re.compile(r"\bwhen\b|いつ|期限|締切", re.I)),
    ("where", re.compile(r"\bwhere\b|どこ", re.I)),
    ("who", re.compile(r"\bwho(?:m|se)?\b|誰|だれ", re.I)),
    ("yesno", re.compile(r"^\s*(?:is|are|am|was|were|do|does|did|can|could|will|would|should|has|have|may)\b", re.I | re.M)),
]

GUARD_CHECKS = ("negation", "numeric", "proper_noun", "temporal", "question_type")


_REQUEST_FRAME_RE = re.compile(
    r"^\s*(?:(?:could|can|would|will)\s+you\s+(?:kindly\s+|please\s+)?"
    r"(?:tell\s+me|let\s+me\s+know|explain|clarify|confirm|advise)|"
    r"i(?:'m|\s+am)?\s*(?:was\s+)?wondering(?:\s+if|\s+whether)?|"
    r"(?:just\s+a\s+)?quick\s+question|please\s+let\s+me\s+know)\b[,:]?\s*",
    re.I,
)


def _strip_request_frame(text: str) -> str:
    """Guarantees: removes a leading polite-request wrapper so it cannot masquerade as a question type."""
    return _REQUEST_FRAME_RE.sub("", text or "", count=1)


def _negation_profile(text: str) -> bool:
    """Guarantees: True iff the text carries at least one explicit negation cue (EN or JA)."""
    return bool(_NEG_EN_RE.search(text) or _NEG_JA_RE.search(text))


def _numeric_profile(text: str) -> Tuple[frozenset, frozenset]:
    """Guarantees: (bare numbers, number+unit pairs) with thousands separators folded away."""
    numbers = set()
    for raw in _NUM_RE.findall(text):
        cleaned = raw.replace(",", "")
        try:
            value = float(cleaned)
            numbers.add(str(int(value)) if value.is_integer() else str(value))
        except Exception:
            numbers.add(cleaned)
    units = set()
    for value, unit in _UNIT_RE.findall(text):
        units.add(value.replace(",", "") + unit.lower().replace("％", "%"))
    return frozenset(numbers), frozenset(units)


def _proper_noun_profile(text: str) -> frozenset:
    """Guarantees: proper nouns / model identifiers only, with sentence-initial capitalization ignored."""
    found = set()
    for sentence in _SENT_SPLIT_RE.split(text):
        sentence = sentence.strip()
        if not sentence:
            continue
        tokens = _PROPER_RE.findall(sentence)
        if tokens and sentence.startswith(tokens[0]):
            tokens = tokens[1:]  # sentence-initial capital carries no information
        for token in tokens:
            if token.lower() in _CAP_STOPWORDS:
                continue
            if len(token) < 2:
                continue
            found.add(token.lower())
    for token in _MODEL_RE.findall(text):
        if any(ch.isdigit() for ch in token) and any(ch.isalpha() for ch in token):
            found.add(token.lower())
    return frozenset(found)


def _temporal_profile(text: str) -> frozenset:
    """Guarantees: the set of time anchors (years, relative-date words, month names) in the text."""
    lowered = text.lower()
    found = set(_YEAR_RE.findall(text))
    found.update(m.lower() for m in _TIME_RE_EN.findall(lowered))
    # Japanese has no word boundaries, so a substring test is the correct test.
    for word in _TIME_WORDS_JA:
        if word in text:
            found.add(word)
    return frozenset(found)


def _question_profile(text: str) -> frozenset:
    """Guarantees: the interrogative types present, with a polite request wrapper ignored.

    "Could you please tell me how do I reset my password?" asks the same KIND of
    question as "How do I reset my password?"; without stripping the wrapper the
    leading "Could" registers as a yes/no question and vetoes a valid reuse.
    """
    stripped = _strip_request_frame(text)
    return frozenset(label for label, pattern in _QUESTION_PATTERNS if pattern.search(stripped))


def _fmt_set(values: Iterable[str], limit: int = 4) -> str:
    """Guarantees: a short, stable, human-readable rendering of a set for the audit screen."""
    items = sorted(str(v) for v in values)
    if not items:
        return "{}"
    shown = items[:limit]
    suffix = ",+{0}".format(len(items) - limit) if len(items) > limit else ""
    return "{" + ",".join(shown) + suffix + "}"


def is_semantically_safe(query: str, candidate: str) -> Tuple[bool, List[str]]:
    """Guarantees: never raises and returns (False, reasons) whenever a meaning-changing difference is detected."""
    reasons: List[str] = []
    try:
        q = light_normalize(query)
        c = light_normalize(candidate)
        if not q or not c:
            return False, ["empty_text"]

        # 1) NEGATION FLIP.
        #    "Can I cancel my plan?" vs "Can I not cancel my plan?" -> cosine ~0.97,
        #    opposite answers. Japanese: 「返品できます」 vs 「返品できません」.
        if "negation" not in GUARD_DISABLE:
            if _negation_profile(q) != _negation_profile(c):
                reasons.append("negation_mismatch:q={0},c={1}".format(_negation_profile(q), _negation_profile(c)))

        # 2) QUANTITY / UNIT DIFFERENCE.
        #    "retry after 3 seconds" vs "retry after 30 seconds" differ by one
        #    character; "a 100GB plan" vs "a 100MB plan" differ by one letter.
        #    Any difference in the numeric token set invalidates reuse.
        if "numeric" not in GUARD_DISABLE:
            qn, qu = _numeric_profile(q)
            cn, cu = _numeric_profile(c)
            if qn != cn:
                reasons.append("numeric_mismatch:q={0},c={1}".format(_fmt_set(qn), _fmt_set(cn)))
            elif qu != cu:
                reasons.append("unit_mismatch:q={0},c={1}".format(_fmt_set(qu), _fmt_set(cu)))

        # 3) PROPER NOUN / MODEL NUMBER DIFFERENCE.
        #    "Does the X1 Carbon support 32GB?" vs "Does the X1 Yoga support 32GB?"
        #    or "migrate from Postgres" vs "migrate from Oracle": same shape,
        #    different product, completely different answer.
        if "proper_noun" not in GUARD_DISABLE:
            # q/c come from light_normalize(), which folds full-width to ASCII but
            # KEEPS case - exactly what proper-noun detection needs ("ＡＢＣ-100" and
            # "ABC-100" are the same product, "Carbon" and "Yoga" are not).
            qp = _proper_noun_profile(q)
            cp = _proper_noun_profile(c)
            if qp != cp:
                reasons.append("proper_noun_mismatch:q={0},c={1}".format(_fmt_set(qp), _fmt_set(cp)))

        # 4) TENSE / TIME ANCHOR DIFFERENCE.
        #    "What is the 2024 tax rate?" vs "What is the 2025 tax rate?", or
        #    "latest release" vs "last year's release". Time-anchored answers go
        #    stale the moment the anchor moves.
        if "temporal" not in GUARD_DISABLE:
            qt = _temporal_profile(q)
            ct = _temporal_profile(c)
            if qt != ct:
                reasons.append("temporal_mismatch:q={0},c={1}".format(_fmt_set(qt), _fmt_set(ct)))

        # 5) QUESTION TYPE SWITCH.
        #    "How do I reset my password?" vs "Why do I need to reset my password?"
        #    share almost every n-gram but ask for a procedure vs a rationale.
        if "question_type" not in GUARD_DISABLE:
            qq = _question_profile(q)
            cq = _question_profile(c)
            if qq != cq:
                reasons.append("question_type_mismatch:q={0},c={1}".format(_fmt_set(qq), _fmt_set(cq)))

        return (len(reasons) == 0), reasons
    except Exception as exc:  # fail closed - an unusable guard must not create wrong hits
        return False, ["guard_error:{0}".format(type(exc).__name__)]


# =============================================================================
# (G) Accounting
# =============================================================================

_CJK_RANGES = (
    (0x3000, 0x303F), (0x3040, 0x30FF), (0x3400, 0x4DBF), (0x4E00, 0x9FFF),
    (0xF900, 0xFAFF), (0xFF00, 0xFFEF), (0xAC00, 0xD7AF), (0x1100, 0x11FF),
)


def _is_cjk(ch: str) -> bool:
    """Guarantees: True for CJK/Kana/Hangul/full-width code points, decided offline by range."""
    code = ord(ch)
    for lo, hi in _CJK_RANGES:
        if lo <= code <= hi:
            return True
    return False


def estimate_tokens(text: Any) -> int:
    """Guarantees: a dependency-free token estimate (~4 ASCII chars = 1 token, ~1 CJK char = 1 token)."""
    if not text:
        return 0
    try:
        s = text if isinstance(text, str) else str(text)
    except Exception:
        return 0
    cjk = 0
    for ch in s:
        if _is_cjk(ch):
            cjk += 1
    other = len(s) - cjk
    return int(cjk + math.ceil(other / 4.0))


def savings(entry: Dict[str, Any], price_in: float = None, price_out: float = None) -> Dict[str, Any]:
    """Guarantees: the estimated tokens and USD avoided by one reuse of this entry; never raises."""
    try:
        pin = PRICE_IN_PER_1K if price_in is None else float(price_in)
        pout = PRICE_OUT_PER_1K if price_out is None else float(price_out)
        tokens_in = int(entry.get("tokens_in", 0) or 0)
        tokens_out = int(entry.get("tokens_out", 0) or 0)
        cost = (tokens_in / 1000.0) * pin + (tokens_out / 1000.0) * pout
        return {
            "saved_tokens_in": tokens_in,
            "saved_tokens_out": tokens_out,
            "saved_tokens": tokens_in + tokens_out,
            "saved_cost_usd": round(float(cost), 6),
        }
    except Exception:
        return {"saved_tokens_in": 0, "saved_tokens_out": 0, "saved_tokens": 0, "saved_cost_usd": 0.0}


class Accounting:
    """Guarantees: thread-safe cumulative savings counters keyed by tenant and by route."""

    def __init__(self) -> None:
        """Guarantees: an empty, lock-protected counter table; no I/O and no allocation of size."""
        self._lock = threading.RLock()
        self._rows: Dict[Tuple[str, str], Dict[str, float]] = {}

    def add(self, tenant_id: str, route: str, event: str, saved: Dict[str, Any]) -> None:
        """Guarantees: accumulates one event without ever raising into the caller."""
        try:
            with self._lock:
                key = (str(tenant_id), str(route or "default"))
                row = self._rows.get(key)
                if row is None:
                    row = {"hits": 0.0, "misses": 0.0, "stores": 0.0, "rejects": 0.0,
                           "saved_tokens": 0.0, "saved_cost_usd": 0.0}
                    if len(self._rows) < MAX_TENANTS * 64:
                        self._rows[key] = row
                    else:
                        return
                if event in ("hit", "miss", "store", "reject"):
                    row[{"hit": "hits", "miss": "misses", "store": "stores", "reject": "rejects"}[event]] += 1.0
                row["saved_tokens"] += float(saved.get("saved_tokens", 0) or 0)
                row["saved_cost_usd"] += float(saved.get("saved_cost_usd", 0.0) or 0.0)
        except Exception:
            return

    def frame(self) -> pd.DataFrame:
        """Guarantees: a DataFrame snapshot (possibly empty but always with the documented columns)."""
        cols = ["tenant_id", "route", "hits", "misses", "stores", "rejects",
                "hit_rate", "saved_tokens", "saved_cost_usd"]
        try:
            with self._lock:
                rows = [dict(tenant_id=k[0], route=k[1], **v) for k, v in self._rows.items()]
            if not rows:
                return pd.DataFrame(columns=cols)
            df = pd.DataFrame(rows)
            total = df["hits"] + df["misses"]
            df["hit_rate"] = np.where(total > 0, (df["hits"] / total).round(4), 0.0)
            df["saved_cost_usd"] = df["saved_cost_usd"].round(6)
            for col in ("hits", "misses", "stores", "rejects", "saved_tokens"):
                df[col] = df[col].astype(int)
            return df[cols].sort_values(["tenant_id", "route"]).reset_index(drop=True)
        except Exception:
            return pd.DataFrame(columns=cols)

    def reset(self) -> None:
        """Guarantees: clears every counter; used by tests and by the reset control."""
        with self._lock:
            self._rows.clear()


# =============================================================================
# (J) Enrichment - optional remote embeddings, never on the critical path
# =============================================================================


class Embedder:
    """Guarantees: an optional re-ranking signal that can fail in any way without affecting correctness."""

    def __init__(self) -> None:
        """Guarantees: no client is built and no network is touched until the first embed() call."""
        self._lock = threading.RLock()
        self._client = None
        self._cache: "OrderedDict[str, np.ndarray]" = OrderedDict()
        self._fails = 0
        self._disabled_until = 0.0
        self._calls = 0
        self._errors = 0
        self._last_error = ""
        self._dim: Optional[int] = None

    # -- internals ---------------------------------------------------------
    def _get_client(self):
        """Guarantees: lazily builds an InferenceClient (never at import time); returns None on any failure."""
        if self._client is not None:
            return self._client
        if not EMBED_ENABLED:
            return None
        try:
            from huggingface_hub import InferenceClient  # imported lazily on purpose
        except Exception as exc:
            self._last_error = "import: {0}".format(exc)
            return None
        for kwargs in (
            {"model": EMBED_MODEL, "token": HF_TOKEN, "timeout": EMBED_TIMEOUT},
            {"model": EMBED_MODEL, "api_key": HF_TOKEN, "timeout": EMBED_TIMEOUT},
            {"model": EMBED_MODEL, "token": HF_TOKEN},
            {"model": EMBED_MODEL},
        ):
            try:
                self._client = InferenceClient(**kwargs)
                return self._client
            except TypeError:
                continue
            except Exception as exc:
                self._last_error = "client: {0}".format(exc)
                return None
        return None

    def _trip(self, exc: Exception) -> None:
        """Guarantees: opens the circuit breaker after repeated failures so we stop burning credits."""
        self._errors += 1
        self._fails += 1
        self._last_error = "{0}: {1}".format(type(exc).__name__, str(exc)[:160])
        if self._fails >= EMBED_FAIL_THRESHOLD:
            self._disabled_until = time.time() + EMBED_COOLDOWN_SEC

    @staticmethod
    def _to_matrix(raw: Any, n: int) -> Optional[np.ndarray]:
        """Guarantees: converts any feature-extraction payload shape into an (n, d) float32 matrix or None."""
        arr = np.asarray(raw, dtype=np.float32)
        if arr.ndim == 3:            # (batch, tokens, dim) -> mean pool
            arr = arr.mean(axis=1)
        if arr.ndim == 1:
            arr = arr.reshape(1, -1)
        if arr.ndim != 2 or arr.shape[0] != n or arr.shape[1] < 2:
            return None
        norms = np.linalg.norm(arr, axis=1, keepdims=True)
        norms[norms == 0.0] = 1.0
        return (arr / norms).astype(np.float32)

    def _cache_get(self, key: str) -> Optional[np.ndarray]:
        """Guarantees: returns a cached embedding (marking it recently used) or None."""
        with self._lock:
            vec = self._cache.get(key)
            if vec is not None:
                self._cache.move_to_end(key)
            return vec

    def _cache_put(self, key: str, vec: np.ndarray) -> None:
        """Guarantees: stores one embedding and evicts the oldest so the cache stays at EMBED_CACHE_SIZE."""
        with self._lock:
            self._cache[key] = vec
            self._cache.move_to_end(key)
            while len(self._cache) > EMBED_CACHE_SIZE:
                self._cache.popitem(last=False)

    # -- public ------------------------------------------------------------
    def available(self) -> bool:
        """Guarantees: True only when a token exists and the circuit breaker is closed."""
        return bool(EMBED_ENABLED) and time.time() >= self._disabled_until

    def embed(self, texts: Sequence[str]) -> Optional[np.ndarray]:
        """Guarantees: returns an (n, d) L2-normalized matrix, or None on ANY problem - callers must handle None."""
        try:
            if not texts:
                return None
            texts = [(t or "")[:1000] for t in texts]
            keys = [hashlib.sha1(t.encode("utf-8", "replace")).hexdigest() for t in texts]
            cached = [self._cache_get(k) for k in keys]
            missing = [i for i, v in enumerate(cached) if v is None]

            if missing:
                if not self.available():
                    return None
                client = self._get_client()
                if client is None:
                    return None
                payload = [texts[i] for i in missing]
                self._calls += 1
                try:
                    raw = client.feature_extraction(payload if len(payload) > 1 else payload[0])
                except TypeError:
                    raw = client.feature_extraction(payload if len(payload) > 1 else payload[0], model=EMBED_MODEL)
                matrix = self._to_matrix(raw, len(payload))
                if matrix is None:
                    self._trip(ValueError("unexpected embedding shape"))
                    return None
                self._fails = 0
                self._dim = int(matrix.shape[1])
                for pos, idx in enumerate(missing):
                    vec = matrix[pos]
                    cached[idx] = vec
                    self._cache_put(keys[idx], vec)

            dims = {v.shape[0] for v in cached if v is not None}
            if len(dims) != 1 or any(v is None for v in cached):
                return None
            return np.vstack(cached).astype(np.float32)
        except Exception as exc:  # 402 / 429 / timeout / DNS / anything at all
            try:
                self._trip(exc)
            except Exception:
                pass
            return None

    def status(self) -> Dict[str, Any]:
        """Guarantees: a JSON-safe snapshot of embedding health, with no token material in it."""
        now = time.time()
        return {
            "enabled": bool(EMBED_ENABLED),
            "available_now": self.available(),
            "model": EMBED_MODEL if EMBED_ENABLED else None,
            "weight": EMBED_WEIGHT if EMBED_ENABLED else 0.0,
            "calls": self._calls,
            "errors": self._errors,
            "consecutive_failures": self._fails,
            "cooldown_remaining_sec": max(0, int(self._disabled_until - now)),
            "dim": self._dim,
            "cache_size": len(self._cache),
            "last_error": self._last_error[:160],
        }


_EMBEDDER = Embedder()


def embed_remote(texts: Sequence[str]) -> Optional[np.ndarray]:
    """Guarantees: optional remote embeddings, or None - the caller must remain fully functional without it."""
    return _EMBEDDER.embed(texts)


# =============================================================================
# (D) CacheIndex - strictly partitioned per tenant
# =============================================================================
#
# TENANT ISOLATION
# ----------------
# Every read and write goes through _partition(tenant_id), which returns exactly
# one dict. No method accepts more than one partition, no method iterates over
# self._parts to answer a lookup, and vectors/fingerprints live inside the
# partition. There is therefore no code path that can surface tenant A's entry
# to tenant B - isolation is structural, not a filter applied after the fact.

_TENANT_SAFE_RE = re.compile(r"[^A-Za-z0-9_.\-:@]+")


def normalize_tenant(tenant_id: Any) -> str:
    """Guarantees: a non-empty, bounded, filesystem/JSON-safe partition id ("default" when unset)."""
    try:
        raw = "" if tenant_id is None else str(tenant_id)
    except Exception:
        raw = ""
    raw = _TENANT_SAFE_RE.sub("_", raw.strip())[:64]
    return raw or "default"


class CacheIndex:
    """Guarantees: bounded memory, per-tenant isolation, and a three-stage lookup that never raises."""

    def __init__(
        self,
        max_entries: int = MAX_ENTRIES,
        max_text_chars: int = MAX_TEXT_CHARS,
        dim: int = VECTOR_DIM,
        name: str = "live",
        record_events: bool = True,
        max_total_entries: int = None,
    ) -> None:
        """Guarantees: an empty index with bounded limits; allocates nothing per tenant until first use."""
        self.max_entries = max(1, int(max_entries))
        self.max_total_entries = max(self.max_entries,
                                     int(max_total_entries or MAX_TOTAL_ENTRIES))
        self.max_text_chars = max(16, int(max_text_chars))
        self.dim = max(16, int(dim))
        self.name = name
        self.record_events = record_events
        self._lock = threading.RLock()
        self._parts: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
        self._last_expire_sweep = 0.0
        self._counter = 0

    # -- partition plumbing ------------------------------------------------
    def _new_partition(self, tenant_id: str) -> Dict[str, Any]:
        """Guarantees: a fresh, empty partition with the documented structure and no shared buffers."""
        cap = max(1, min(64, self.max_entries))
        return {
            "tenant_id": tenant_id,
            "exact": {},                                              # norm-key -> slot
            "entries": [],                                            # slot -> entry dict | None
            "matrix": np.zeros((cap, self.dim), dtype=np.float32),    # slot -> unit vector
            "simhashes": np.zeros(cap, dtype=np.uint64),              # slot -> fingerprint
            "alive": np.zeros(cap, dtype=bool),
            "lru": OrderedDict(),                                     # slot -> True (oldest first)
            "free": [],
            "stats": {"stored": 0, "evicted": 0, "expired": 0, "invalidated": 0,
                      "rejected": 0, "hits": 0, "misses": 0, "updates": 0},
        }

    def _partition(self, tenant_id: str, create: bool = True) -> Optional[Dict[str, Any]]:
        """Guarantees: returns exactly one tenant's partition - the single access point for all data."""
        tid = normalize_tenant(tenant_id)
        part = self._parts.get(tid)
        if part is None and create:
            if len(self._parts) >= MAX_TENANTS:
                # Evict the least-recently-used *tenant* so an unbounded tenant id
                # space cannot exhaust memory.
                self._parts.popitem(last=False)
            part = self._new_partition(tid)
            self._parts[tid] = part
        if part is not None:
            self._parts.move_to_end(tid)
        return part

    def _grow(self, part: Dict[str, Any], needed: int) -> None:
        """Guarantees: capacity grows geometrically and never beyond max_entries."""
        cap = int(part["matrix"].shape[0])
        if needed < cap:
            return
        new_cap = min(self.max_entries, max(64, cap * 2))
        while new_cap <= needed and new_cap < self.max_entries:
            new_cap = min(self.max_entries, new_cap * 2)
        if new_cap <= cap:
            return
        matrix = np.zeros((new_cap, self.dim), dtype=np.float32)
        matrix[:cap] = part["matrix"]
        part["matrix"] = matrix
        sims = np.zeros(new_cap, dtype=np.uint64)
        sims[:cap] = part["simhashes"]
        part["simhashes"] = sims
        alive = np.zeros(new_cap, dtype=bool)
        alive[:cap] = part["alive"]
        part["alive"] = alive

    def _drop_slot(self, part: Dict[str, Any], slot: int, stat_key: str) -> None:
        """Guarantees: fully removes one entry (exact key, vector, fingerprint, LRU) and frees its slot."""
        entries = part["entries"]
        if slot < 0 or slot >= len(entries):
            return
        entry = entries[slot]
        if entry is None:
            return
        part["exact"].pop(entry.get("key", ""), None)
        part["lru"].pop(slot, None)
        part["alive"][slot] = False
        part["matrix"][slot] = 0.0
        part["simhashes"][slot] = np.uint64(0)
        entries[slot] = None
        part["free"].append(slot)
        part["stats"][stat_key] = part["stats"].get(stat_key, 0) + 1

    def _live_total(self) -> int:
        """Guarantees: the number of live entries across every partition (memory accounting)."""
        return sum(len(p["lru"]) for p in self._parts.values())

    def _enforce_global_budget(self) -> None:
        """Guarantees: total entries stay under max_total_entries by evicting the LRU entry of the largest partition.

        This is an eviction-only cross-partition operation: it deletes, it never
        reads or moves data between tenants, so tenant isolation is preserved.
        """
        total = self._live_total()
        while total >= self.max_total_entries and self._parts:
            victim = max(self._parts.values(), key=lambda p: len(p["lru"]))
            if not victim["lru"]:
                break
            slot, _ = next(iter(victim["lru"].items()))
            self._drop_slot(victim, int(slot), "evicted")
            total -= 1

    def _alloc_slot(self, part: Dict[str, Any]) -> int:
        """Guarantees: returns a usable slot index, evicting the LRU entry when the partition or budget is full."""
        self._enforce_global_budget()
        while (len(part["lru"]) >= self.max_entries) and part["lru"]:
            old_slot, _ = next(iter(part["lru"].items()))
            self._drop_slot(part, int(old_slot), "evicted")
        if part["free"]:
            return int(part["free"].pop())
        slot = len(part["entries"])
        self._grow(part, slot)
        if slot >= int(part["matrix"].shape[0]):
            # Capacity is capped at max_entries: force an eviction and reuse a slot.
            if part["lru"]:
                old_slot, _ = next(iter(part["lru"].items()))
                self._drop_slot(part, int(old_slot), "evicted")
            if part["free"]:
                return int(part["free"].pop())
            raise MemoryError("index capacity exhausted")
        part["entries"].append(None)
        return slot

    def _next_id(self) -> int:
        """Guarantees: a monotonically increasing per-index counter (lock is held by the caller)."""
        self._counter += 1
        return self._counter

    def _features(self, text: str, provided: Any = None) -> Dict[str, Any]:
        """Guarantees: features that always match THIS index's vector dimension, recomputing if they do not."""
        if isinstance(provided, dict):
            vector = provided.get("vector")
            if isinstance(vector, np.ndarray) and vector.shape == (self.dim,) and provided.get("key"):
                return provided
        return featurize(text, self.dim)

    # -- (D-1) put ---------------------------------------------------------
    def put(self, tenant_id: str, prompt: str, response: str, meta: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Guarantees: stores only screened content, bounded in size and count, inside one tenant partition."""
        try:
            meta = dict(meta or {})
            tid = normalize_tenant(tenant_id)
            route = str(meta.get("route") or "default")[:64]
            prompt = "" if prompt is None else str(prompt)
            response = "" if response is None else str(response)
            allow_empty = bool(meta.get("allow_empty_response", False))
            skip_safety = bool(meta.get("skip_safety", False))

            if not prompt.strip():
                return err(ErrorCode.EMPTY_PROMPT, "prompt is empty", stored=False, reason=ReasonCode.EMPTY_PROMPT.value)
            if not response.strip() and not allow_empty:
                return err(ErrorCode.EMPTY_RESPONSE, "response is empty", stored=False,
                           reason=ReasonCode.EMPTY_RESPONSE.value)

            truncated = len(prompt) > self.max_text_chars or len(response) > self.max_text_chars
            prompt = prompt[: self.max_text_chars]
            response = response[: self.max_text_chars]

            if not skip_safety:
                safe, reason = should_cache(prompt, response if response.strip() else "x")
                if not safe:
                    with self._lock:
                        part = self._partition(tid)
                        part["stats"]["rejected"] += 1
                    return ok(stored=False, tenant_id=tid, reason=reason,
                              reason_help=REASON_HELP.get(reason, ""), truncated=truncated)

            features = self._features(prompt, meta.get("features"))
            norm = features["norm"]
            key = features["key"]
            if not norm:
                return err(ErrorCode.EMPTY_PROMPT, "prompt is empty after normalization", stored=False,
                           reason=ReasonCode.EMPTY_PROMPT.value)

            now = _safe_float(meta.get("now"), 0.0) or time.time()
            ttl = max(1, min(MAX_TTL_SEC, _safe_int(meta.get("ttl_sec"), DEFAULT_TTL_SEC) or DEFAULT_TTL_SEC))
            # An imported snapshot is untrusted input: a bogus created_at would
            # report a negative age, and a bogus expires_at (1e18, inf, NaN)
            # would create an entry that sweep_expired can never remove.
            created_at = min(_safe_float(meta.get("created_at"), 0.0) or now, now)
            expires_at = _safe_float(meta.get("expires_at"), 0.0) or (created_at + ttl)
            expires_at = min(max(expires_at, now - 1.0), now + MAX_TTL_SEC)

            tokens_in = estimate_tokens(prompt)
            tokens_out = estimate_tokens(response)

            with self._lock:
                part = self._partition(tid)
                updated = False
                slot = part["exact"].get(key)
                if slot is None or slot >= len(part["entries"]) or part["entries"][slot] is None:
                    slot = self._alloc_slot(part)
                else:
                    updated = True

                entry = {
                    "id": key,
                    "key": key,
                    "tenant_id": tid,
                    "slot": int(slot),
                    "prompt": prompt,
                    "norm": norm,
                    "response": response,
                    "route": route,
                    "meta": {k: v for k, v in meta.items()
                             if k not in ("features", "skip_safety", "allow_empty_response", "now")},
                    "created_at": created_at,
                    "expires_at": expires_at,
                    "ttl_sec": ttl,
                    "hits": max(0, _safe_int(meta.get("hits"), 0)),
                    "last_hit_at": _safe_float(meta.get("last_hit_at"), 0.0),
                    "tokens_in": tokens_in,
                    "tokens_out": tokens_out,
                    "chars": len(prompt) + len(response),
                }
                part["entries"][slot] = entry
                part["matrix"][slot] = features["vector"]
                part["simhashes"][slot] = np.uint64(int(features["simhash"]) & _MASK64)
                part["alive"][slot] = True
                part["exact"][key] = slot
                part["lru"][slot] = True
                part["lru"].move_to_end(slot)
                part["stats"]["updates" if updated else "stored"] += 1
                live = len(part["lru"])

            if self.record_events:
                OBS.record(tenant_id=tid, route=route, event="store", stage="store",
                           similarity=1.0 if updated else 0.0, latency_ms=0.0,
                           saved_tokens=0, saved_cost=0.0, guard_reasons=[], entry_id=key,
                           prompt=prompt)
                ACC.add(tid, route, "store", {"saved_tokens": 0, "saved_cost_usd": 0.0})

            return ok(stored=True, updated=updated, tenant_id=tid, entry_id=key, route=route,
                      reason=ReasonCode.OK.value, ttl_sec=ttl, expires_at=expires_at,
                      tokens_in=tokens_in, tokens_out=tokens_out, truncated=truncated,
                      entries_in_tenant=live)
        except Exception as exc:
            return err(ErrorCode.INTERNAL, "{0}: {1}".format(type(exc).__name__, exc), stored=False, where="put")

    # -- (D-2) lookup ------------------------------------------------------
    def lookup(self, tenant_id: str, prompt: str, threshold: float = None, **options: Any) -> Dict[str, Any]:
        """Guarantees: a three-stage (exact -> SimHash band -> cosine+guard) match inside one tenant partition only."""
        t0 = time.perf_counter()
        try:
            tid = normalize_tenant(tenant_id)
            route = str(options.get("route") or "default")[:64]
            use_guard = bool(options.get("use_guard", True))
            use_embed = bool(options.get("use_embed", True))
            record = bool(options.get("record", self.record_events))
            now = float(options.get("now") or time.time())
            try:
                thr = DEFAULT_THRESHOLD if threshold is None else float(threshold)
            except Exception:
                thr = DEFAULT_THRESHOLD
            thr = max(0.0, min(1.0, thr))
            prompt = "" if prompt is None else str(prompt)[: self.max_text_chars]

            if not prompt.strip():
                return self._miss(tid, route, "empty_prompt", thr, t0, record, extra={"candidates": 0})

            if not options.get("skip_expire_sweep"):
                self._maybe_sweep_expired(now)

            features = self._features(prompt, options.get("features"))
            qnorm, qkey, qvec, qsim = features["norm"], features["key"], features["vector"], features["simhash"]

            with self._lock:
                part = self._partition(tid)
                entries = part["entries"]
                n = len(entries)

                # ---- STAGE 1: exact match on the normalized key ----------
                slot = part["exact"].get(qkey)
                if slot is not None and 0 <= slot < n and entries[slot] is not None:
                    entry = entries[slot]
                    if entry["expires_at"] <= now:
                        self._drop_slot(part, slot, "expired")
                    else:
                        self._touch(part, slot, now)
                        payload = self._hit_payload(entry, 1.0, "exact", now, [])
                        part["stats"]["hits"] += 1
                        return self._finish_hit(tid, route, payload, t0, record, prompt)

                if n == 0 or not bool(part["alive"][:n].any()):
                    return self._miss(tid, route, "empty_index", thr, t0, record, extra={"candidates": 0})

                # ---- STAGE 2: SimHash band to shrink the candidate set ----
                alive = part["alive"][:n]
                dists = hamming_array(part["simhashes"][:n], qsim)
                band = alive & (dists <= SIMHASH_MAX_DIST)
                cand = np.nonzero(band)[0]
                scan_mode = "simhash_band"
                if cand.size == 0:
                    live_count = int(alive.sum())
                    if 0 < live_count <= FULL_SCAN_LIMIT:
                        # Recall safety net for small partitions only: a wide
                        # fingerprint band would otherwise hide valid paraphrases.
                        cand = np.nonzero(alive)[0]
                        scan_mode = "full_scan"
                    else:
                        return self._miss(tid, route, "no_candidates", thr, t0, record,
                                          extra={"candidates": 0, "scan_mode": scan_mode})
                if cand.size > CANDIDATE_LIMIT:
                    order = np.argsort(dists[cand], kind="stable")[:CANDIDATE_LIMIT]
                    cand = cand[order]

                # ---- STAGE 3: cosine on candidates only ------------------
                sims_local = part["matrix"][cand] @ qvec
                sims_local = np.clip(sims_local.astype(np.float64), -1.0, 1.0)
                cand_prompts = [entries[int(i)]["prompt"] for i in cand]
                cand_ids = [entries[int(i)]["id"] for i in cand]
                cand_expiry = [float(entries[int(i)]["expires_at"]) for i in cand]

            # --- optional embedding re-rank (outside the lock: it does I/O) ---
            sims_final = sims_local
            embed_used = False
            if use_embed and EMBED_WEIGHT > 0.0 and _EMBEDDER.available() and cand.size:
                top = np.argsort(-sims_local, kind="stable")[: min(EMBED_TOPK, cand.size)]
                matrix = embed_remote([prompt] + [cand_prompts[int(i)] for i in top])
                if matrix is not None and matrix.shape[0] == len(top) + 1:
                    qemb = matrix[0]
                    blended = sims_local.copy()
                    for pos, idx in enumerate(top):
                        esim = float(np.clip(float(np.dot(qemb, matrix[pos + 1])), -1.0, 1.0))
                        blended[int(idx)] = (1.0 - EMBED_WEIGHT) * sims_local[int(idx)] + EMBED_WEIGHT * esim
                    sims_final = blended
                    embed_used = True

            order = np.argsort(-sims_final, kind="stable")
            best_sim = float(sims_final[order[0]]) if order.size else 0.0
            guard_reasons: List[str] = []
            rejected_by_guard: List[Dict[str, Any]] = []

            for rank, pos in enumerate(order[:GUARD_MAX_CHECK]):
                pos = int(pos)
                sim = float(sims_final[pos])
                if sim < thr:
                    break
                if cand_expiry[pos] <= now:
                    continue
                if use_guard:
                    safe, reasons = is_semantically_safe(prompt, cand_prompts[pos])
                    if not safe:
                        rejected_by_guard.append({"entry_id": cand_ids[pos], "similarity": round(sim, 4),
                                                  "reasons": reasons})
                        guard_reasons = reasons
                        continue
                slot = int(cand[pos])
                with self._lock:
                    part = self._partition(tid)
                    if slot >= len(part["entries"]) or part["entries"][slot] is None:
                        continue
                    entry = part["entries"][slot]
                    if entry["id"] != cand_ids[pos]:
                        continue  # slot was recycled while we were re-ranking
                    self._touch(part, slot, now)
                    payload = self._hit_payload(entry, sim, "embedding" if embed_used else "cosine", now, [])
                    part["stats"]["hits"] += 1
                payload["scan_mode"] = scan_mode
                payload["candidates"] = int(cand.size)
                payload["embed_used"] = embed_used
                payload["rank"] = rank
                payload["guard_rejected"] = rejected_by_guard
                payload["borderline"] = bool(abs(sim - thr) <= BORDERLINE_MARGIN)
                return self._finish_hit(tid, route, payload, t0, record, prompt)

            reason = "guard_rejected" if rejected_by_guard else "below_threshold"
            return self._miss(
                tid, route, reason, thr, t0, record,
                extra={
                    "candidates": int(cand.size),
                    "best_similarity": round(best_sim, 4),
                    "scan_mode": scan_mode,
                    "embed_used": embed_used,
                    "guard_reasons": guard_reasons,
                    "guard_rejected": rejected_by_guard,
                    "borderline": bool(abs(best_sim - thr) <= BORDERLINE_MARGIN),
                },
            )
        except Exception as exc:
            return err(ErrorCode.INTERNAL, "{0}: {1}".format(type(exc).__name__, exc), hit=False, where="lookup")

    # -- lookup helpers ----------------------------------------------------
    def _touch(self, part: Dict[str, Any], slot: int, now: float) -> None:
        """Guarantees: marks an entry as most-recently-used and counts the reuse."""
        entry = part["entries"][slot]
        if entry is None:
            return
        entry["hits"] = int(entry.get("hits", 0)) + 1
        entry["last_hit_at"] = now
        part["lru"][slot] = True
        part["lru"].move_to_end(slot)

    def _hit_payload(self, entry: Dict[str, Any], sim: float, stage: str, now: float,
                     guard_reasons: List[str]) -> Dict[str, Any]:
        """Guarantees: a uniform hit payload carrying stage, similarity, entry_id and age_sec."""
        return {
            "hit": True,
            "stage": stage,
            "similarity": round(float(sim), 6),
            "entry_id": entry["id"],
            "age_sec": round(max(0.0, now - float(entry["created_at"])), 3),
            "ttl_remaining_sec": round(max(0.0, float(entry["expires_at"]) - now), 3),
            "response": entry["response"],
            "route": entry["route"],
            "hits": entry["hits"],
            "tokens_in": entry["tokens_in"],
            "tokens_out": entry["tokens_out"],
            "guard_reasons": list(guard_reasons),
            "cached_prompt_preview": entry["prompt"][:200],
        }

    def _finish_hit(self, tid: str, route: str, payload: Dict[str, Any], t0: float,
                    record: bool, prompt: str) -> Dict[str, Any]:
        """Guarantees: attaches savings + latency to a hit and records exactly one observability event."""
        latency_ms = round((time.perf_counter() - t0) * 1000.0, 3)
        saved = savings({"tokens_in": payload.get("tokens_in", 0), "tokens_out": payload.get("tokens_out", 0)})
        payload.setdefault("candidates", 0)
        payload.setdefault("scan_mode", "exact")
        payload.setdefault("embed_used", False)
        payload.setdefault("borderline", False)
        payload.setdefault("guard_rejected", [])
        payload["latency_ms"] = latency_ms
        payload["savings"] = saved
        payload["tenant_id"] = tid
        if record:
            OBS.record(tenant_id=tid, route=route, event="hit", stage=payload["stage"],
                       similarity=payload["similarity"], latency_ms=latency_ms,
                       saved_tokens=saved["saved_tokens"], saved_cost=saved["saved_cost_usd"],
                       guard_reasons=payload.get("guard_reasons", []), entry_id=payload["entry_id"],
                       borderline=payload["borderline"], prompt=prompt)
            ACC.add(tid, route, "hit", saved)
        return ok(**payload)

    def _miss(self, tid: str, route: str, reason: str, thr: float, t0: float, record: bool,
              extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Guarantees: a uniform miss payload that always states WHY nothing was reused."""
        latency_ms = round((time.perf_counter() - t0) * 1000.0, 3)
        payload: Dict[str, Any] = {
            "hit": False,
            "stage": "miss",
            "similarity": 0.0,
            "entry_id": None,
            "age_sec": None,
            "reason": reason,
            "threshold": round(thr, 4),
            "latency_ms": latency_ms,
            "tenant_id": tid,
            "savings": {"saved_tokens_in": 0, "saved_tokens_out": 0, "saved_tokens": 0, "saved_cost_usd": 0.0},
        }
        if extra:
            payload.update(extra)
        payload["similarity"] = float(payload.get("best_similarity", 0.0) or 0.0)
        if record:
            with self._lock:
                self._partition(tid)["stats"]["misses"] += 1
            OBS.record(tenant_id=tid, route=route, event="miss", stage=reason,
                       similarity=payload["similarity"], latency_ms=latency_ms,
                       saved_tokens=0, saved_cost=0.0,
                       guard_reasons=payload.get("guard_reasons", []), entry_id=None,
                       borderline=bool(payload.get("borderline", False)), prompt=None)
            ACC.add(tid, route, "miss", {"saved_tokens": 0, "saved_cost_usd": 0.0})
        return ok(**payload)

    # -- (D-3) invalidate --------------------------------------------------
    def invalidate(self, tenant_id: str, key_prefix: str) -> Dict[str, Any]:
        """Guarantees: removes only entries of this tenant whose key starts with the prefix; returns the count."""
        try:
            tid = normalize_tenant(tenant_id)
            prefix = "" if key_prefix is None else str(key_prefix).strip()
            if not prefix:
                return err(ErrorCode.EMPTY_PREFIX,
                           'key_prefix is required; pass "*" to invalidate the whole tenant', removed=0)
            with self._lock:
                part = self._partition(tid, create=False)
                if part is None:
                    return ok(removed=0, tenant_id=tid, key_prefix=prefix)
                targets = []
                for slot, entry in enumerate(part["entries"]):
                    if entry is None:
                        continue
                    if prefix == "*" or entry["id"].startswith(prefix) or entry["route"] == prefix:
                        targets.append(slot)
                for slot in targets:
                    self._drop_slot(part, slot, "invalidated")
                remaining = len(part["lru"])
            return ok(removed=len(targets), tenant_id=tid, key_prefix=prefix, remaining=remaining)
        except Exception as exc:
            return err(ErrorCode.INTERNAL, "{0}: {1}".format(type(exc).__name__, exc), removed=0, where="invalidate")

    # -- (D-4) TTL sweep ---------------------------------------------------
    def sweep_expired(self, now: Optional[float] = None) -> int:
        """Guarantees: drops every entry whose TTL has passed, in every partition, and returns how many."""
        removed = 0
        try:
            now = float(now or time.time())
            with self._lock:
                for part in list(self._parts.values()):
                    for slot, entry in enumerate(part["entries"]):
                        if entry is not None and float(entry["expires_at"]) <= now:
                            self._drop_slot(part, slot, "expired")
                            removed += 1
                self._last_expire_sweep = now
        except Exception:
            return removed
        return removed

    def _maybe_sweep_expired(self, now: float) -> None:
        """Guarantees: the per-lookup TTL sweep stays cheap by running at most once per interval."""
        if now - self._last_expire_sweep >= EXPIRE_SWEEP_INTERVAL:
            self.sweep_expired(now)

    # -- (D-5) export / import --------------------------------------------
    def export_json(self, include_responses: bool = True, tenant_id: Optional[str] = None) -> Dict[str, Any]:
        """Guarantees: a portable JSON snapshot (vectors recomputed on import) with responses optional."""
        try:
            now = time.time()
            with self._lock:
                tenants = {}
                items = self._parts.items() if not tenant_id else [
                    (normalize_tenant(tenant_id), self._parts.get(normalize_tenant(tenant_id)))
                ]
                for tid, part in items:
                    if part is None:
                        continue
                    rows = []
                    for entry in part["entries"]:
                        if entry is None or float(entry["expires_at"]) <= now:
                            continue
                        rows.append({
                            "id": entry["id"],
                            "prompt": entry["prompt"],
                            "response": entry["response"] if include_responses else None,
                            "route": entry["route"],
                            "created_at": entry["created_at"],
                            "expires_at": entry["expires_at"],
                            "ttl_sec": entry["ttl_sec"],
                            "hits": entry["hits"],
                            "tokens_in": entry["tokens_in"],
                            "tokens_out": entry["tokens_out"],
                        })
                    tenants[tid] = {"entries": rows, "stats": dict(part["stats"])}
            return {
                "format": "echocache-index",
                "version": 1,
                "exported_at": now,
                "include_responses": bool(include_responses),
                "config": {"vector_dim": self.dim, "ngram_n": NGRAM_N, "max_text_chars": self.max_text_chars},
                "tenants": tenants,
            }
        except Exception as exc:
            return {"format": "echocache-index", "version": 1, "error": str(exc), "tenants": {}}

    def import_json(self, payload: Any, skip_safety: bool = False) -> Dict[str, Any]:
        """Guarantees: merges a snapshot without ever crossing tenant boundaries; re-screens and re-vectorizes."""
        try:
            if isinstance(payload, (bytes, bytearray)):
                payload = payload.decode("utf-8", "replace")
            if isinstance(payload, str):
                if len(payload) > IMPORT_MAX_BYTES:
                    return err(ErrorCode.PAYLOAD_TOO_LARGE, "payload exceeds IMPORT_MAX_BYTES", imported=0)
                payload = json.loads(payload)
            if not isinstance(payload, dict) or not isinstance(payload.get("tenants"), dict):
                return err(ErrorCode.BAD_PAYLOAD, "expected an echocache-index JSON object", imported=0)

            now = time.time()
            imported = skipped_no_response = expired = rejected = failed = 0
            for tid, blob in payload["tenants"].items():
                if not isinstance(blob, dict):
                    continue
                for row in blob.get("entries", []) or []:
                    try:
                        response = row.get("response")
                        if not response:
                            skipped_no_response += 1
                            continue
                        expires_at = _safe_float(row.get("expires_at"), 0.0) or (now + DEFAULT_TTL_SEC)
                        expires_at = min(expires_at, now + MAX_TTL_SEC)
                        if expires_at <= now:
                            expired += 1
                            continue
                        res = self.put(tid, row.get("prompt") or "", response, {
                            "route": row.get("route") or "default",
                            "created_at": min(_safe_float(row.get("created_at"), 0.0) or now, now),
                            "expires_at": expires_at,
                            "ttl_sec": _safe_int(row.get("ttl_sec"), DEFAULT_TTL_SEC),
                            "hits": max(0, _safe_int(row.get("hits"), 0)),
                            "skip_safety": bool(skip_safety),
                            "now": now,
                        })
                        if res.get("ok") and res.get("stored"):
                            imported += 1
                        elif res.get("ok"):
                            rejected += 1
                        else:
                            failed += 1
                    except Exception:
                        failed += 1
            return ok(imported=imported, skipped_no_response=skipped_no_response,
                      expired=expired, rejected_by_safety=rejected, failed=failed,
                      tenants=len(payload["tenants"]))
        except Exception as exc:
            return err(ErrorCode.BAD_PAYLOAD, "{0}: {1}".format(type(exc).__name__, exc), imported=0)

    # -- introspection -----------------------------------------------------
    def health(self) -> Dict[str, Any]:
        """Guarantees: a JSON-safe snapshot of size, eviction counters and approximate memory use."""
        try:
            with self._lock:
                tenants = []
                total_entries = total_chars = 0
                matrix_bytes = 0
                agg = {"stored": 0, "evicted": 0, "expired": 0, "invalidated": 0,
                       "rejected": 0, "hits": 0, "misses": 0, "updates": 0}
                for tid, part in self._parts.items():
                    live = len(part["lru"])
                    chars = sum(e["chars"] for e in part["entries"] if e is not None)
                    matrix_bytes += int(part["matrix"].nbytes) + int(part["simhashes"].nbytes)
                    total_entries += live
                    total_chars += chars
                    for k in agg:
                        agg[k] += int(part["stats"].get(k, 0))
                    tenants.append({"tenant_id": tid, "entries": live, "chars": chars,
                                    "capacity": int(part["matrix"].shape[0]), **part["stats"]})
            approx_mb = round((matrix_bytes + total_chars * 2) / (1024.0 * 1024.0), 2)
            return {
                "entries": total_entries,
                "tenants": len(tenants),
                "capacity_per_tenant": self.max_entries,
                "capacity_total": self.max_total_entries,
                "vector_matrix_mb_per_full_tenant": round(self.max_entries * self.dim * 4 / (1024.0 * 1024.0), 1),
                "approx_memory_mb": approx_mb,
                "vector_bytes_per_entry": self.dim * 4,
                "totals": agg,
                "per_tenant": tenants,
            }
        except Exception as exc:
            return {"entries": 0, "tenants": 0, "error": str(exc), "per_tenant": []}

    def top_entries(self, limit: int = 20, tenant_id: Optional[str] = None) -> pd.DataFrame:
        """Guarantees: the most-reused live entries as a DataFrame, never containing response bodies."""
        cols = ["tenant_id", "entry_id", "route", "hits", "age_sec", "ttl_remaining_sec",
                "tokens_in", "tokens_out", "prompt_preview"]
        try:
            now = time.time()
            rows = []
            with self._lock:
                for tid, part in self._parts.items():
                    if tenant_id and tid != normalize_tenant(tenant_id):
                        continue
                    for entry in part["entries"]:
                        if entry is None:
                            continue
                        rows.append({
                            "tenant_id": tid,
                            "entry_id": entry["id"][:16],
                            "route": entry["route"],
                            "hits": entry["hits"],
                            "age_sec": round(now - entry["created_at"], 1),
                            "ttl_remaining_sec": round(max(0.0, entry["expires_at"] - now), 1),
                            "tokens_in": entry["tokens_in"],
                            "tokens_out": entry["tokens_out"],
                            "prompt_preview": entry["prompt"][:80].replace("\n", " "),
                        })
            if not rows:
                return pd.DataFrame(columns=cols)
            df = pd.DataFrame(rows).sort_values(["hits", "age_sec"], ascending=[False, True])
            return df.head(max(1, int(limit)))[cols].reset_index(drop=True)
        except Exception:
            return pd.DataFrame(columns=cols)


# =============================================================================
# (I) Observability - ring buffer + pandas aggregation, prompt-free by default
# =============================================================================

EVENT_COLUMNS = ["ts", "datetime", "tenant_id", "route", "event", "stage", "similarity",
                 "latency_ms", "saved_tokens", "saved_cost", "guard_reasons", "entry_id",
                 "borderline", "prompt_prefix"]


class EventLog:
    """Guarantees: a bounded, thread-safe event ring buffer that never stores prompt text unless opted in."""

    def __init__(self, maxlen: int = EVENT_BUFFER) -> None:
        """Guarantees: a fixed-size ring buffer; memory cannot grow past maxlen events."""
        self._lock = threading.RLock()
        self._buf: deque = deque(maxlen=max(100, int(maxlen)))
        self._dropped = 0
        self._total = 0

    def record(self, **kw: Any) -> None:
        """Guarantees: records one event, truncating prompt text per LOG_PROMPT_PREFIX (0 = never)."""
        try:
            now = time.time()
            prompt = kw.get("prompt")
            prefix = ""
            if LOG_PROMPT_PREFIX > 0 and isinstance(prompt, str):
                prefix = prompt[:LOG_PROMPT_PREFIX].replace("\n", " ")
            reasons = kw.get("guard_reasons") or []
            if not isinstance(reasons, (list, tuple)):
                reasons = [str(reasons)]
            row = {
                "ts": now,
                "tenant_id": str(kw.get("tenant_id") or "default"),
                "route": str(kw.get("route") or "default"),
                "event": str(kw.get("event") or "unknown"),
                "stage": str(kw.get("stage") or ""),
                "similarity": float(kw.get("similarity") or 0.0),
                "latency_ms": float(kw.get("latency_ms") or 0.0),
                "saved_tokens": int(kw.get("saved_tokens") or 0),
                "saved_cost": float(kw.get("saved_cost") or 0.0),
                "guard_reasons": [str(r)[:160] for r in reasons][:8],
                "entry_id": (str(kw.get("entry_id"))[:16] if kw.get("entry_id") else ""),
                "borderline": bool(kw.get("borderline", False)),
                "prompt_prefix": prefix,
            }
            with self._lock:
                if len(self._buf) == self._buf.maxlen:
                    self._dropped += 1
                self._buf.append(row)
                self._total += 1
        except Exception:
            return

    def frame(self) -> pd.DataFrame:
        """Guarantees: a DataFrame with EVENT_COLUMNS, even when the buffer is empty."""
        try:
            with self._lock:
                rows = list(self._buf)
            if not rows:
                return pd.DataFrame(columns=EVENT_COLUMNS)
            df = pd.DataFrame(rows)
            df["datetime"] = pd.to_datetime(df["ts"], unit="s", utc=True)
            df["guard_reasons"] = df["guard_reasons"].apply(lambda r: " | ".join(r) if isinstance(r, list) else str(r))
            return df[EVENT_COLUMNS]
        except Exception:
            return pd.DataFrame(columns=EVENT_COLUMNS)

    def stats(self) -> Dict[str, Any]:
        """Guarantees: buffer occupancy figures so operators know when events are being dropped."""
        with self._lock:
            return {"buffered": len(self._buf), "capacity": self._buf.maxlen,
                    "total_recorded": self._total, "dropped": self._dropped}

    def reset(self) -> None:
        """Guarantees: empties the ring buffer without disturbing the cache index."""
        with self._lock:
            self._buf.clear()
            self._dropped = 0
            self._total = 0


OBS = EventLog()
ACC = Accounting()
INDEX = CacheIndex()


def _filter_events(df: pd.DataFrame, tenant_id: str = "", route: str = "") -> pd.DataFrame:
    """Guarantees: applies optional tenant/route filters without mutating the source frame."""
    out = df
    if tenant_id:
        out = out[out["tenant_id"] == normalize_tenant(tenant_id)]
    if route:
        out = out[out["route"] == str(route)]
    return out


def hit_rate_timeseries(df: pd.DataFrame, freq: str = "1min", window: int = 5) -> pd.DataFrame:
    """Guarantees: a per-bucket hit-rate series plus its moving average, empty-safe."""
    cols = ["bucket", "hits", "misses", "requests", "hit_rate", "hit_rate_ma"]
    try:
        served = df[df["event"].isin(["hit", "miss"])].copy()
        if served.empty:
            return pd.DataFrame(columns=cols)
        served = served.set_index("datetime")
        grouped = served.groupby([pd.Grouper(freq=freq), "event"]).size().unstack(fill_value=0)
        for col in ("hit", "miss"):
            if col not in grouped.columns:
                grouped[col] = 0
        out = grouped.reset_index().rename(columns={"datetime": "bucket", "hit": "hits", "miss": "misses"})
        out["requests"] = out["hits"] + out["misses"]
        out["hit_rate"] = np.where(out["requests"] > 0, out["hits"] / out["requests"], 0.0).round(4)
        out["hit_rate_ma"] = out["hit_rate"].rolling(max(1, int(window)), min_periods=1).mean().round(4)
        out["bucket"] = out["bucket"].dt.strftime("%Y-%m-%d %H:%M")
        return out[cols]
    except Exception:
        return pd.DataFrame(columns=cols)


def savings_cumulative(df: pd.DataFrame, by: str = "tenant_id") -> pd.DataFrame:
    """Guarantees: cumulative saved tokens/cost grouped by tenant or route, empty-safe."""
    cols = [by, "bucket", "saved_tokens", "saved_cost", "cum_saved_tokens", "cum_saved_cost"]
    try:
        hits = df[df["event"] == "hit"].copy()
        if hits.empty:
            return pd.DataFrame(columns=cols)
        hits = hits.set_index("datetime")
        grouped = (hits.groupby([by, pd.Grouper(freq="1min")])[["saved_tokens", "saved_cost"]]
                   .sum().reset_index().rename(columns={"datetime": "bucket"}))
        grouped = grouped.sort_values([by, "bucket"])
        grouped["cum_saved_tokens"] = grouped.groupby(by)["saved_tokens"].cumsum()
        grouped["cum_saved_cost"] = grouped.groupby(by)["saved_cost"].cumsum().round(6)
        grouped["bucket"] = grouped["bucket"].dt.strftime("%Y-%m-%d %H:%M")
        return grouped[cols]
    except Exception:
        return pd.DataFrame(columns=cols)


def latency_distribution(df: pd.DataFrame) -> pd.DataFrame:
    """Guarantees: p50/p90/p99 latency split by hit vs miss, empty-safe."""
    cols = ["event", "count", "mean_ms", "p50_ms", "p90_ms", "p99_ms", "max_ms"]
    try:
        served = df[df["event"].isin(["hit", "miss"])]
        if served.empty:
            return pd.DataFrame(columns=cols)
        rows = []
        for event, group in served.groupby("event"):
            values = group["latency_ms"].astype(float)
            rows.append({
                "event": event,
                "count": int(values.size),
                "mean_ms": round(float(values.mean()), 3),
                "p50_ms": round(float(values.quantile(0.50)), 3),
                "p90_ms": round(float(values.quantile(0.90)), 3),
                "p99_ms": round(float(values.quantile(0.99)), 3),
                "max_ms": round(float(values.max()), 3),
            })
        return pd.DataFrame(rows)[cols]
    except Exception:
        return pd.DataFrame(columns=cols)


def borderline_events(df: pd.DataFrame, limit: int = 200) -> pd.DataFrame:
    """Guarantees: every decision whose similarity fell within +/- BORDERLINE_MARGIN of the threshold."""
    cols = ["datetime", "tenant_id", "route", "event", "stage", "similarity", "entry_id", "guard_reasons"]
    try:
        sub = df[df["borderline"] == True]  # noqa: E712 - explicit for pandas masks
        if sub.empty:
            return pd.DataFrame(columns=cols)
        sub = sub.sort_values("ts", ascending=False).head(max(1, int(limit))).copy()
        sub["datetime"] = sub["datetime"].dt.strftime("%Y-%m-%d %H:%M:%S")
        return sub[cols].reset_index(drop=True)
    except Exception:
        return pd.DataFrame(columns=cols)


def guard_rejection_events(df: pd.DataFrame, limit: int = 200) -> pd.DataFrame:
    """Guarantees: every case where MismatchGuard refused a similar-enough candidate, with its reasons."""
    cols = ["datetime", "tenant_id", "route", "stage", "similarity", "guard_reasons", "prompt_prefix"]
    try:
        sub = df[(df["guard_reasons"].astype(str).str.len() > 0)]
        if sub.empty:
            return pd.DataFrame(columns=cols)
        sub = sub.sort_values("ts", ascending=False).head(max(1, int(limit))).copy()
        sub["datetime"] = sub["datetime"].dt.strftime("%Y-%m-%d %H:%M:%S")
        return sub[cols].reset_index(drop=True)
    except Exception:
        return pd.DataFrame(columns=cols)


def _prune_temp_files(max_age_sec: float = 3600.0, keep: int = 200) -> None:
    """Guarantees: old generated files are removed so a long-lived Space cannot fill its temp dir."""
    try:
        now = time.time()
        names = os.listdir(_TMP_DIR)
        if len(names) <= keep:
            names = [n for n in names if n.startswith("echocache-")]
        for name in names:
            if not name.startswith("echocache-"):
                continue
            path = os.path.join(_TMP_DIR, name)
            try:
                if now - os.path.getmtime(path) > max_age_sec:
                    os.remove(path)
            except Exception:
                continue
    except Exception:
        return


def write_temp_file(content: str, prefix: str, suffix: str) -> Optional[str]:
    """Guarantees: a UNIQUE file per call - two concurrent exports can never overwrite each other."""
    try:
        _prune_temp_files()
        stamp = time.strftime("%Y%m%d-%H%M%S")
        # mkstemp gives an atomically unique name: a plain timestamp collides for
        # two downloads in the same second, and one tenant would then receive the
        # other tenant's file.
        fd, path = tempfile.mkstemp(prefix="{0}-{1}-".format(prefix, stamp), suffix=suffix, dir=_TMP_DIR)
        with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle:
            handle.write(content)
        return path
    except Exception:
        return None


# =============================================================================
# Tiny dependency-free SVG charts (no matplotlib, no gradio plot API drift)
# =============================================================================

_PALETTE = ["#2f81f7", "#e36209", "#2da44e", "#a371f7", "#cf222e", "#0969da"]


def _svg_frame(body: str, title: str, width: int, height: int) -> str:
    """Guarantees: a themed SVG wrapper that renders correctly in both light and dark Gradio themes."""
    return (
        '<div style="width:100%;overflow-x:auto">'
        '<div style="font-weight:600;margin:4px 0 2px 2px">{title}</div>'
        '<svg viewBox="0 0 {w} {h}" width="100%" height="{h}" preserveAspectRatio="xMidYMid meet" '
        'style="color:var(--body-text-color,#888);font-family:ui-sans-serif,system-ui,sans-serif">{body}</svg>'
        "</div>"
    ).format(title=html_lib.escape(title or ""), w=width, h=height, body=body)


def _empty_chart(title: str, message: str = "no data yet") -> str:
    """Guarantees: a readable placeholder instead of a broken chart when there is nothing to draw."""
    return (
        '<div style="padding:12px;border:1px dashed var(--border-color-primary,#666);border-radius:8px;opacity:.7">'
        "<b>{0}</b><br/>{1}</div>"
    ).format(html_lib.escape(title or ""), html_lib.escape(message))


def svg_line_chart(x_labels: Sequence[str], series: Dict[str, Sequence[float]], title: str = "",
                   y_suffix: str = "", width: int = 760, height: int = 260) -> str:
    """Guarantees: renders one or more aligned series as inline SVG, or a placeholder when empty."""
    try:
        series = {k: [float(v) for v in vals] for k, vals in (series or {}).items() if vals is not None}
        n = len(x_labels or [])
        if n == 0 or not series:
            return _empty_chart(title)
        pad_l, pad_r, pad_t, pad_b = 56, 16, 12, 34
        plot_w = max(10, width - pad_l - pad_r)
        plot_h = max(10, height - pad_t - pad_b)
        all_values = [v for vals in series.values() for v in vals if v == v]
        lo = min(all_values + [0.0])
        hi = max(all_values + [0.0])
        if hi - lo < 1e-12:
            hi = lo + 1.0
        span = hi - lo

        def xpos(i: int) -> float:
            """Guarantees: maps a series index onto the x axis, including the single-point case."""
            return pad_l + (plot_w * (i / max(1, n - 1)) if n > 1 else plot_w / 2.0)

        def ypos(v: float) -> float:
            """Guarantees: maps a value onto the y axis; a flat series never divides by zero."""
            return pad_t + plot_h - ((v - lo) / span) * plot_h

        parts = ['<rect x="0" y="0" width="{0}" height="{1}" fill="none"/>'.format(width, height)]
        for frac in (0.0, 0.25, 0.5, 0.75, 1.0):
            y = pad_t + plot_h * frac
            value = hi - span * frac
            parts.append('<line x1="{0}" y1="{1:.1f}" x2="{2}" y2="{1:.1f}" stroke="currentColor" '
                         'stroke-opacity=".18" stroke-width="1"/>'.format(pad_l, y, pad_l + plot_w))
            parts.append('<text x="{0}" y="{1:.1f}" font-size="10" fill="currentColor" fill-opacity=".75" '
                         'text-anchor="end">{2}</text>'.format(pad_l - 6, y + 3, html_lib.escape(
                             "{0:,.4g}{1}".format(value, y_suffix))))
        for idx, (name, values) in enumerate(list(series.items())[:6]):
            color = _PALETTE[idx % len(_PALETTE)]
            points = " ".join("{0:.1f},{1:.1f}".format(xpos(i), ypos(v))
                              for i, v in enumerate(values[:n]) if v == v)
            if not points:
                continue
            parts.append('<polyline points="{0}" fill="none" stroke="{1}" stroke-width="2" '
                         'stroke-linejoin="round"/>'.format(points, color))
            if n == 1:
                parts.append('<circle cx="{0:.1f}" cy="{1:.1f}" r="3" fill="{2}"/>'.format(
                    xpos(0), ypos(values[0]), color))
            parts.append('<rect x="{0}" y="{1}" width="9" height="9" fill="{2}"/>'.format(
                pad_l + 6 + idx * 110, height - 12, color))
            parts.append('<text x="{0}" y="{1}" font-size="10" fill="currentColor">{2}</text>'.format(
                pad_l + 18 + idx * 110, height - 4, html_lib.escape(str(name)[:14])))
        for i in (0, n // 2, n - 1) if n > 2 else range(n):
            parts.append('<text x="{0:.1f}" y="{1}" font-size="9" fill="currentColor" fill-opacity=".75" '
                         'text-anchor="middle">{2}</text>'.format(
                             xpos(i), pad_t + plot_h + 14, html_lib.escape(str(x_labels[i])[:16])))
        return _svg_frame("".join(parts), title, width, height)
    except Exception:
        return _empty_chart(title, "chart failed to render")


def svg_bar_chart(labels: Sequence[str], values: Sequence[float], title: str = "", y_suffix: str = "",
                  width: int = 760, height: int = 260) -> str:
    """Guarantees: renders a labelled bar chart as inline SVG, or a placeholder when empty."""
    try:
        labels = [str(x) for x in (labels or [])]
        values = [float(v) for v in (values or [])]
        if not labels or not values:
            return _empty_chart(title)
        n = min(len(labels), len(values))
        labels, values = labels[:n], values[:n]
        pad_l, pad_r, pad_t, pad_b = 56, 16, 12, 40
        plot_w = max(10, width - pad_l - pad_r)
        plot_h = max(10, height - pad_t - pad_b)
        hi = max(values + [0.0])
        lo = min(values + [0.0])
        if hi - lo < 1e-12:
            hi = lo + 1.0
        span = hi - lo
        bar_w = plot_w / max(1, n)
        parts = []
        for frac in (0.0, 0.5, 1.0):
            y = pad_t + plot_h * frac
            parts.append('<line x1="{0}" y1="{1:.1f}" x2="{2}" y2="{1:.1f}" stroke="currentColor" '
                         'stroke-opacity=".18"/>'.format(pad_l, y, pad_l + plot_w))
            parts.append('<text x="{0}" y="{1:.1f}" font-size="10" fill="currentColor" fill-opacity=".75" '
                         'text-anchor="end">{2}</text>'.format(pad_l - 6, y + 3,
                                                               html_lib.escape("{0:,.4g}{1}".format(hi - span * frac, y_suffix))))
        for i, value in enumerate(values):
            h = max(1.0, ((value - lo) / span) * plot_h)
            x = pad_l + i * bar_w + bar_w * 0.15
            y = pad_t + plot_h - h
            parts.append('<rect x="{0:.1f}" y="{1:.1f}" width="{2:.1f}" height="{3:.1f}" fill="{4}" '
                         'rx="2"/>'.format(x, y, bar_w * 0.7, h, _PALETTE[i % len(_PALETTE)]))
            parts.append('<text x="{0:.1f}" y="{1:.1f}" font-size="9" fill="currentColor" '
                         'text-anchor="middle">{2}</text>'.format(x + bar_w * 0.35, y - 3,
                                                                  html_lib.escape("{0:,.4g}".format(value))))
            parts.append('<text x="{0:.1f}" y="{1}" font-size="9" fill="currentColor" fill-opacity=".8" '
                         'text-anchor="middle">{2}</text>'.format(x + bar_w * 0.35, pad_t + plot_h + 14,
                                                                  html_lib.escape(labels[i][:10])))
        return _svg_frame("".join(parts), title, width, height)
    except Exception:
        return _empty_chart(title, "chart failed to render")


# =============================================================================
# (H) ThresholdSweep - "what would a conservative threshold actually save?"
# =============================================================================

DEFAULT_SWEEP_THRESHOLDS = [0.70, 0.74, 0.78, 0.82, 0.86, 0.90, 0.92, 0.94, 0.96, 0.98]
SWEEP_COLUMNS = ["threshold", "guard", "requests", "hits", "reuse_rate", "exact_hits", "semantic_hits",
                 "borderline_hits", "guard_blocks", "safety_rejects", "stored_entries",
                 "saved_tokens", "saved_cost_usd", "avg_latency_ms"]


def _coerce_logs(logs_df: Any) -> pd.DataFrame:
    """Guarantees: returns a frame with tenant_id/prompt/response/route columns, or raises ValueError with a clear message."""
    if logs_df is None:
        raise ValueError("no log data provided")
    df = logs_df if isinstance(logs_df, pd.DataFrame) else pd.DataFrame(logs_df)
    df = df.copy()
    df.columns = [str(c).strip().lower() for c in df.columns]
    # Two headers can collide once lower-cased ("prompt" + "PROMPT"); keeping both
    # makes df["prompt"] a DataFrame and every later .str call explodes.
    if df.columns.duplicated().any():
        df = df.loc[:, ~df.columns.duplicated()]
    if "prompt" not in df.columns:
        raise ValueError("CSV must contain a 'prompt' column (got: {0})".format(", ".join(map(str, df.columns))[:200]))
    if "tenant_id" not in df.columns:
        df["tenant_id"] = "default"
    if "response" not in df.columns:
        df["response"] = ""
    if "route" not in df.columns:
        df["route"] = "default"
    df = df[["tenant_id", "prompt", "response", "route"]].copy()
    df["prompt"] = df["prompt"].astype(str)
    df["response"] = df["response"].fillna("").astype(str)
    df["tenant_id"] = df["tenant_id"].fillna("default").astype(str)
    df["route"] = df["route"].fillna("default").astype(str)
    df = df[df["prompt"].str.strip() != ""]
    return df.reset_index(drop=True)


def sweep(logs_df: Any, thresholds: Sequence[float] = None, price_in: float = None, price_out: float = None,
          max_rows: int = SWEEP_MAX_ROWS, time_budget_sec: float = SWEEP_TIME_BUDGET_SEC) -> pd.DataFrame:
    """Guarantees: replays a log against a throwaway index per configuration; the live cache is never touched."""
    thresholds = list(thresholds or DEFAULT_SWEEP_THRESHOLDS)
    thresholds = sorted({round(_safe_float(t, -1.0), 4) for t in thresholds
                         if 0.0 <= _safe_float(t, -1.0) <= 1.0}) or list(DEFAULT_SWEEP_THRESHOLDS)
    price_in = _safe_float(price_in, PRICE_IN_PER_1K)
    price_out = _safe_float(price_out, PRICE_OUT_PER_1K)

    df = _coerce_logs(logs_df)
    total_rows = len(df)
    truncated = total_rows > max_rows
    if truncated:
        df = df.head(max_rows)
    rows_used = len(df)
    if rows_used == 0:
        return pd.DataFrame(columns=SWEEP_COLUMNS)

    # Features are computed once and shared by every configuration: identical
    # inputs, no recomputation, and a fair comparison across thresholds.
    feature_cache: Dict[str, Dict[str, Any]] = {}
    records = []
    for row in df.itertuples(index=False):
        prompt = str(row.prompt)[:MAX_TEXT_CHARS]
        key = hashlib.sha1(prompt.encode("utf-8", "replace")).hexdigest()
        feats = feature_cache.get(key)
        if feats is None:
            feats = featurize(prompt)
            feature_cache[key] = feats
        records.append((normalize_tenant(row.tenant_id), prompt, str(row.response)[:MAX_TEXT_CHARS],
                        str(row.route or "default"), feats))

    start = time.time()
    out_rows = []
    aborted = False
    now_base = time.time()
    for threshold in thresholds:
        for guard in (True, False):
            if time.time() - start > time_budget_sec:
                aborted = True
                break
            sim_index = CacheIndex(max_entries=min(MAX_ENTRIES, max(16, rows_used)),
                                   max_text_chars=MAX_TEXT_CHARS, dim=VECTOR_DIM,
                                   name="sweep", record_events=False)
            hits = exact_hits = semantic_hits = borderline = guard_blocks = rejects = 0
            saved_tokens = 0
            saved_cost = 0.0
            latency_total = 0.0
            for tenant, prompt, response, route, feats in records:
                res = sim_index.lookup(tenant, prompt, threshold, features=feats, use_guard=guard,
                                       use_embed=False, record=False, route=route,
                                       skip_expire_sweep=True, now=now_base)
                latency_total += float(res.get("latency_ms", 0.0) or 0.0)
                if res.get("ok") and res.get("hit"):
                    hits += 1
                    if res.get("stage") == "exact":
                        exact_hits += 1
                    else:
                        semantic_hits += 1
                    if res.get("borderline"):
                        borderline += 1
                    tokens = int(res.get("tokens_in", 0) or 0) + int(res.get("tokens_out", 0) or 0)
                    saved_tokens += tokens
                    saved_cost += ((int(res.get("tokens_in", 0) or 0) / 1000.0) * price_in
                                   + (int(res.get("tokens_out", 0) or 0) / 1000.0) * price_out)
                else:
                    if res.get("guard_rejected"):
                        guard_blocks += 1
                    put = sim_index.put(tenant, prompt, response, {
                        "route": route, "features": feats, "allow_empty_response": True, "now": now_base,
                    })
                    if put.get("ok") and not put.get("stored"):
                        rejects += 1
            health = sim_index.health()
            out_rows.append({
                "threshold": threshold,
                "guard": "on" if guard else "off",
                "requests": rows_used,
                "hits": hits,
                "reuse_rate": round(hits / rows_used, 4) if rows_used else 0.0,
                "exact_hits": exact_hits,
                "semantic_hits": semantic_hits,
                "borderline_hits": borderline,
                "guard_blocks": guard_blocks,
                "safety_rejects": rejects,
                "stored_entries": int(health.get("entries", 0)),
                "saved_tokens": int(saved_tokens),
                "saved_cost_usd": round(float(saved_cost), 6),
                "avg_latency_ms": round(latency_total / rows_used, 3) if rows_used else 0.0,
            })
            del sim_index
        if aborted:
            break

    result = pd.DataFrame(out_rows, columns=SWEEP_COLUMNS)
    result.attrs["rows_total"] = total_rows
    result.attrs["rows_used"] = rows_used
    result.attrs["truncated"] = bool(truncated)
    result.attrs["aborted"] = bool(aborted)
    result.attrs["thresholds_requested"] = thresholds
    result.attrs["elapsed_sec"] = round(time.time() - start, 2)
    return result


# =============================================================================
# (K-1) Public API layer - every function returns JSON and never raises
# =============================================================================

def _private_event_kwargs() -> Dict[str, Any]:
    """Guarantees: the correct "keep this event out of the public API" kwarg for the running Gradio version."""
    try:
        import inspect
        params = inspect.signature(gr.Button.click).parameters
        if "api_visibility" in params:      # Gradio 6+
            return {"api_visibility": "private"}
    except Exception:
        pass
    return {"api_name": False}              # Gradio 4/5


PRIVATE_EVENT = _private_event_kwargs()

API_DESCRIPTIONS = OrderedDict([
    ("lookup", "Semantic cache lookup. Returns hit/miss, stage, similarity, age and the cached response."),
    ("store", "Store a prompt/response pair after the safety screen. Returns stored=false plus a reason when refused."),
    ("invalidate", "Invalidate entries of one tenant by entry-id prefix, by route name, or '*' for all."),
    ("stats", "Observability snapshot: hit rate, savings, latency, index health and audit tables."),
    ("sweep", "Replay an uploaded CSV log across thresholds, with and without MismatchGuard."),
    ("health", "Liveness: index size, mode (local-only / local+embedding), uptime and configuration."),
    ("export_index", "Export the in-memory index as JSON (responses optional)."),
    ("import_index", "Import a previously exported index JSON."),
    ("audit", "Borderline hits and MismatchGuard rejections."),
])


@guarded
def api_lookup(tenant_id: str, prompt: str, threshold: float = None, route: str = "default") -> Dict[str, Any]:
    """Guarantees: a complete lookup verdict (hit or an explicit miss reason) without ever raising."""
    result = INDEX.lookup(tenant_id, prompt, threshold, route=route)
    if result.get("ok") and result.get("hit"):
        result["summary"] = "HIT via {0} (similarity {1:.4f}, age {2}s, reused {3}x)".format(
            result.get("stage"), float(result.get("similarity") or 0.0),
            result.get("age_sec"), result.get("hits"))
    elif result.get("ok"):
        reason_text = {
            "empty_index": "no entry stored for this tenant yet",
            "no_candidates": "no fingerprint-similar candidate",
            "below_threshold": "best candidate scored below the threshold",
            "guard_rejected": "candidate was above threshold but MismatchGuard rejected it",
            "empty_prompt": "prompt was empty",
        }.get(result.get("reason"), result.get("reason"))
        result["summary"] = "MISS ({0})".format(reason_text)
    return result


@guarded
def api_store(tenant_id: str, prompt: str, response: str, route: str = "default",
              ttl_sec: int = None) -> Dict[str, Any]:
    """Guarantees: stores only screened content and always reports stored=true/false with a reason code."""
    ttl = _safe_int(ttl_sec, DEFAULT_TTL_SEC) or DEFAULT_TTL_SEC
    result = INDEX.put(tenant_id, prompt, response, {"route": route or "default", "ttl_sec": ttl})
    if result.get("ok") and not result.get("stored"):
        code = result.get("reason", "")
        result["summary"] = "REFUSED: {0} - {1}".format(code, REASON_HELP.get(code, ""))
        OBS.record(tenant_id=normalize_tenant(tenant_id), route=route, event="reject", stage=code,
                   similarity=0.0, latency_ms=0.0, saved_tokens=0, saved_cost=0.0,
                   guard_reasons=[], entry_id=None, prompt=prompt)
        ACC.add(normalize_tenant(tenant_id), route or "default", "reject",
                {"saved_tokens": 0, "saved_cost_usd": 0.0})
    elif result.get("ok"):
        result["summary"] = "STORED entry {0} (ttl {1}s)".format(str(result.get("entry_id"))[:16], result.get("ttl_sec"))
    return result


@guarded
def api_invalidate(tenant_id: str, key_prefix: str) -> Dict[str, Any]:
    """Guarantees: invalidation is scoped to the given tenant only and reports how many entries were removed."""
    return INDEX.invalidate(tenant_id, key_prefix)


@guarded
def api_health() -> Dict[str, Any]:
    """Guarantees: a cheap liveness payload that works even when the index is empty."""
    health = INDEX.health()
    uptime = time.time() - START_TIME
    return ok(
        app=APP_NAME,
        version=APP_VERSION,
        mode=MODE,
        embedding=_EMBEDDER.status(),
        uptime_sec=round(uptime, 1),
        uptime_human="{0}h {1}m".format(int(uptime // 3600), int((uptime % 3600) // 60)),
        entries=health.get("entries", 0),
        tenants=health.get("tenants", 0),
        approx_memory_mb=health.get("approx_memory_mb", 0.0),
        index=health,
        events=OBS.stats(),
        config=config_snapshot(),
        warning="Hugging Face free Spaces have no persistent disk and sleep after 48h of inactivity: "
                "the index lives in RAM only. Use the Backup tab to export before a sleep.",
    )


def _records(df: pd.DataFrame, limit: int = 50) -> List[Dict[str, Any]]:
    """Guarantees: a JSON-safe list of row dicts, bounded in length."""
    try:
        if df is None or df.empty:
            return []
        return _jsonable(df.head(max(1, int(limit))).to_dict(orient="records"))
    except Exception:
        return []


@guarded
def api_stats(tenant_id: str = "", route: str = "") -> Dict[str, Any]:
    """Guarantees: one JSON snapshot containing every dashboard table; safe on an empty buffer."""
    events = _filter_events(OBS.frame(), tenant_id, route)
    served = events[events["event"].isin(["hit", "miss"])] if not events.empty else events
    hits = int((served["event"] == "hit").sum()) if not served.empty else 0
    misses = int((served["event"] == "miss").sum()) if not served.empty else 0
    requests = hits + misses
    stores = int((events["event"] == "store").sum()) if not events.empty else 0
    rejects = int((events["event"] == "reject").sum()) if not events.empty else 0
    saved_tokens = int(events.loc[events["event"] == "hit", "saved_tokens"].sum()) if not events.empty else 0
    saved_cost = float(events.loc[events["event"] == "hit", "saved_cost"].sum()) if not events.empty else 0.0
    acc = ACC.frame()
    # The Dashboard renders these next to the tenant/route filter boxes, so they
    # must honour the same filter - otherwise one tenant sees another's savings.
    if not acc.empty and tenant_id:
        acc = acc[acc["tenant_id"] == normalize_tenant(tenant_id)]
    if not acc.empty and route:
        acc = acc[acc["route"] == str(route)]
    per_tenant = _jsonable(INDEX.health().get("per_tenant", []))
    if tenant_id:
        per_tenant = [t for t in per_tenant if t.get("tenant_id") == normalize_tenant(tenant_id)]
    return ok(
        generated_at=time.time(),
        mode=MODE,
        filters={"tenant_id": normalize_tenant(tenant_id) if tenant_id else "", "route": route or ""},
        summary={
            "requests": requests,
            "hits": hits,
            "misses": misses,
            "hit_rate": round(hits / requests, 4) if requests else 0.0,
            "stores": stores,
            "rejects": rejects,
            "saved_tokens": saved_tokens,
            "saved_cost_usd": round(saved_cost, 6),
            "price_in_per_1k": PRICE_IN_PER_1K,
            "price_out_per_1k": PRICE_OUT_PER_1K,
        },
        index=INDEX.health(),
        events_buffer=OBS.stats(),
        embedding=_EMBEDDER.status(),
        tables={
            "hit_rate": _records(hit_rate_timeseries(events), 120),
            "savings_by_tenant": _records(savings_cumulative(events, "tenant_id"), 120),
            "savings_by_route": _records(savings_cumulative(events, "route"), 120),
            "latency": _records(latency_distribution(events), 10),
            "top_entries": _records(INDEX.top_entries(20, tenant_id), 20),
            "borderline": _records(borderline_events(events, 100), 100),
            "guard_rejections": _records(guard_rejection_events(events, 100), 100),
            "accounting": _records(acc, 100),
            "per_tenant": per_tenant[:50],
        },
        notes="saved_cost_usd is an ESTIMATE from PRICE_IN_PER_1K / PRICE_OUT_PER_1K and a local token "
              "approximation. It is not a bill.",
    )


@guarded
def api_audit(tenant_id: str = "", limit: int = 100) -> Dict[str, Any]:
    """Guarantees: the two audit tables (borderline hits, guard rejections) as JSON."""
    events = _filter_events(OBS.frame(), tenant_id, "")
    return ok(
        borderline_margin=BORDERLINE_MARGIN,
        borderline=_records(borderline_events(events, limit), limit),
        guard_rejections=_records(guard_rejection_events(events, limit), limit),
        guard_checks_enabled=[c for c in GUARD_CHECKS if c not in GUARD_DISABLE],
        guard_checks_disabled=sorted(GUARD_DISABLE),
    )


@guarded
def api_export_index(include_responses: bool = True, tenant_id: str = "") -> Dict[str, Any]:
    """Guarantees: a JSON snapshot payload plus its entry count; responses are opt-out."""
    payload = INDEX.export_json(bool(include_responses), tenant_id or None)
    count = sum(len(t.get("entries", [])) for t in payload.get("tenants", {}).values())
    return ok(entries=count, include_responses=bool(include_responses), payload=payload)


@guarded
def api_import_index(payload: Any, skip_safety: bool = False) -> Dict[str, Any]:
    """Guarantees: imports a snapshot (string, bytes or object) and reports exactly what happened to each row."""
    if isinstance(payload, dict) and "payload" in payload and "tenants" not in payload:
        payload = payload["payload"]
    return INDEX.import_json(payload, skip_safety=skip_safety)


def _read_table(path: str) -> pd.DataFrame:
    """Guarantees: reads a CSV in UTF-8/UTF-8-SIG/CP932 and raises a clear ValueError when unreadable."""
    last_error = None
    for encoding in ("utf-8-sig", "utf-8", "cp932", "shift_jis", "latin-1"):
        try:
            return pd.read_csv(path, encoding=encoding)
        except UnicodeDecodeError as exc:
            last_error = exc
            continue
        except Exception as exc:
            raise ValueError("could not parse CSV: {0}".format(exc))
    raise ValueError("could not decode CSV: {0}".format(last_error))


def _parse_thresholds(text: str) -> List[float]:
    """Guarantees: parses a comma/space separated threshold list, falling back to the documented default."""
    values = []
    for token in re.split(r"[,\s;]+", str(text or "")):
        if not token:
            continue
        try:
            value = float(token)
            if 0.0 <= value <= 1.0:
                values.append(round(value, 4))
        except Exception:
            continue
    return sorted(set(values)) or list(DEFAULT_SWEEP_THRESHOLDS)


@guarded
def api_sweep(csv_path: Any, thresholds_text: str = "", price_in: float = None, price_out: float = None,
              max_rows: int = SWEEP_MAX_ROWS) -> Dict[str, Any]:
    """Guarantees: threshold-by-threshold reuse and savings estimates computed on a throwaway index."""
    if not csv_path:
        return err(ErrorCode.NO_FILE, "upload a CSV with at least a 'prompt' column")
    path = csv_path if isinstance(csv_path, str) else getattr(csv_path, "name", None)
    if not path or not os.path.exists(path):
        return err(ErrorCode.NO_FILE, "uploaded file is not readable")
    try:
        logs = _read_table(path)
    except ValueError as exc:
        return err(ErrorCode.BAD_INPUT, str(exc))
    try:
        table = sweep(logs, _parse_thresholds(thresholds_text),
                      _safe_float(price_in, PRICE_IN_PER_1K),
                      _safe_float(price_out, PRICE_OUT_PER_1K),
                      max_rows=_safe_int(max_rows, SWEEP_MAX_ROWS) or SWEEP_MAX_ROWS)
    except ValueError as exc:
        return err(ErrorCode.BAD_INPUT, str(exc))
    if table.empty:
        return err(ErrorCode.BAD_INPUT, "no usable rows in the uploaded log")

    guarded_rows = table[table["guard"] == "on"]
    best = None
    if not guarded_rows.empty:
        row = guarded_rows.sort_values(["saved_cost_usd", "reuse_rate", "threshold"],
                                       ascending=[False, False, False]).iloc[0]
        conservative = guarded_rows[guarded_rows["threshold"] >= DEFAULT_THRESHOLD]
        safe_row = conservative.iloc[0] if not conservative.empty else row
        best = {
            "max_saving": _jsonable(row.to_dict()),
            "recommended_conservative": _jsonable(safe_row.to_dict()),
        }
    return ok(
        rows_total=int(table.attrs.get("rows_total", 0)),
        rows_used=int(table.attrs.get("rows_used", 0)),
        truncated=bool(table.attrs.get("truncated", False)),
        aborted_on_time_budget=bool(table.attrs.get("aborted", False)),
        elapsed_sec=table.attrs.get("elapsed_sec", 0.0),
        price_in_per_1k=_safe_float(price_in, PRICE_IN_PER_1K),
        price_out_per_1k=_safe_float(price_out, PRICE_OUT_PER_1K),
        table=_records(table, 200),
        best=best,
        notes="'guard=on' is MismatchGuard enabled (production setting). The gap between guard on/off is the "
              "reuse you are deliberately giving up to avoid wrong answers. Savings are estimates, not a bill.",
    )


# =============================================================================
# (K-2) Gradio UI
# =============================================================================

SLEEP_WARNING = (
    "### ⚠️ Free-tier reality check\n"
    "Hugging Face free Spaces have **no persistent disk** and **sleep after 48h of inactivity** "
    "(the sleep timer cannot be changed on the free tier). The index lives **in RAM only**: a sleep, "
    "a restart or a new commit wipes it. Export from this tab before you go idle, and import after wake-up."
)


def _df_or_empty(records: Any, columns: Sequence[str]) -> pd.DataFrame:
    """Guarantees: a DataFrame with the requested columns whatever the input shape."""
    try:
        if not records:
            return pd.DataFrame(columns=list(columns))
        df = pd.DataFrame(records)
        for col in columns:
            if col not in df.columns:
                df[col] = None
        return df[list(columns)]
    except Exception:
        return pd.DataFrame(columns=list(columns))


def ui_lookup(tenant_id: str, prompt: str, threshold: Any, route: str) -> Dict[str, Any]:
    """Guarantees: the Lookup tab always renders a JSON verdict, even for malformed or non-numeric input."""
    return _jsonable(api_lookup(tenant_id, prompt, _safe_float(threshold, DEFAULT_THRESHOLD), route))


def ui_store(tenant_id: str, prompt: str, response: str, route: str, ttl_sec: Any) -> Dict[str, Any]:
    """Guarantees: the Store tab always renders a JSON result, even when ttl_sec is NaN, inf or not a number."""
    return _jsonable(api_store(tenant_id, prompt, response, route, _safe_int(ttl_sec, DEFAULT_TTL_SEC)))


def ui_invalidate(tenant_id: str, key_prefix: str) -> Dict[str, Any]:
    """Guarantees: the Invalidate tab always renders how many entries were removed."""
    return _jsonable(api_invalidate(tenant_id, key_prefix))


def ui_health() -> Dict[str, Any]:
    """Guarantees: the Health tab always renders, even when the index is empty."""
    return _jsonable(api_health())


def ui_stats(tenant_id: str, route: str) -> Dict[str, Any]:
    """Guarantees: the dashboard JSON payload that every dashboard widget is rendered from."""
    return _jsonable(api_stats(tenant_id, route))


def render_dashboard(stats: Any):
    """Guarantees: renders the dashboard widgets from an already-computed stats payload (no recomputation)."""
    try:
        stats = stats if isinstance(stats, dict) else {}
        tables = stats.get("tables", {}) if stats.get("ok", True) else {}
        summary = stats.get("summary", {})

        hit_rows = tables.get("hit_rate", [])
        if hit_rows:
            labels = [r.get("bucket", "") for r in hit_rows]
            hit_chart = svg_line_chart(
                labels,
                {"hit rate": [float(r.get("hit_rate", 0) or 0) for r in hit_rows],
                 "moving avg": [float(r.get("hit_rate_ma", 0) or 0) for r in hit_rows]},
                "Hit rate per minute (with moving average)")
        else:
            hit_chart = _empty_chart("Hit rate per minute", "no hit/miss events yet - run a Lookup first")

        save_rows = tables.get("savings_by_tenant", [])
        if save_rows:
            by_tenant: Dict[str, List[float]] = {}
            buckets = sorted({r.get("bucket", "") for r in save_rows})
            index = {b: i for i, b in enumerate(buckets)}
            for row in save_rows:
                name = str(row.get("tenant_id", "default"))
                series = by_tenant.setdefault(name, [0.0] * len(buckets))
                series[index.get(row.get("bucket", ""), 0)] = float(row.get("cum_saved_tokens", 0) or 0)
            for series in by_tenant.values():
                running = 0.0
                for i, value in enumerate(series):
                    running = max(running, value)
                    series[i] = running
            save_chart = svg_line_chart(buckets, by_tenant, "Cumulative saved tokens by tenant")
        else:
            save_chart = _empty_chart("Cumulative saved tokens by tenant", "no cache hits recorded yet")

        latency = _df_or_empty(tables.get("latency", []),
                               ["event", "count", "mean_ms", "p50_ms", "p90_ms", "p99_ms", "max_ms"])
        if not latency.empty:
            lat_chart = svg_bar_chart(latency["event"].tolist(),
                                      [float(v) for v in latency["p90_ms"].tolist()],
                                      "p90 latency by outcome", " ms")
        else:
            lat_chart = _empty_chart("p90 latency by outcome", "no served requests yet")

        health = stats.get("index", {})
        totals = health.get("totals", {})
        health_df = pd.DataFrame([{
            "entries": health.get("entries", 0),
            "tenants": health.get("tenants", 0),
            "capacity_per_tenant": health.get("capacity_per_tenant", MAX_ENTRIES),
            "approx_memory_mb": health.get("approx_memory_mb", 0.0),
            "evicted": totals.get("evicted", 0),
            "expired": totals.get("expired", 0),
            "invalidated": totals.get("invalidated", 0),
            "rejected_by_safety": totals.get("rejected", 0),
            "events_buffered": stats.get("events_buffer", {}).get("buffered", 0),
            "events_dropped": stats.get("events_buffer", {}).get("dropped", 0),
        }])
        summary_md = (
            "**requests** {requests} | **hits** {hits} | **misses** {misses} | "
            "**hit rate** {rate:.1%} | **saved tokens** {tokens:,} | **estimated saving** ${cost:,.4f}"
        ).format(requests=summary.get("requests", 0), hits=summary.get("hits", 0),
                 misses=summary.get("misses", 0), rate=float(summary.get("hit_rate", 0.0) or 0.0),
                 tokens=int(summary.get("saved_tokens", 0) or 0),
                 cost=float(summary.get("saved_cost_usd", 0.0) or 0.0))
        if not (PRICE_IN_PER_1K or PRICE_OUT_PER_1K):
            summary_md += "\n\n> Prices are 0.00, so the USD column stays at zero. Set `PRICE_IN_PER_1K` / `PRICE_OUT_PER_1K`."

        return (
            summary_md,
            hit_chart,
            save_chart,
            lat_chart,
            _df_or_empty(tables.get("accounting", []),
                         ["tenant_id", "route", "hits", "misses", "stores", "rejects", "hit_rate",
                          "saved_tokens", "saved_cost_usd"]),
            latency,
            _df_or_empty(tables.get("top_entries", []),
                         ["tenant_id", "entry_id", "route", "hits", "age_sec", "ttl_remaining_sec",
                          "tokens_in", "tokens_out", "prompt_preview"]),
            _df_or_empty(tables.get("borderline", []),
                         ["datetime", "tenant_id", "route", "event", "stage", "similarity", "entry_id",
                          "guard_reasons"]),
            health_df,
        )
    except Exception as exc:
        empty = pd.DataFrame()
        message = _empty_chart("dashboard", "render failed: {0}".format(exc))
        return ("render failed: {0}".format(exc), message, message, message, empty, empty, empty, empty, empty)


def ui_export_events_csv(tenant_id: str, route: str) -> Optional[str]:
    """Guarantees: writes the filtered event buffer to a CSV file and returns its path (None when empty)."""
    try:
        events = _filter_events(OBS.frame(), tenant_id, route)
        buffer = io.StringIO()
        events.to_csv(buffer, index=False)
        return write_temp_file(buffer.getvalue(), "echocache-events", ".csv")
    except Exception:
        return None


def ui_audit(tenant_id: str):
    """Guarantees: the audit tables always render, with the guard configuration shown alongside."""
    try:
        payload = api_audit(tenant_id, 200)
        borderline = _df_or_empty(payload.get("borderline", []),
                                  ["datetime", "tenant_id", "route", "event", "stage", "similarity",
                                   "entry_id", "guard_reasons"])
        rejections = _df_or_empty(payload.get("guard_rejections", []),
                                  ["datetime", "tenant_id", "route", "stage", "similarity",
                                   "guard_reasons", "prompt_prefix"])
        info = (
            "Borderline window: similarity within **±{margin}** of the threshold. "
            "Guard checks enabled: `{on}`. Disabled via `GUARD_DISABLE`: `{off}`.\n\n"
            "A row in *Guard rejections* means the candidate was similar enough **but its meaning differed** "
            "(negation, numbers, proper nouns, time anchors or question type). Those are wrong answers avoided."
        ).format(margin=BORDERLINE_MARGIN,
                 on=", ".join(payload.get("guard_checks_enabled", [])) or "-",
                 off=", ".join(payload.get("guard_checks_disabled", [])) or "-")
        return info, borderline, rejections
    except Exception as exc:
        return "audit failed: {0}".format(exc), pd.DataFrame(), pd.DataFrame()


def ui_export_index(include_responses: bool, tenant_id: str):
    """Guarantees: produces a downloadable JSON snapshot plus a short receipt, never raising."""
    try:
        result = api_export_index(bool(include_responses), tenant_id)
        if not result.get("ok"):
            return None, _jsonable(result)
        payload = result.get("payload", {})
        text = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
        path = write_temp_file(text, "echocache-index", ".json")
        receipt = {"ok": True, "entries": result.get("entries", 0),
                   "include_responses": result.get("include_responses"),
                   "bytes": len(text.encode("utf-8")), "file": os.path.basename(path or "")}
        return path, receipt
    except Exception as exc:
        return None, {"ok": False, "error": {"code": ErrorCode.INTERNAL.value, "message": str(exc)}}


def ui_import_index(file_path: Any, pasted: str, skip_safety: bool) -> Dict[str, Any]:
    """Guarantees: imports from an uploaded file or a pasted blob and reports per-row outcomes."""
    try:
        text = None
        if file_path:
            path = file_path if isinstance(file_path, str) else getattr(file_path, "name", None)
            if path and os.path.exists(path):
                if os.path.getsize(path) > IMPORT_MAX_BYTES:
                    return _jsonable(err(ErrorCode.PAYLOAD_TOO_LARGE, "file exceeds IMPORT_MAX_BYTES"))
                with open(path, "r", encoding="utf-8", errors="replace") as handle:
                    text = handle.read()
        if not text and pasted and pasted.strip():
            text = pasted
        if not text:
            return _jsonable(err(ErrorCode.NO_FILE, "upload a file or paste the JSON snapshot"))
        return _jsonable(api_import_index(text, bool(skip_safety)))
    except Exception as exc:
        return _jsonable(err(ErrorCode.INTERNAL, str(exc)))


def render_sweep(payload: Any):
    """Guarantees: renders the sweep table, charts and the plain-language recommendation from one payload."""
    empty = pd.DataFrame(columns=SWEEP_COLUMNS)
    try:
        payload = payload if isinstance(payload, dict) else {}
        if not payload.get("ok"):
            message = (payload.get("error") or {}).get("message", "upload a CSV to run a sweep")
            return empty, _empty_chart("Reuse rate by threshold", str(message)), "**{0}**".format(message)
        table = _df_or_empty(payload.get("table", []), SWEEP_COLUMNS)
        if table.empty:
            return empty, _empty_chart("Reuse rate by threshold", "no rows"), "**no usable rows**"

        on = table[table["guard"] == "on"].sort_values("threshold")
        off = table[table["guard"] == "off"].sort_values("threshold")
        labels = ["{0:.2f}".format(float(t)) for t in on["threshold"].tolist()]
        chart = svg_line_chart(
            labels,
            {"guard ON (prod)": [float(v) for v in on["reuse_rate"].tolist()],
             "guard OFF": [float(v) for v in off["reuse_rate"].tolist()]},
            "Reuse rate by threshold (MismatchGuard on vs off)")

        best = payload.get("best") or {}
        rec = best.get("recommended_conservative") or {}
        top = best.get("max_saving") or {}
        rows_used = payload.get("rows_used", 0)
        lines = [
            "## What this log would have saved",
            "",
            "Replayed **{0}** rows{1}.".format(
                rows_used,
                " (truncated from {0}; raise SWEEP_MAX_ROWS to replay more)".format(payload.get("rows_total"))
                if payload.get("truncated") else ""),
            "",
        ]
        if payload.get("aborted_on_time_budget"):
            lines.append("> ⏱️ The sweep hit its time budget, so some thresholds were **not** evaluated.")
            lines.append("")
        if rec:
            lines += [
                "**Conservative setting (recommended): threshold {thr:.2f}, MismatchGuard ON**".format(
                    thr=float(rec.get("threshold", DEFAULT_THRESHOLD))),
                "",
                "- reuse rate: **{0:.1%}** ({1} of {2} requests served from cache)".format(
                    float(rec.get("reuse_rate", 0)), int(rec.get("hits", 0)), int(rec.get("requests", 0))),
                "- estimated saving: **${0:,.4f}** ({1:,} tokens)".format(
                    float(rec.get("saved_cost_usd", 0.0)), int(rec.get("saved_tokens", 0))),
                "- borderline hits: {0} | guard blocks: {1} | safety rejects: {2}".format(
                    int(rec.get("borderline_hits", 0)), int(rec.get("guard_blocks", 0)),
                    int(rec.get("safety_rejects", 0))),
                "",
            ]
        if top and rec and float(top.get("threshold", 0)) != float(rec.get("threshold", 0)):
            lines += [
                "**Highest-saving setting in this sweep: threshold {0:.2f}** -> reuse {1:.1%}, ${2:,.4f}. "
                "Lower thresholds reuse more but rely entirely on MismatchGuard to stay correct.".format(
                    float(top.get("threshold", 0)), float(top.get("reuse_rate", 0)),
                    float(top.get("saved_cost_usd", 0.0))),
                "",
            ]
        if not on.empty and not off.empty:
            delta = float(off["reuse_rate"].mean() - on["reuse_rate"].mean())
            lines.append("**Cost of safety:** MismatchGuard gives up about **{0:.1%}** of reuse on average "
                         "in exchange for blocking meaning-changed hits.".format(max(0.0, delta)))
            lines.append("")
        lines.append("_Savings are estimates from local token counts and the prices you entered - not a bill._")
        return table, chart, "\n".join(lines)
    except Exception as exc:
        return empty, _empty_chart("sweep", "render failed"), "**render failed:** {0}".format(exc)


def ui_sample_csv() -> Optional[str]:
    """Guarantees: writes a tiny demo CSV (including guard traps) so the Sweep tab can be tried immediately."""
    rows = [
        ("acme", "How do I reset my password?", "Open Settings > Security > Reset password.", "support"),
        ("acme", "how do i reset my password", "Open Settings > Security > Reset password.", "support"),
        ("acme", "Hi team, how do I reset my password? Thanks!", "Open Settings > Security > Reset password.", "support"),
        ("acme", "Why do I need to reset my password?", "Because the policy rotates credentials every 90 days.", "support"),
        ("acme", "How do I reset my password on 2 devices?", "Repeat the reset flow on each device.", "support"),
        ("acme", "Can I cancel my subscription?", "Yes, from Billing > Cancel.", "billing"),
        ("acme", "Can I not cancel my subscription?", "You can cancel at any time; only annual plans are prorated.", "billing"),
        ("globex", "What is the 2024 tax rate?", "The 2024 corporate rate is 21%.", "finance"),
        ("globex", "What is the 2025 tax rate?", "The 2025 corporate rate is 21% with a new surtax.", "finance"),
        ("globex", "what is the 2025 tax rate", "The 2025 corporate rate is 21% with a new surtax.", "finance"),
    ]
    buffer = io.StringIO()
    writer = csv.writer(buffer)
    writer.writerow(["tenant_id", "prompt", "response", "route"])
    for row in rows:
        writer.writerow(row)
    return write_temp_file(buffer.getvalue(), "echocache-sample-log", ".csv")


def ui_seed_demo(tenant_id: str) -> Dict[str, Any]:
    """Guarantees: stores a handful of demo entries so a fresh Space can be verified in one click."""
    seeds = [
        ("How do I reset my password?", "Open Settings > Security > Reset password.", "support"),
        ("Can I cancel my subscription?", "Yes, from Billing > Cancel.", "billing"),
        ("What is the 2025 tax rate?", "The 2025 corporate rate is 21% with a new surtax.", "finance"),
    ]
    results = []
    for prompt, response, route in seeds:
        results.append(_jsonable(api_store(tenant_id, prompt, response, route, DEFAULT_TTL_SEC)))
    return {"ok": True, "seeded": len(results), "results": results,
            "next": "Try Lookup with 'hi team, how do i reset my password? thanks' - it should HIT."}


# =============================================================================
# (K-9) API documentation, assembled at runtime (paths are NEVER hardcoded)
# =============================================================================


def _api_prefix() -> str:
    """Guarantees: the API path prefix of the *running* Gradio version, discovered at runtime."""
    try:
        from gradio import route_utils  # type: ignore
        prefix = getattr(route_utils, "API_PREFIX", None)
        if isinstance(prefix, str):
            return prefix
    except Exception:
        pass
    try:
        if int(str(getattr(gr, "__version__", "0")).split(".")[0]) >= 5:
            return "/gradio_api"
    except Exception:
        pass
    return ""


def _root_url(request: Any = None) -> str:
    """Guarantees: the externally reachable base URL, derived from the request or SPACE_ID, never hardcoded."""
    try:
        headers = getattr(request, "headers", None) or {}
        get = headers.get if hasattr(headers, "get") else (lambda *_: None)
        host = get("x-forwarded-host") or get("host")
        if host:
            hostname = str(host).split(":")[0].lower()
            local = (hostname in ("localhost", "127.0.0.1", "0.0.0.0", "::1", "[::1]")
                     or hostname.endswith(".local") or hostname.startswith("192.168.")
                     or hostname.startswith("10.") or "." not in hostname)
            proto = get("x-forwarded-proto") or ("http" if local else "https")
            return "{0}://{1}".format(proto, host)
    except Exception:
        pass
    space = _env_str("SPACE_HOST")
    if space:
        return "https://{0}".format(space)
    space_id = _env_str("SPACE_ID")
    if space_id and "/" in space_id:
        owner, name = space_id.split("/", 1)
        slug = re.sub(r"[^a-zA-Z0-9]+", "-", "{0}-{1}".format(owner, name)).strip("-").lower()
        return "https://{0}.hf.space".format(slug)
    port = _env_str("GRADIO_SERVER_PORT", "7860")
    return "http://localhost:{0}".format(port)


def _discover_api_names() -> List[str]:
    """Guarantees: the live endpoint list taken from the running app, falling back to the documented names."""
    try:
        info = demo.get_api_info()  # type: ignore[name-defined]
        named = (info or {}).get("named_endpoints", {})
        names = [str(k).lstrip("/") for k in named.keys()]
        if names:
            return names
    except Exception:
        pass
    return list(API_DESCRIPTIONS.keys())


def ui_api_docs(request: gr.Request = None) -> str:
    """Guarantees: API docs whose paths are assembled from the running server, so a Gradio upgrade cannot stale them."""
    try:
        root = _root_url(request)
        prefix = _api_prefix()
        names = _discover_api_names()
        client_target = _env_str("SPACE_ID") or root
        rows = ["| endpoint | POST path | what it does |", "|---|---|---|"]
        for name in names:
            rows.append("| `/{0}` | `{1}{2}/call/{0}` | {3} |".format(
                name, root, prefix, API_DESCRIPTIONS.get(name, "-")))
        table = "\n".join(rows)
        return """## API (paths built at runtime - do not copy them from another Space)

Detected Gradio **{version}**, API prefix **`{prefix}`**, base URL **`{root}`**.

{table}

### curl (two steps: POST returns an EVENT_ID, GET streams the result)

```bash
EVENT_ID=$(curl -s -X POST "{root}{prefix}/call/lookup" \\
  -H "Content-Type: application/json" \\
  -d '{{"data": ["default", "How do I reset my password?", {threshold}, "support"]}}' \\
  | python3 -c "import sys,json; print(json.load(sys.stdin)['event_id'])")

curl -s -N "{root}{prefix}/call/lookup/$EVENT_ID"
```

```bash
# store
curl -s -X POST "{root}{prefix}/call/store" -H "Content-Type: application/json" \\
  -d '{{"data": ["default", "How do I reset my password?", "Open Settings > Security > Reset password.", "support", {ttl}]}}'
```

### gradio_client (recommended - it resolves the paths for you)

```python
from gradio_client import Client

client = Client("{client_target}")            # or the full https URL of this Space
print(client.predict("default", "How do I reset my password?", {threshold}, "support", api_name="/lookup"))
print(client.predict("default", "How do I reset my password?",
                     "Open Settings > Security > Reset password.", "support", {ttl}, api_name="/store"))
print(client.predict("", "", api_name="/stats"))
print(client.predict(api_name="/health"))
```

### Argument order

| endpoint | arguments |
|---|---|
| `/lookup` | `tenant_id, prompt, threshold, route` |
| `/store` | `tenant_id, prompt, response, route, ttl_sec` |
| `/invalidate` | `tenant_id, key_prefix` (`"*"` = whole tenant) |
| `/stats` | `tenant_id, route` (empty strings = no filter) |
| `/sweep` | `csv_file, thresholds, price_in, price_out, max_rows` |
| `/audit` | `tenant_id` |
| `/export_index` | `include_responses, tenant_id` |
| `/import_index` | `file, pasted_json, skip_safety` |
| `/health` | *(no arguments)* |

> The authoritative, always-current list is the **"Use via API"** link at the very bottom of this page
> (footer). It is generated by Gradio itself and reflects this exact build.
""".format(version=getattr(gr, "__version__", "unknown"), prefix=prefix, root=root, table=table,
           threshold=DEFAULT_THRESHOLD, ttl=DEFAULT_TTL_SEC, client_target=client_target)
    except Exception as exc:
        return "API docs failed to render: {0}\n\nUse the **Use via API** link in the page footer.".format(exc)


# =============================================================================
# (K) Blocks
# =============================================================================

_EMPTY_ACC = pd.DataFrame(columns=["tenant_id", "route", "hits", "misses", "stores", "rejects",
                                   "hit_rate", "saved_tokens", "saved_cost_usd"])
_EMPTY_LAT = pd.DataFrame(columns=["event", "count", "mean_ms", "p50_ms", "p90_ms", "p99_ms", "max_ms"])
_EMPTY_TOP = pd.DataFrame(columns=["tenant_id", "entry_id", "route", "hits", "age_sec", "ttl_remaining_sec",
                                   "tokens_in", "tokens_out", "prompt_preview"])
_EMPTY_BORDER = pd.DataFrame(columns=["datetime", "tenant_id", "route", "event", "stage", "similarity",
                                      "entry_id", "guard_reasons"])
_EMPTY_GUARD = pd.DataFrame(columns=["datetime", "tenant_id", "route", "stage", "similarity",
                                     "guard_reasons", "prompt_prefix"])
_EMPTY_HEALTH = pd.DataFrame(columns=["entries", "tenants", "capacity_per_tenant", "approx_memory_mb",
                                      "evicted", "expired", "invalidated", "rejected_by_safety",
                                      "events_buffered", "events_dropped"])

with gr.Blocks(title="EchoCache", analytics_enabled=False) as demo:
    gr.Markdown(
        "# ♻️ EchoCache\n"
        "**Semantic cache + cost observability for LLM APIs.** "
        "Mode: **{mode}** {mode_note} · index cap **{cap:,} entries/tenant** · "
        "default threshold **{thr}** · TTL **{ttl}s**".format(
            mode=MODE,
            mode_note="(HF_TOKEN not set - similarity is 100% local, which is the supported production mode)"
            if not EMBED_ENABLED else "(embeddings only re-rank; local similarity still decides everything)",
            cap=MAX_ENTRIES, thr=DEFAULT_THRESHOLD, ttl=DEFAULT_TTL_SEC)
    )

    with gr.Tabs():
        # ------------------------------------------------------------------ 1
        with gr.Tab("Lookup"):
            with gr.Row():
                lk_tenant = gr.Textbox(label="tenant_id", value="default", max_lines=1)
                lk_route = gr.Textbox(label="route", value="default", max_lines=1)
                lk_threshold = gr.Slider(label="threshold", minimum=0.50, maximum=1.0, step=0.005,
                                         value=DEFAULT_THRESHOLD)
            lk_prompt = gr.Textbox(label="prompt", lines=6, placeholder="Ask the same question two different ways...")
            lk_btn = gr.Button("Lookup", variant="primary")
            lk_out = gr.JSON(label="result")
            gr.Markdown(
                "A **hit** returns the cached response, the matching stage (`exact` / `cosine` / `embedding`), "
                "the similarity and the entry age. A **miss** always states why: `empty_index`, `no_candidates`, "
                "`below_threshold` or `guard_rejected`."
            )
            lk_btn.click(ui_lookup, [lk_tenant, lk_prompt, lk_threshold, lk_route], lk_out, api_name="lookup")

        # ------------------------------------------------------------------ 2
        with gr.Tab("Store"):
            with gr.Row():
                st_tenant = gr.Textbox(label="tenant_id", value="default", max_lines=1)
                st_route = gr.Textbox(label="route", value="default", max_lines=1)
                st_ttl = gr.Number(label="ttl_sec", value=DEFAULT_TTL_SEC, precision=0)
            st_prompt = gr.Textbox(label="prompt", lines=4)
            st_response = gr.Textbox(label="response", lines=6)
            with gr.Row():
                st_btn = gr.Button("Store", variant="primary")
                st_seed = gr.Button("Seed 3 demo entries")
            st_out = gr.JSON(label="result")
            gr.Markdown(
                "Content is screened **before** storage. Refusals return `stored=false` with a reason code: "
                "`credit_card_luhn`, `credential_prefix`, `jwt_structure`, `high_entropy_secret`, "
                "`contact_pii_excess`, `private_key_block`."
            )
            st_btn.click(ui_store, [st_tenant, st_prompt, st_response, st_route, st_ttl], st_out, api_name="store")
            st_seed.click(ui_seed_demo, [st_tenant], st_out, **PRIVATE_EVENT)

        # ------------------------------------------------------------------ 3
        with gr.Tab("Invalidate"):
            with gr.Row():
                iv_tenant = gr.Textbox(label="tenant_id", value="default", max_lines=1)
                iv_prefix = gr.Textbox(label="key prefix / route / *", value="", max_lines=1,
                                       placeholder='entry-id prefix, a route name, or "*" for the whole tenant')
            iv_btn = gr.Button("Invalidate", variant="stop")
            iv_out = gr.JSON(label="result")
            gr.Markdown("Invalidation is **scoped to one tenant**. Entry ids are shown in Dashboard → *most reused*.")
            iv_btn.click(ui_invalidate, [iv_tenant, iv_prefix], iv_out, api_name="invalidate")

        # ------------------------------------------------------------------ 4
        with gr.Tab("Dashboard"):
            with gr.Row():
                db_tenant = gr.Textbox(label="filter tenant_id (blank = all)", value="", max_lines=1)
                db_route = gr.Textbox(label="filter route (blank = all)", value="", max_lines=1)
                db_btn = gr.Button("Refresh", variant="primary")
            db_summary = gr.Markdown("Press **Refresh**.")
            db_hit_chart = gr.HTML()
            db_save_chart = gr.HTML()
            db_lat_chart = gr.HTML()
            with gr.Row():
                db_acc = gr.Dataframe(value=_EMPTY_ACC, label="savings by tenant / route", interactive=False, wrap=True)
                db_lat = gr.Dataframe(value=_EMPTY_LAT, label="latency distribution", interactive=False, wrap=True)
            db_top = gr.Dataframe(value=_EMPTY_TOP, label="most reused entries", interactive=False, wrap=True)
            db_border = gr.Dataframe(value=_EMPTY_BORDER, label="borderline decisions (threshold ± margin)",
                                     interactive=False, wrap=True)
            db_health = gr.Dataframe(value=_EMPTY_HEALTH, label="index health", interactive=False, wrap=True)
            with gr.Row():
                db_csv_btn = gr.Button("Export events CSV")
                db_csv = gr.File(label="events.csv", interactive=False)
            with gr.Accordion("raw stats JSON", open=False):
                db_json = gr.JSON(label="stats")
            db_btn.click(ui_stats, [db_tenant, db_route], db_json, api_name="stats").then(
                render_dashboard, db_json,
                [db_summary, db_hit_chart, db_save_chart, db_lat_chart, db_acc, db_lat, db_top,
                 db_border, db_health], **PRIVATE_EVENT)
            db_csv_btn.click(ui_export_events_csv, [db_tenant, db_route], db_csv, **PRIVATE_EVENT)

        # ------------------------------------------------------------------ 5
        with gr.Tab("Sweep"):
            gr.Markdown(
                "Upload a request log (**CSV**: `tenant_id`, `prompt`, optional `response`, optional `route`) "
                "and see, per threshold, **how much you would have reused and how much you would have saved** - "
                "with MismatchGuard **on** (production) and **off** (upper bound)."
            )
            with gr.Row():
                sw_file = gr.File(label="request log CSV", file_types=[".csv"], type="filepath")
                with gr.Column():
                    sw_thresholds = gr.Textbox(label="thresholds",
                                               value=", ".join(str(t) for t in DEFAULT_SWEEP_THRESHOLDS))
                    with gr.Row():
                        sw_price_in = gr.Number(label="price in / 1k tokens (USD)", value=PRICE_IN_PER_1K)
                        sw_price_out = gr.Number(label="price out / 1k tokens (USD)", value=PRICE_OUT_PER_1K)
                    sw_max_rows = gr.Number(label="max rows", value=SWEEP_MAX_ROWS, precision=0)
            with gr.Row():
                sw_btn = gr.Button("Run sweep", variant="primary")
                sw_sample_btn = gr.Button("Download a sample log")
            sw_sample = gr.File(label="sample CSV", interactive=False)
            sw_md = gr.Markdown()
            sw_chart = gr.HTML()
            sw_table = gr.Dataframe(value=pd.DataFrame(columns=SWEEP_COLUMNS), label="sweep result",
                                    interactive=False, wrap=True)
            with gr.Accordion("raw sweep JSON", open=False):
                sw_json = gr.JSON(label="sweep")
            sw_btn.click(api_sweep, [sw_file, sw_thresholds, sw_price_in, sw_price_out, sw_max_rows],
                         sw_json, api_name="sweep").then(
                render_sweep, sw_json, [sw_table, sw_chart, sw_md], **PRIVATE_EVENT)
            sw_sample_btn.click(ui_sample_csv, None, sw_sample, **PRIVATE_EVENT)

        # ------------------------------------------------------------------ 6
        with gr.Tab("Audit"):
            with gr.Row():
                au_tenant = gr.Textbox(label="filter tenant_id (blank = all)", value="", max_lines=1)
                au_btn = gr.Button("Refresh", variant="primary")
            au_info = gr.Markdown()
            au_border = gr.Dataframe(value=_EMPTY_BORDER, label="borderline decisions", interactive=False, wrap=True)
            au_guard = gr.Dataframe(value=_EMPTY_GUARD, label="MismatchGuard rejections", interactive=False, wrap=True)
            gr.Markdown("#### Saw a wrong hit? Raise the threshold and invalidate the entry here.")
            with gr.Row():
                au_inv_tenant = gr.Textbox(label="tenant_id", value="default", max_lines=1)
                au_inv_prefix = gr.Textbox(label="entry id prefix", value="", max_lines=1)
                au_inv_btn = gr.Button("Invalidate entry", variant="stop")
            au_inv_out = gr.JSON(label="invalidate result")
            au_btn.click(ui_audit, [au_tenant], [au_info, au_border, au_guard], api_name="audit")
            au_inv_btn.click(ui_invalidate, [au_inv_tenant, au_inv_prefix], au_inv_out, **PRIVATE_EVENT)

        # ------------------------------------------------------------------ 7
        with gr.Tab("Backup"):
            gr.Markdown(SLEEP_WARNING)
            with gr.Row():
                bk_include = gr.Checkbox(label="include response bodies", value=True)
                bk_tenant = gr.Textbox(label="tenant_id (blank = all tenants)", value="", max_lines=1)
                bk_export_btn = gr.Button("Export index", variant="primary")
            with gr.Row():
                bk_file = gr.File(label="index.json", interactive=False)
                bk_receipt = gr.JSON(label="export receipt")
            gr.Markdown(
                "Exports carry prompts (needed to rebuild vectors on import). Untick *include response bodies* "
                "to share an index for analysis without shipping answers - those rows are then skipped on import."
            )
            gr.Markdown("---\n### Import")
            with gr.Row():
                bk_upload = gr.File(label="index.json", file_types=[".json"], type="filepath")
                bk_paste = gr.Textbox(label="...or paste the JSON", lines=4)
                bk_skip_safety = gr.Checkbox(label="skip the safety re-screen (not recommended)", value=False)
            bk_import_btn = gr.Button("Import index", variant="primary")
            bk_import_out = gr.JSON(label="import result")
            bk_export_btn.click(ui_export_index, [bk_include, bk_tenant], [bk_file, bk_receipt],
                                api_name="export_index")
            bk_import_btn.click(ui_import_index, [bk_upload, bk_paste, bk_skip_safety], bk_import_out,
                                api_name="import_index")

        # ------------------------------------------------------------------ 8
        with gr.Tab("Health"):
            hl_btn = gr.Button("Refresh", variant="primary")
            hl_out = gr.JSON(label="health")
            gr.Markdown(SLEEP_WARNING)
            hl_btn.click(ui_health, None, hl_out, api_name="health")

        # ------------------------------------------------------------------ 9
        with gr.Tab("API Docs"):
            api_md = gr.Markdown()
            api_btn = gr.Button("Rebuild paths for this session")
            api_btn.click(ui_api_docs, None, api_md, **PRIVATE_EVENT)

    # PRIVATE_EVENT: page-load refreshes are not part of the public API surface.
    demo.load(ui_api_docs, None, api_md, **PRIVATE_EVENT)
    demo.load(ui_health, None, hl_out, **PRIVATE_EVENT)

    gr.Markdown(
        "---\n"
        "Savings are **estimates** from a local token approximation and the prices you configured - not a bill. "
        "Semantic caching returns an answer written for a *different* prompt: MismatchGuard exists because a "
        "wrong hit costs far more than the tokens it saves."
    )


def _configure_queue(blocks: Any) -> Any:
    """Guarantees: a conservative queue on any Gradio version, falling back safely if a kwarg is unsupported."""
    for kwargs in ({"max_size": 32, "default_concurrency_limit": 2}, {"max_size": 32}, {}):
        try:
            return blocks.queue(**kwargs)
        except TypeError:
            continue
        except Exception:
            return blocks
    return blocks


demo = _configure_queue(demo)

if __name__ == "__main__":
    demo.launch()