app.py

"""
Flask API — local LLM inference (llama-cpp-python + GGUF only)

Multi-model: up to 2 models loaded simultaneously.
Each model lives in ./model/<ModelName>/chunks/*.gguf

Endpoints:
  GET  /                           — API index + examples
  GET  /health                     — all models status
  GET/POST /chat?message=...       — JSON reply
  GET/POST /chat/stream?message=   — streaming text/plain reply
  GET  /memory?session=X           — list conversation memory
  POST /memory/clear               — clear session memory
  GET  /models                     — all configured models + disk status
  GET  /models/quants?model=X      — HF quantization list

Params:
  message    — required prompt
  model      — which model to use (default: first active model)
               e.g. ?model=Qwen3.5-4B or ?model=DeepSeek-R1-8B
  mode       — fast | thinking | balanced | code | auto (default: auto)
  max_tokens — int | "auto"  (overrides mode token budget)
  greedy     — 1/true  → deterministic, fastest decode
  thinking   — 1/true  → enable <think> CoT
  raw        — 1/true  → NO system prompt (jailbreak / base mode)
  session    — str (default "default") → isolate memory namespaces
  no_memory  — 1/true  → skip memory lookup
"""

import os, re, json, time, threading, subprocess, sys
from flask import Flask, request, jsonify, Response, stream_with_context
from flask.json.provider import DefaultJSONProvider

_CPU_THREADS = os.cpu_count() or 4


class _PrettyJSON(DefaultJSONProvider):
    def dumps(self, obj, **kw):
        kw.setdefault("indent", 2)
        kw.setdefault("ensure_ascii", False)
        kw.setdefault("sort_keys", False)
        return json.dumps(obj, **kw)


app = Flask(__name__)
app.json_provider_class = _PrettyJSON
app.json = _PrettyJSON(app)


# ── Memory (lazy-loaded) ──────────────────────────────────────────────────────
def _mem():
    try:
        import memory as _m
        return _m
    except Exception:
        return None


# ── Config ────────────────────────────────────────────────────────────────────
def _load_json() -> dict:
    p = os.path.join(os.path.dirname(os.path.abspath(__file__)), "install.json")
    try:
        with open(p) as f:
            return json.load(f)
    except Exception:
        return {}


def _cfg(key, section, default):
    v = os.environ.get(key)
    if v is not None:
        return v
    c = _load_json()
    key_map = {
        "MAX_NEW_TOKENS": "max_new_tokens", "TEMPERATURE": "temperature",
        "TOP_P": "top_p", "MAX_SEQ_LEN": "max_seq_len", "PORT": "port",
    }
    sk = key_map.get(key, key.lower())
    if section == "server":
        v = c.get("server", {}).get(sk)
    return v if v is not None else default


def _get_active_models(cfg: dict) -> list:
    v = cfg.get("active_models")
    if isinstance(v, list):
        return [x for x in v if x]
    s = cfg.get("active_model", "")
    return [s] if s else []


# ── Constants (read once at startup) ─────────────────────────────────────────
_CFG_J         = _load_json()
MAX_NEW_TOKENS = int(_cfg("MAX_NEW_TOKENS", "server", 512))
TEMPERATURE    = float(_cfg("TEMPERATURE",  "server", 0.6))
TOP_P          = float(_cfg("TOP_P",        "server", 0.9))
MAX_SEQ_LEN    = int(_cfg("MAX_SEQ_LEN",   "server", 4096))
PORT           = int(_cfg("PORT",          "server", 5000))

# ── Runtime state ─────────────────────────────────────────────────────────────
_backends:       dict  = {}   # model_name → GGUFBackend (loaded & ready)
_backend_errors: dict  = {}   # model_name → error string
_loading_names:  set   = set()# model names currently being loaded
model_loading    = True       # True until all models finish (or fail)
load_start       = time.time()
_setup_status    = "starting"


def _get_backend(model_param: str | None = None):
    """
    Resolve model_param to a loaded GGUFBackend.
    Returns (backend, error_msg). backend is None on error.
    """
    if not model_param:
        if _backends:
            return next(iter(_backends.values())), None
        return None, "No models loaded yet"

    # Exact match
    if model_param in _backends:
        return _backends[model_param], None

    # Case-insensitive / partial match
    p = model_param.lower()
    for name, be in _backends.items():
        if p in name.lower():
            return be, None

    loaded = list(_backends.keys())
    if not loaded:
        return None, f"No models loaded yet. Requested: '{model_param}'"
    return None, (
        f"Model '{model_param}' not loaded. "
        f"Loaded: {loaded}. "
        f"Use ?model= with one of the loaded model names."
    )


# ── Keyword sets ──────────────────────────────────────────────────────────────
_CODING_KW = {
    "code", "html", "css", "javascript", "python", "java", "c++", "bash", "sql",
    "express", "flask", "django", "react", "vue", "node", "typescript", "rust",
    "write", "create", "implement", "function", "class", "script", "snippet",
    "algorithm", "debug", "fix", "api", "loop", "array", "dict", "json",
    "program", "build", "make", "compile", "parse", "regex", "database",
}
_DETAIL_KW = {
    "explain", "describe", "history", "compare", "difference",
    "analyze", "summarize", "how does", "why does",
    "tell me about", "in detail", "step by step",
    "advantages", "disadvantages", "pros and cons",
    "what is", "what are", "ano ang", "anong", "paano", "bakit",
    "pilipinas", "philippines", "kultura", "culture", "traditional",
}
_THINKING_KW = {
    "solve", "calculate", "prove", "reason", "logic",
    "math", "equation", "plan", "strategy", "optimize",
    "figure out", "work out", "deduce",
}
_GREET = {
    "hi", "hello", "hey", "sup", "yo", "hola",
    "kumusta", "kamusta", "greetings", "howdy",
    "magandang umaga", "magandang hapon", "magandang gabi",
    "good morning", "good afternoon", "good evening", "good day",
}


def _is_greeting(msg: str) -> bool:
    m = msg.lower().strip().rstrip("!?.,")
    w = m.split()
    return m in _GREET or (len(w) <= 3 and any(m.startswith(g) for g in _GREET))


# ── Token budgets per mode ────────────────────────────────────────────────────
_MODE_TOKENS = {
    "fast":          120,
    "thinking":      800,
    "thinking_fast": 400,
    "balanced":      None,
    "code":          600,
    "auto":          None,
}
_MODE_GREEDY = {
    "fast":          True,
    "thinking":      False,
    "thinking_fast": False,
    "balanced":      None,
    "code":          False,
    "auto":          None,
}


def _auto_max_tokens(message: str, thinking: bool) -> int:
    if thinking:
        return 700
    msg = message.lower().strip()
    n   = len(msg.split())
    if _is_greeting(message):
        return 100
    if any(k in msg for k in _CODING_KW):
        return 350
    if any(k in msg for k in _THINKING_KW):
        return 200
    if any(k in msg for k in _DETAIL_KW) or n > 8:
        return 200
    return 120


def _resolve_mode_params(message: str, mode_raw: str,
                         thinking_flag: bool, greedy_flag: bool,
                         max_tok_raw) -> tuple:
    mode = (mode_raw or "auto").lower().strip()
    if mode not in _MODE_TOKENS:
        mode = "auto"

    thinking = thinking_flag
    if mode in ("thinking", "thinking_fast"):
        thinking = True
    elif mode == "fast":
        thinking = False

    if max_tok_raw is not None and str(max_tok_raw).lower() != "auto":
        try:
            max_tok = max(1, min(int(max_tok_raw), 4096))
        except (TypeError, ValueError):
            max_tok = _auto_max_tokens(message, thinking)
    else:
        budget  = _MODE_TOKENS[mode]
        max_tok = budget if budget is not None else _auto_max_tokens(message, thinking)

    if greedy_flag:
        greedy = True
    elif _MODE_GREEDY[mode] is not None:
        greedy = _MODE_GREEDY[mode]
    else:
        greedy = _is_greeting(message) and not thinking

    return thinking, greedy, max_tok, mode


# ── System prompts ────────────────────────────────────────────────────────────
_SYS = {
    "greet": (
        "You are a friendly, helpful AI assistant. "
        "Greet the user warmly in 2–3 sentences. "
        "Mention you can answer questions, write code, explain topics, and chat."
    ),
    "brief": (
        "You are a knowledgeable assistant. "
        "Give a clear, accurate answer in 2–3 sentences. Be direct and helpful."
    ),
    "detail": (
        "You are a knowledgeable assistant. "
        "Explain clearly and accurately in 3–6 sentences. Cover the key points."
    ),
    "code": (
        "You are an expert programmer. "
        "Write clean, working code with brief inline comments. "
        "No lengthy preamble — code first."
    ),
    "think": (
        "You are an expert problem-solver and programmer. "
        "Think step by step inside <think> tags, then give a clear final answer."
    ),
    "fast": (
        "You are a concise assistant. "
        "Answer in 1–2 sentences. Be direct and accurate. No padding."
    ),
}


def _auto_sys_mode(message: str, thinking: bool, mode: str) -> str:
    if mode == "thinking" or thinking:
        return "think"
    if mode == "fast":
        return "fast"
    if mode == "code":
        return "code"
    if _is_greeting(message):
        return "greet"
    msg = message.lower()
    if any(k in msg for k in _CODING_KW):
        return "code"
    if any(k in msg for k in _DETAIL_KW) or len(message.split()) > 8:
        return "detail"
    return "brief"


_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)


def _clean(text: str) -> tuple:
    m = re.search(r"<think>(.*?)</think>", text, re.DOTALL)
    if m:
        think = m.group(1).strip()
        clean = _THINK_RE.sub("", text).strip()
        return clean, think if think else None
    if "<think>" in text:
        idx   = text.index("<think>")
        think = text[idx + len("<think>"):].strip()
        clean = text[:idx].strip()
        return clean, think if think else None
    return text.strip(), None


def _build_messages(message: str, thinking: bool = False,
                    raw: bool = False, mode: str = "auto",
                    context: list | None = None,
                    backend=None) -> list:
    msgs = []

    if not raw:
        sys_key = _auto_sys_mode(message, thinking, mode)
        msgs.append({"role": "system", "content": _SYS[sys_key]})

    if context:
        ctx_text = "\n---\n".join(context[-2:])
        msgs.append({
            "role":    "assistant",
            "content": f"[Relevant past context]\n{ctx_text}",
        })

    user_content = message
    if not thinking and backend is not None:
        try:
            from gguf_backend import GGUFBackend
            if isinstance(backend, GGUFBackend):
                user_content = message + " /no_think"
        except ImportError:
            pass

    msgs.append({"role": "user", "content": user_content})
    return msgs


# ── Background setup helpers ──────────────────────────────────────────────────
def _pip_install(*packages: str) -> None:
    cmd = [sys.executable, "-m", "pip", "install", "-q", "--no-cache-dir"] + list(packages)
    subprocess.run(cmd, capture_output=True)


def _pkg_ok(mod: str) -> bool:
    try:
        __import__(mod)
        if mod == "llama_cpp":
            import llama_cpp as _lc
            _ = _lc.llama_backend_init
        return True
    except (ImportError, RuntimeError, OSError, AttributeError):
        return False


def _ensure_optional_packages() -> None:
    global _setup_status
    core = []
    for mod, pkg in [
        ("llama_cpp",       "llama-cpp-python"),
        ("huggingface_hub", "huggingface-hub"),
        ("numpy",           "numpy"),
    ]:
        if not _pkg_ok(mod):
            core.append(pkg)
    if core:
        _setup_status = f"installing: {', '.join(core)}"
        print(f"[startup] Installing core packages: {core}", flush=True)
        if "llama-cpp-python" in core:
            subprocess.run(
                [sys.executable, "-m", "pip", "install", "-q", "--no-cache-dir",
                 "llama-cpp-python",
                 "--extra-index-url", "https://abetlen.github.io/llama-cpp-python/whl/cpu"],
                capture_output=True,
            )
            core = [p for p in core if p != "llama-cpp-python"]
        if core:
            _pip_install(*core)
        print("[startup] Core packages ready", flush=True)

    optional = []
    for mod, pkg in [
        ("chromadb",             "chromadb"),
        ("sentence_transformers","sentence-transformers"),
    ]:
        if not _pkg_ok(mod):
            optional.append(pkg)
    if optional:
        _setup_status = f"installing: {', '.join(optional)}"
        print(f"[startup] Installing memory packages: {optional}", flush=True)
        _pip_install(*optional)
        print("[startup] Memory packages ready", flush=True)


def _resolve_model_path(name: str, cfg: dict) -> str:
    """Return full path to the model file (may not exist yet — chunks handle it)."""
    model_dir = cfg.get("model_dir", "./model")
    mc        = cfg.get("models", {}).get(name, {})
    fn        = mc.get("file", "")
    if not fn:
        raise RuntimeError(f"Model '{name}' has no file configured in install.json")
    return os.path.join(model_dir, name, fn)


def _chunks_exist_for(model_path: str) -> bool:
    """True if split chunks exist for the model file."""
    base      = os.path.splitext(os.path.basename(model_path))[0]
    chunk_dir = os.path.join(os.path.dirname(model_path), "chunks")
    if not os.path.isdir(chunk_dir):
        return False
    return any(
        f.startswith(base) and (f.endswith(".gguf") or f.endswith(".part"))
        for f in os.listdir(chunk_dir)
    )


# ── Model loading (background thread) ────────────────────────────────────────
def load_all_models():
    global model_loading, _setup_status

    try:
        _setup_status = "checking packages"
        _ensure_optional_packages()

        from gguf_backend import GGUFBackend

        cfg     = _load_json()
        actives = _get_active_models(cfg)

        if not actives:
            _setup_status = "error: no active_models in install.json"
            print(f"[startup] ERROR: active_models is empty in install.json", flush=True)
            return

        _loading_names.update(actives)

        for name in actives:
            _setup_status = f"loading {name}"
            try:
                model_path = _resolve_model_path(name, cfg)
                # Check that file or chunks exist
                if not os.path.isfile(model_path) and not _chunks_exist_for(model_path):
                    raise RuntimeError(
                        f"Model file not found: {model_path}\n"
                        f"Download with: python model_manager.py download\n"
                        f"Or switch: python model_manager.py use <ModelName>"
                    )
                print(f"[startup] Loading {name} ...", flush=True)
                backend = GGUFBackend(
                    model_path,
                    n_threads=_CPU_THREADS,
                    n_ctx=MAX_SEQ_LEN,
                )
                _backends[name] = backend
                elapsed = round(time.time() - load_start, 1)
                print(f"[startup] {name} ready — {backend.backend}  ({elapsed}s)", flush=True)
            except Exception as e:
                _backend_errors[name] = str(e)
                print(f"[startup] {name} FAILED: {e}", flush=True)
                import traceback; traceback.print_exc()
            finally:
                _loading_names.discard(name)

    except Exception as e:
        _setup_status = f"error: {e}"
        print(f"[startup] FATAL: {e}", flush=True)
        import traceback; traceback.print_exc()
    finally:
        model_loading = False
        if _backends:
            names = list(_backends.keys())
            _setup_status = f"ready ({len(names)} model{'s' if len(names)>1 else ''})"
        else:
            _setup_status = "error: all models failed"
        total = round(time.time() - load_start, 1)
        print(f"[startup] Load complete in {total}s — ready: {list(_backends.keys())}", flush=True)


threading.Thread(target=load_all_models, daemon=True).start()


# ── Request helpers ───────────────────────────────────────────────────────────
def _get(key, default=None):
    if request.method == "GET":
        return request.args.get(key, default)
    return (request.get_json(silent=True) or {}).get(key, default)


def _ready():
    """Check if at least one model is ready to serve requests."""
    if model_loading and not _backends:
        loading_list = list(_loading_names)
        return False, jsonify({
            "error":   "Models loading — check /health",
            "stage":   _setup_status,
            "loading": loading_list,
        }), 503
    if not _backends and _backend_errors:
        errs = _backend_errors
        return False, jsonify({
            "error":  "All models failed to load",
            "errors": errs,
        }), 500
    return True, None, None


def _parse_bool(val, default: bool = False) -> bool:
    if val is None:
        return default
    return str(val).lower() in ("1", "true", "yes")


# ── Routes ────────────────────────────────────────────────────────────────────
@app.route("/", methods=["GET"])
def index():
    base   = request.host_url.rstrip("/")
    status = "loading" if (model_loading and not _backends) else ("error" if (not _backends) else "ok")
    loaded = list(_backends.keys())
    return jsonify({
        "name":    "Local LLM API (multi-model)",
        "status":  status,
        "loaded_models": loaded,
        "threads": _CPU_THREADS,
        "params": {
            "message":    "required — the prompt",
            "model":      f"which model — e.g. {loaded[0] if loaded else 'ModelName'}",
            "mode":       "fast | thinking | balanced | code | auto (default: auto)",
            "max_tokens": "int | auto  (overrides mode budget)",
            "greedy":     "1/true → deterministic, fastest",
            "thinking":   "1/true → CoT reasoning (same as mode=thinking)",
            "raw":        "1/true → no system prompt (jailbreak)",
            "session":    "string → memory namespace",
            "no_memory":  "1/true → skip memory lookup",
        },
        "modes": {
            "fast":          f"greedy + 120 tok — ~1-2s   ({base}/chat?message=hi&mode=fast)",
            "thinking":      f"CoT   + 800 tok  — ~20-40s ({base}/chat?message=solve+x&mode=thinking)",
            "thinking_fast": f"CoT   + 400 tok  — ~10-20s ({base}/chat?message=solve+x&mode=thinking_fast)",
            "balanced":      f"auto detect      — 120-350 tok (default)",
            "code":          f"code prompt + 600 tok ({base}/chat?message=make+a+flask+api&mode=code)",
        },
        "examples": {
            "fast_hello":     f"{base}/chat?message=hi&mode=fast&no_memory=1",
            "with_deepseek":  f"{base}/chat?message=solve+fibonacci&model=DeepSeek-R1-8B&mode=thinking",
            "with_qwen":      f"{base}/chat?message=write+hello+world&model=Qwen3.5-4B&mode=code",
            "balanced":       f"{base}/chat/stream?message=what+is+gravity&no_memory=1",
            "memory":         f"{base}/memory?session=default",
        },
        "model_management": {
            "list_models":    f"{base}/models",
            "list_quants":    f"{base}/models/quants?model=Qwen3.5-4B",
            "cli_list":       "python model_manager.py list",
            "cli_add":        "python model_manager.py use DeepSeek-R1-8B",
            "cli_remove":     "python model_manager.py remove DeepSeek-R1-8B",
            "cli_quants":     "python model_manager.py quants Qwen3.5-4B",
        },
    }), 200


@app.route("/health", methods=["GET"])
def health():
    elapsed = round(time.time() - load_start, 1)

    if model_loading and not _backends:
        return jsonify({
            "status":   "loading",
            "stage":    _setup_status,
            "loading":  list(_loading_names),
            "elapsed":  elapsed,
        }), 503

    mem_info = {}
    m = _mem()
    if m:
        try:
            mem_info = m.stats()
        except Exception:
            mem_info = {"error": "unavailable"}

    models_info = {}
    for name, be in _backends.items():
        models_info[name] = {
            "status":       "ok",
            "backend":      be.backend,
            "model_file":   getattr(be, "model_fn", "N/A"),
            "quantization": getattr(be, "quant",    "N/A"),
        }
    for name, err in _backend_errors.items():
        if name not in models_info:
            models_info[name] = {"status": "error", "error": err}

    overall = "ok" if _backends else "error"
    if _backends and _backend_errors:
        overall = "partial"  # some loaded, some failed

    return jsonify({
        "status":      overall,
        "loaded":      list(_backends.keys()),
        "failed":      list(_backend_errors.keys()),
        "models":      models_info,
        "threads":     _CPU_THREADS,
        "n_ctx":       MAX_SEQ_LEN,
        "startup_sec": elapsed,
        "memory":      mem_info,
        "modes":       list(_MODE_TOKENS.keys()),
    }), 200 if overall != "error" else 500


def _chat_common(stream_mode: bool):
    ok, *rest = _ready()
    if not ok:
        return rest[0], rest[1]

    # Resolve which model to use
    model_param = (_get("model") or "").strip() or None
    backend, err_msg = _get_backend(model_param)
    if backend is None:
        loaded = list(_backends.keys())
        code   = 404 if (model_param and loaded) else 503
        return jsonify({
            "error":         err_msg,
            "loaded_models": loaded,
        }), code

    message = (_get("message") or "").strip()
    if not message:
        return jsonify({"error": "Missing 'message'."}), 400

    mode_raw    = (_get("mode") or "auto").strip().lower()
    think_flag  = _parse_bool(_get("thinking"), False)
    greedy_flag = _parse_bool(_get("greedy"),   False)
    no_memory   = _parse_bool(_get("no_memory"),False)
    raw         = _parse_bool(_get("raw"),      False)
    session     = (_get("session") or "default").strip()

    thinking, greedy, max_tok, mode = _resolve_mode_params(
        message, mode_raw, think_flag, greedy_flag, _get("max_tokens")
    )

    if raw:
        thinking = think_flag
        mode     = "raw"

    context: list = []
    m = _mem()
    if m and not no_memory and not raw and not _is_greeting(message):
        try:
            context = m.retrieve(message, session=session, top_k=2)
        except Exception:
            pass

    msgs = _build_messages(message, thinking=thinking, raw=raw,
                           mode=mode, context=context, backend=backend)

    be_name = getattr(backend, "backend", "GGUF")

    if stream_mode:
        def stream_gen():
            try:
                full = ""
                for chunk in backend.stream(msgs, max_tokens=max_tok,
                                            temperature=TEMPERATURE, top_p=TOP_P,
                                            top_k=40, greedy=greedy):
                    if chunk.startswith("\x00"):
                        clean = chunk[1:]
                        if m and not no_memory and clean:
                            threading.Thread(
                                target=m.store, args=(message, clean),
                                kwargs={"session": session}, daemon=True,
                            ).start()
                    else:
                        full += chunk
                        yield chunk
            except Exception as e:
                yield f"\n[ERROR] {e}"

        return Response(
            stream_with_context(stream_gen()),
            content_type="text/plain; charset=utf-8",
        )

    try:
        t0 = time.time()
        response, thk = backend.generate(msgs, max_tokens=max_tok,
                                         temperature=TEMPERATURE, top_p=TOP_P,
                                         top_k=40, greedy=greedy)
        elapsed = round(time.time() - t0, 2)

        if m and not no_memory and response:
            threading.Thread(target=m.store, args=(message, response),
                             kwargs={"session": session}, daemon=True).start()

        result = {
            "message":         message,
            "response":        response,
            "elapsed_seconds": elapsed,
            "max_tokens_used": max_tok,
            "mode":            mode,
            "greedy":          greedy,
            "thinking":        thinking,
            "raw":             raw,
            "session":         session,
            "memory_context":  len(context),
            "model":           model_param or list(_backends.keys())[0],
            "backend":         be_name,
        }
        if thk:
            result["think_block"] = thk
        return jsonify(result), 200

    except Exception as e:
        import traceback; traceback.print_exc()
        return jsonify({"error": str(e)}), 500


@app.route("/chat", methods=["GET", "POST"])
def chat():
    return _chat_common(stream_mode=False)


@app.route("/chat/stream", methods=["GET", "POST"])
def chat_stream():
    return _chat_common(stream_mode=True)


# ── Memory endpoints ──────────────────────────────────────────────────────────
@app.route("/memory", methods=["GET"])
def memory_list():
    m = _mem()
    if not m:
        return jsonify({"error": "Memory module not available"}), 503
    session = (_get("session") or "default").strip()
    try:
        return jsonify({
            "session": session,
            "stats":   m.stats(),
            "recent":  m.list_recent(session=session, limit=20),
        }), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/memory/clear", methods=["POST", "GET"])
def memory_clear():
    m = _mem()
    if not m:
        return jsonify({"error": "Memory module not available"}), 503
    session = (_get("session") or "default").strip()
    try:
        deleted = m.clear(session=session)
        return jsonify({"session": session, "deleted": deleted}), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500


# ── Model list / HF quant query endpoints ────────────────────────────────────
@app.route("/models", methods=["GET"])
def models_list():
    try:
        from model_manager import get_models_status
        status = get_models_status(_load_json())
        # Annotate which models are currently loaded in memory
        for name in status.get("models", {}):
            status["models"][name]["loaded_in_memory"] = name in _backends
        status["loaded_in_memory"] = list(_backends.keys())
        return jsonify(status), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/models/quants", methods=["GET"])
def models_quants():
    target = (request.args.get("model") or "").strip()
    if not target:
        return jsonify({
            "error":        "Missing ?model= parameter",
            "example":      "/models/quants?model=Qwen3.5-4B",
            "known_models": list(_load_json().get("models", {}).keys()),
        }), 400
    try:
        from model_manager import get_quants_for_model
        files = get_quants_for_model(target, _load_json())

        by_bits: dict = {}
        for f in files:
            by_bits.setdefault(f["bits"], []).append({
                "file":    f["name"],
                "quant":   f["quant"],
                "size_gb": f["size_gb"],
            })

        return jsonify({
            "model":   target,
            "total":   len(files),
            "by_bits": by_bits,
            "all":     files,
        }), 200
    except RuntimeError as e:
        return jsonify({"error": str(e)}), 502
    except Exception as e:
        return jsonify({"error": str(e)}), 500


if __name__ == "__main__":
    print(f"[server] LLM API  port={PORT}  threads={_CPU_THREADS}")
    print(f"[server] Open: http://localhost:{PORT}/")
    app.run(host="0.0.0.0", port=PORT, threaded=True)