app.py

import os
# Keep all model inference on CPU; this must be set before importing torch.
os.environ["CUDA_VISIBLE_DEVICES"] = ""
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
import platform
import sys
import time
import shutil
import tempfile
import threading
import spaces
import torch
import torchaudio
import gradio as gr
from fastapi import Body
from fastapi.responses import JSONResponse
from einops import rearrange
from huggingface_hub import login
from stable_audio_3 import StableAudioModel
import urllib.request
import urllib.parse
import json as _json

import node_probe
import node_runtime
import searx_bridge
import searx_runtime

for _stream in (sys.stdout, sys.stderr):
    if hasattr(_stream, "reconfigure"):
        _stream.reconfigure(encoding="utf-8", errors="backslashreplace")

hf_token = os.environ.get("HF_TOKEN")
if hf_token:
    login(token=hf_token)

@spaces.GPU(duration=1)
def _gpu_startup_touch():
    return "ok"

# Keep the decorated symbol so ZeroGPU Spaces recognize the app, but do not call it.
# The Respite API and all CPU benchmarks must not consume GPU time.


def _log(msg):
    print(msg.encode("ascii", "backslashreplace").decode("ascii"), flush=True)


def _read_cgroup(path):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read().strip()
    except (FileNotFoundError, OSError):
        return None


def _parse_cpu_count():
    v = _read_cgroup("/sys/fs/cgroup/cpu.max")
    if v:
        p = v.split()
        if len(p) == 2 and p[0] != "max":
            try:
                q, per = int(p[0]), int(p[1])
                if per > 0: return q // per
            except: pass
    v = _read_cgroup("/sys/fs/cgroup/cpuset.cpus")
    if v:
        c = 0
        for part in v.split(","):
            if "-" in part:
                s, e = part.split("-", 1)
                c += int(e) - int(s) + 1
            else:
                c += 1
        if c > 0: return c
    return os.cpu_count()


def _parse_ram():
    v = _read_cgroup("/sys/fs/cgroup/memory.max")
    if v and v != "max":
        try:
            val = int(v)
            if val < 2**62: return val
        except: pass
    try:
        with open("/proc/meminfo", "r", encoding="utf-8") as f:
            for line in f:
                if line.startswith("MemTotal:"):
                    return int(line.split()[1]) * 1024
    except: pass
    return None


# ---------------------------------------------------------------------------
# API metadata
# ---------------------------------------------------------------------------

API_RESOURCES = {
    "audio_generation": {
        "name": "Audio generation",
        "description": "Generate music or sound effects from a text prompt.",
        "endpoint": "/respite/audio/generate",
        "method": "POST",
        "input": {
            "prompt": "string",
            "duration": "number (1-120 seconds)",
            "steps": "integer (1-50)",
            "cfg_scale": "number (0-10)",
            "seed": "integer (-1 for random)",
            "model": "small-music | small-sfx",
        },
        "output": "WAV audio file",
    },
    "qwen38_inference": {
        "name": "Qwen3.8-27B text generation",
        "description": "Generate text with Qwen3.8-27B using the Space CPU; no GPU allocation.",
        "endpoint": "/respite/text/qwen38",
        "method": "POST",
        "input": {
            "prompt": "string",
            "max_new_tokens": "integer (1-256, default 64)",
            "temperature": "number (0-2, default 0.7)",
            "thinking": "boolean (default true; Qwen3.8 is a reasoning model)",
        },
        "output": "JSON containing generated text and CPU timing",
    },
    "tinyllama_inference": {
        "name": "TinyLlama text generation",
        "description": "Generate text with TinyLlama 1.1B using the Space CPU; no GPU allocation.",
        "endpoint": "/respite/text/generate",
        "method": "POST",
        "input": {
            "prompt": "string",
            "max_new_tokens": "integer (1-256, default 64)",
            "temperature": "number (0-2, default 0.7)",
        },
        "output": "JSON containing generated text and CPU timing",
    },
    "web_search": {
        "name": "Web search",
        "description": "Search the web via DuckDuckGo.",
        "endpoint": "/respite/search",
        "method": "GET",
        "input": {
            "q": "string",
            "backend": "'text', 'news', or 'images'",
            "max_results": "integer (1-50, default 10)",
            "extract_top": "integer (0-5, default 0)",
            "region": "string (default 'wt-wt')",
        },
        "output": "JSON with title, url, content per result.",
    },
}

API_SPECS = {
    "name": "Respite API",
    "version": "1.0.0",
    "description": "General-purpose AI API server with audio generation capabilities.",
    "base_path": "/respite",
    "authentication": "none",
    "content_types": ["application/json", "audio/wav"],
    "resources_endpoint": "/respite/resources",
    "specs_endpoint": "/respite/specs",
    "limits": {
        "max_concurrent_requests": 1,
        "max_queue_size": 4,
        "audio_max_duration_seconds": 120,
    },
}


def _get_runtime_specs():
    storage = shutil.disk_usage(os.getcwd())
    hw = os.environ.get("SPACE_HARDWARE", "zero-a10g")
    is_zgpu = "zero" in hw.lower()
    return {
        "platform": platform.platform(),
        "python_version": platform.python_version(),
        "cpu_cores": _parse_cpu_count(),
        "host_cpu_cores": os.cpu_count(),
        "ram_bytes": _parse_ram(),
        "storage_total_bytes": storage.total,
        "storage_used_bytes": storage.used,
        "storage_free_bytes": storage.free,
        "hardware": hw,
        "gpu": {"type": "NVIDIA RTX Pro 6000 Blackwell", "vram_bytes": 48*1024**3, "shared_zero_gpu": True} if is_zgpu else None,
    }


# ---------------------------------------------------------------------------
# Web search
# ---------------------------------------------------------------------------

try:
    from ddgs import DDGS

    def _search_web(query, backend="text", max_results=10, extract_top=0, region="wt-wt"):
        extract_top = max(0, min(5, extract_top))
        max_results = max(1, min(50, max_results))
        with DDGS() as ddgs:
            if backend == "news":
                results = [{"title": r.get("title",""), "url": r.get("url",""), "content": r.get("body",""), "source": r.get("source",""), "date": r.get("date","")} for r in ddgs.news(query, max_results=max_results, region=region)]
            elif backend == "images":
                results = [{"title": r.get("title",""), "url": r.get("image",""), "source": r.get("source","")} for r in ddgs.images(query, max_results=max_results, region=region)]
            else:
                results = [{"title": r.get("title",""), "url": r.get("href",""), "content": r.get("body","")} for r in ddgs.text(query, max_results=max_results, region=region)]
        if extract_top > 0 and results and backend != "images":
            with DDGS() as ddgs:
                for url in [r["url"] for r in results[:extract_top] if r.get("url")]:
                    try:
                        ex = ddgs.extract(url)
                        ct = ex.get("content","")
                        if ct:
                            for r in results:
                                if r["url"] == url:
                                    r["extracted_content"] = ct[:8000]
                                    break
                    except: pass
        return {"query": query, "backend": backend, "results": results, "total_results": len(results)}

except ImportError:
    def _search_web(query, **kw):
        raise RuntimeError("Search not available")


# ---------------------------------------------------------------------------
# Benchmark
# ---------------------------------------------------------------------------

_BENCH_MODELS = [
    {"name": "TinyLlama 1.1B (control)", "repo": "TheBloke/TinyLlama-1.1B-Chat-v1.0-GGUF", "file": "tinyllama-1.1b-chat-v1.0.Q4_K_M.gguf", "size_gb": 0.7},
    {"name": "Gemma 4 E4B", "repo": "unsloth/gemma-4-E4B-it-GGUF", "file": "gemma-4-E4B-it-Q4_K_M.gguf", "size_gb": 2.5},
    {"name": "Qwen3 8B", "repo": "Qwen/Qwen3-8B-GGUF", "file": "Qwen3-8B-Q4_K_M.gguf", "size_gb": 5.0},
    {"name": "Qwen3.8 27B", "repo": "unsloth/Qwen3.8-27B-GGUF", "file": "Qwen3.8-27B-Q4_0.gguf", "size_gb": 16.0},
]
_BENCH_CACHE = os.path.join(tempfile.gettempdir(), "llm_bench_cpu_v2")
_BENCH_RESULTS = []
_BENCH_LOCK = threading.Lock()
_BENCH_JOBS = {}
_BENCH_JOBS_LOCK = threading.Lock()


def _ensure_llama():
    d = os.path.join(_BENCH_CACHE, "bin")
    cli = os.path.join(d, "llama-cli")
    bench = os.path.join(d, "llama-bench")
    if os.path.isfile(cli) and os.path.isfile(bench): return cli
    import subprocess
    os.makedirs(d, exist_ok=True)
    src = os.path.join(_BENCH_CACHE, "src")
    if os.path.isdir(src): shutil.rmtree(src, ignore_errors=True)
    _log("Building llama.cpp...")
    subprocess.check_call(["git","clone","--depth=1","https://github.com/ggml-org/llama.cpp.git",src], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    bd = os.path.join(_BENCH_CACHE, "build")
    os.makedirs(bd, exist_ok=True)
    nt = _parse_cpu_count() or 8
    subprocess.check_call(["cmake","-S",src,"-B",bd,"-DGGML_NATIVE=OFF","-DGGML_OPENMP=OFF","-DGGML_CUDA=OFF","-DGGML_AVX=ON","-DGGML_AVX2=ON","-DGGML_FMA=ON","-DGGML_AVX512=OFF","-DBUILD_SHARED_LIBS=OFF"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    subprocess.check_call(["cmake","--build",bd,"-j",str(nt),"--target","llama-cli","llama-bench"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    import glob as _g
    cands = [p for p in _g.glob(os.path.join(bd,"**","llama-cli"), recursive=True) if os.path.isfile(p)]
    bench_cands = [p for p in _g.glob(os.path.join(bd,"**","llama-bench"), recursive=True) if os.path.isfile(p)]
    if cands:
        shutil.copy2(cands[0], cli); os.chmod(cli, 0o755)
    else: raise RuntimeError("llama-cli not found")
    if bench_cands:
        bench = os.path.join(d, "llama-bench")
        shutil.copy2(bench_cands[0], bench); os.chmod(bench, 0o755)
    return cli


def _run_bench(model_filter=None):
    with _BENCH_LOCK:
        return _run_bench_locked(model_filter)


def _run_bench_locked(model_filter=None):
    global _BENCH_RESULTS
    if _BENCH_RESULTS and not model_filter: return _BENCH_RESULTS
    import subprocess, re
    from huggingface_hub import hf_hub_download
    os.makedirs(_BENCH_CACHE, exist_ok=True)
    cli = _ensure_llama()
    bench = os.path.join(os.path.dirname(cli), "llama-bench")
    if not os.path.isfile(bench):
        raise RuntimeError("llama-bench was not built; redeploy to rebuild llama.cpp")
    nt = _parse_cpu_count() or 8
    models = [m for m in _BENCH_MODELS if m["name"]==model_filter] if model_filter else _BENCH_MODELS
    results = []
    for mi in models:
        r = {"name": mi["name"], "status": "error", "engine": "llama-bench", "gpu_layers": 0}
        try:
            mp = hf_hub_download(repo_id=mi["repo"], filename=mi["file"], cache_dir=_BENCH_CACHE)
            env = os.environ.copy()
            env["LD_LIBRARY_PATH"] = os.path.dirname(cli)
            env["CUDA_VISIBLE_DEVICES"] = ""
            env["GGML_CUDA"] = "0"
            env["OMP_NUM_THREADS"] = str(nt)
            env["OMP_THREAD_LIMIT"] = str(nt)
            env["OMP_DYNAMIC"] = "FALSE"
            env["MKL_NUM_THREADS"] = str(nt)
            env["OPENBLAS_NUM_THREADS"] = str(nt)
            env["VECLIB_MAXIMUM_THREADS"] = str(nt)
            env["NUMEXPR_NUM_THREADS"] = str(nt)
            t0 = time.time()
            to = 90 if mi["size_gb"] <= 6 else 180
            command = [bench, "-m", mp, "-t", str(nt), "-ngl", "0", "-c", "256", "-p", "32", "-n", "64", "-r", "1", "--no-mmap", "-o", "json"]
            _log(f"Running llama-bench for {mi['name']} on CPU with {nt} threads")
            with tempfile.TemporaryFile(mode="w+", encoding="utf-8") as output_file:
                p = subprocess.run(command, stdout=output_file, stderr=subprocess.PIPE, text=True, timeout=to, env=env)
                output_file.seek(0)
                raw = output_file.read().strip()
            wt = time.time()-t0
            diagnostics = (p.stderr or "").strip()
            if p.returncode != 0:
                raise RuntimeError(f"llama-bench exited with code {p.returncode}: {diagnostics[-1500:]}")
            parsed = json.loads(raw)
            row = parsed[0] if isinstance(parsed, list) and parsed else parsed
            r.update({"status":"success", "total_time_sec":round(wt,2), "prompt_tokens":32, "completion_tokens":64, "prompt_tokens_per_sec":row.get("avg_ts", 0), "generation_tokens_per_sec":row.get("avg_ts", 0), "raw":row})
        except subprocess.TimeoutExpired:
            r["error"] = f"llama-bench timed out after {to}s on CPU"
            results.append(r)
            continue
        except Exception as e:
            r["error"] = str(e)
        results.append(r)
        import glob as _g
        for f in _g.glob(os.path.join(_BENCH_CACHE,"**","*.gguf"), recursive=True):
            try: os.unlink(f)
            except: pass
    if not model_filter: _BENCH_RESULTS = results
    return results


def _run_cpu_smoke(model_filter):
    """Run a short CPU-only loader/inference diagnostic for one model."""
    import subprocess
    from huggingface_hub import hf_hub_download

    model = next((m for m in _BENCH_MODELS if m["name"] == model_filter), None)
    if model is None:
        raise ValueError(f"Unknown model: {model_filter}")
    cli = _ensure_llama()
    env = os.environ.copy()
    env["LD_LIBRARY_PATH"] = os.path.dirname(cli)
    env["CUDA_VISIBLE_DEVICES"] = ""
    env["GGML_CUDA"] = "0"
    env["OMP_NUM_THREADS"] = "1"
    env["OMP_THREAD_LIMIT"] = "1"
    env["OMP_DYNAMIC"] = "FALSE"
    env["MKL_NUM_THREADS"] = "1"
    env["OPENBLAS_NUM_THREADS"] = "1"
    env["VECLIB_MAXIMUM_THREADS"] = "1"
    env["NUMEXPR_NUM_THREADS"] = "1"
    result = {
        "name": model["name"],
        "engine": "llama-cli",
        "gpu_layers": 0,
        "threads": 1,
        "no_mmap": True,
        "phases": {},
    }

    t0 = time.time()
    version = subprocess.run([cli, "--version"], capture_output=True, text=True, timeout=15, env=env)
    result["phases"]["binary"] = {
        "ok": version.returncode == 0,
        "elapsed_sec": round(time.time() - t0, 2),
        "returncode": version.returncode,
        "output": (version.stdout + version.stderr)[-1000:],
    }
    if version.returncode != 0:
        return result

    t0 = time.time()
    try:
        path = hf_hub_download(repo_id=model["repo"], filename=model["file"], cache_dir=_BENCH_CACHE)
        result["model_path"] = path
        result["phases"]["download"] = {"ok": True, "elapsed_sec": round(time.time() - t0, 2)}
    except Exception as exc:
        result["phases"]["download"] = {"ok": False, "elapsed_sec": round(time.time() - t0, 2), "error": str(exc)}
        return result

    help_t0 = time.time()
    try:
        help_run = subprocess.run([cli, "--help"], capture_output=True, text=True, timeout=15, env=env)
        result["phases"]["help"] = {"ok": help_run.returncode == 0, "elapsed_sec": round(time.time() - help_t0, 2), "returncode": help_run.returncode}
    except Exception as exc:
        result["phases"]["help"] = {"ok": False, "elapsed_sec": round(time.time() - help_t0, 2), "error": str(exc)}

    command = [cli, "-m", path, "-t", "1", "-ngl", "0", "-c", "256", "-n", "1", "-p", "Say hi.", "--no-display-prompt", "--no-warmup", "--no-mmap", "--log-disable", "-fa", "0"]
    t0 = time.time()
    try:
        run = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, timeout=180, env=env)
        result["phases"]["load_and_one_token"] = {
            "ok": run.returncode == 0,
            "elapsed_sec": round(time.time() - t0, 2),
            "returncode": run.returncode,
            "stdout": (run.stdout or "")[-1000:],
            "stderr": (run.stderr or "")[-3000:],
        }
    except subprocess.TimeoutExpired as exc:
        result["phases"]["load_and_one_token"] = {
            "ok": False,
            "elapsed_sec": round(time.time() - t0, 2),
            "timeout": True,
            "stdout": str(exc.stdout or "")[-1000:],
            "stderr": str(exc.stderr or "")[-3000:],
        }
    return result


def _run_cpu_process_probe(model_filter):
    import subprocess
    from huggingface_hub import hf_hub_download

    model = next((m for m in _BENCH_MODELS if m["name"] == model_filter), None)
    if model is None:
        raise ValueError(f"Unknown model: {model_filter}")
    cli = _ensure_llama()
    path = hf_hub_download(repo_id=model["repo"], filename=model["file"], cache_dir=_BENCH_CACHE)
    env = os.environ.copy()
    env["LD_LIBRARY_PATH"] = os.path.dirname(cli)
    env["CUDA_VISIBLE_DEVICES"] = ""
    env["GGML_CUDA"] = "0"
    env["OMP_NUM_THREADS"] = "1"
    env["OMP_THREAD_LIMIT"] = "1"
    env["OMP_DYNAMIC"] = "FALSE"
    env["MKL_NUM_THREADS"] = "1"
    env["OPENBLAS_NUM_THREADS"] = "1"
    env["VECLIB_MAXIMUM_THREADS"] = "1"
    env["NUMEXPR_NUM_THREADS"] = "1"
    command = [cli, "-m", path, "-t", "1", "-ngl", "0", "-c", "256", "-n", "1", "-p", "Say hi.", "--no-display-prompt", "--no-warmup", "--no-mmap", "--log-disable", "-fa", "0"]
    started = time.time()
    child = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
    time.sleep(8)
    probe = {"pid": child.pid, "elapsed_sec": round(time.time() - started, 2), "command": command, "gpu_layers": 0}
    try:
        probe["ps"] = subprocess.run(["ps", "-o", "pid,stat,pcpu,pmem,rss,wchan:40,cmd", "-p", str(child.pid)], capture_output=True, text=True, timeout=5).stdout
        with open(f"/proc/{child.pid}/io", "r", encoding="utf-8") as f:
            probe["io"] = f.read()
        with open(f"/proc/{child.pid}/status", "r", encoding="utf-8") as f:
            probe["status"] = "\n".join(line for line in f if line.startswith(("State:", "VmRSS:", "Threads:")))
    except Exception as exc:
        probe["probe_error"] = str(exc)
    finally:
        child.kill()
        stdout, stderr = child.communicate(timeout=10)
        probe["stdout"] = (stdout or "")[-2000:]
        probe["stderr"] = (stderr or "")[-4000:]
        probe["returncode"] = child.returncode
    return probe


def _run_cpu_diagnose(model_filter):
    import subprocess
    from huggingface_hub import hf_hub_download

    model = next((m for m in _BENCH_MODELS if m["name"] == model_filter), None)
    if model is None:
        raise ValueError(f"Unknown model: {model_filter}")
    cli = _ensure_llama()
    path = hf_hub_download(repo_id=model["repo"], filename=model["file"], cache_dir=_BENCH_CACHE)
    env = os.environ.copy()
    env["LD_LIBRARY_PATH"] = os.path.dirname(cli)
    env["CUDA_VISIBLE_DEVICES"] = ""
    env["GGML_CUDA"] = "0"

    def command_output(command, timeout=10):
        try:
            p = subprocess.run(command, capture_output=True, text=True, timeout=timeout, env=env)
            return {"returncode": p.returncode, "output": (p.stdout + p.stderr)[-5000:]}
        except Exception as exc:
            return {"error": str(exc)}

    result = {
        "name": model["name"],
        "model_path": path,
        "model_size_bytes": os.path.getsize(path),
        "model_magic": open(path, "rb").read(4).decode("ascii", "replace"),
        "cpu_cores": _parse_cpu_count(),
        "cpu_flags": command_output(["bash", "-lc", "grep -m1 '^flags' /proc/cpuinfo"], 5),
        "filesystem": command_output(["df", "-T", path], 5),
        "binary": command_output(["file", cli], 5),
        "libraries": command_output(["ldd", cli], 5),
    }
    trace_path = os.path.join(_BENCH_CACHE, "llama-load.trace")
    if shutil.which("strace"):
        try:
            p = subprocess.run(
                ["strace", "-f", "-tt", "-o", trace_path, cli, "-m", path,
                 "-t", "1", "-ngl", "0", "-c", "256", "-n", "1", "-p", "Say hi.",
                 "--no-display-prompt", "--no-warmup", "--no-mmap", "-fa", "0"],
                capture_output=True, text=True, timeout=25, env=env,
            )
            result["load_trace"] = {"returncode": p.returncode, "stdout": (p.stdout or "")[-1000:], "stderr": (p.stderr or "")[-2000:]}
        except subprocess.TimeoutExpired:
            result["load_trace"] = {"timeout": True}
        try:
            with open(trace_path, "r", encoding="utf-8", errors="replace") as f:
                result["trace_tail"] = f.read()[-12000:]
        except OSError as exc:
            result["trace_tail_error"] = str(exc)
    else:
        result["load_trace"] = {"available": False}
    return result


def _run_benchmark_job(job_id, kind, model_filter=None, force=False, request=None):
    try:
        if kind == "smoke":
            with _BENCH_LOCK:
                result = _run_cpu_smoke(model_filter)
            payload = {"status": "complete", "result": result}
        elif kind == "diagnose":
            with _BENCH_LOCK:
                result = _run_cpu_diagnose(model_filter)
            payload = {"status": "complete", "result": result}
        elif kind == "probe":
            with _BENCH_LOCK:
                result = _run_cpu_process_probe(model_filter)
            payload = {"status": "complete", "result": result}
        elif kind == "qwen38":
            request = request or _BENCH_JOBS.get(job_id, {}).get("request", {})
            qwen_prompt = request.get("prompt") or model_filter or ""
            with _BENCH_LOCK:
                result = _generate_qwen38(
                    qwen_prompt,
                    request.get("max_new_tokens", 256),
                    request.get("temperature", 0.7),
                    request.get("thinking", False),
                )
            payload = {"status": "complete", "result": result}
        else:
            if force:
                global _BENCH_RESULTS
                _BENCH_RESULTS = []
            payload = {
                "status": "complete",
                "models": _run_bench(model_filter=model_filter),
                "system": {
                    "cpu_cores": _parse_cpu_count(),
                    "ram_bytes": _parse_ram(),
                    "threads": _parse_cpu_count() or 8,
                },
            }
        with _BENCH_JOBS_LOCK:
            _BENCH_JOBS[job_id] = payload
    except Exception as exc:
        with _BENCH_JOBS_LOCK:
            _BENCH_JOBS[job_id] = {"status": "error", "error": str(exc)}


def _start_benchmark_job(kind, model_filter=None, force=False, request=None):
    job_id = f"bench-{int(time.time() * 1000)}-{threading.get_ident()}"
    with _BENCH_JOBS_LOCK:
        _BENCH_JOBS[job_id] = {
            "status": "running",
            "kind": kind,
            "model": model_filter,
            "started_at": time.time(),
            "request": request or {},
        }
    threading.Thread(
        target=_run_benchmark_job,
        args=(job_id, kind, model_filter, force, request or {}),
        daemon=True,
    ).start()
    return job_id


MODEL_CACHE = {}
_TINYLLAMA = None
_TINYLLAMA_TOKENIZER = None
_TINYLLAMA_LOCK = threading.Lock()
_TINYLLAMA_REPO = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"
_QWEN38 = None
_QWEN38_PROCESSOR = None
_QWEN38_LOCK = threading.Lock()
_QWEN38_REPO = "Qwen/Qwen3.8-27B"


def _load_qwen38():
    global _QWEN38, _QWEN38_PROCESSOR
    with _QWEN38_LOCK:
        if _QWEN38 is not None:
            return _QWEN38_PROCESSOR, _QWEN38
        from transformers import AutoModelForImageTextToText, AutoProcessor
        try:
            torch.set_num_threads(min(_parse_cpu_count() or 1, 16))
        except RuntimeError:
            pass
        processor = AutoProcessor.from_pretrained(_QWEN38_REPO)
        model = AutoModelForImageTextToText.from_pretrained(
            _QWEN38_REPO,
            torch_dtype=torch.bfloat16,
            low_cpu_mem_usage=True,
            device_map="cpu",
        )
        model.eval()
        _QWEN38_PROCESSOR = processor
        _QWEN38 = model
        return processor, model


def _generate_qwen38(prompt, max_new_tokens=256, temperature=0.7, thinking=True):
    processor, model = _load_qwen38()
    max_new_tokens = max(1, min(256, int(max_new_tokens)))
    temperature = max(0.0, min(2.0, float(temperature)))
    content = str(prompt).strip()
    messages = [{"role": "user", "content": [{"type": "text", "text": content}]}]
    try:
        text = processor.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True,
            enable_thinking=True,
        )
    except TypeError:
        text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = processor(text=[text], return_tensors="pt")
    inputs = {key: value.to("cpu") if hasattr(value, "to") else value for key, value in inputs.items()}
    tokenizer = processor.tokenizer
    eos_id = tokenizer.eos_token_id
    pad_id = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else eos_id
    input_tokens = int(inputs["input_ids"].shape[-1])
    t0 = time.time()
    with torch.inference_mode():
        output = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            do_sample=temperature > 0,
            temperature=max(temperature, 0.01),
            eos_token_id=eos_id,
            pad_token_id=pad_id,
            return_dict_in_generate=True,
            output_scores=True,
        )
    elapsed = max(time.time() - t0, 0.001)
    sequences = output.sequences
    generated = sequences[0, input_tokens:]
    eos_positions = (generated == eos_id).nonzero(as_tuple=True)[0].tolist() if eos_id is not None else []
    answer = tokenizer.decode(generated, skip_special_tokens=True).strip()
    return {
        "text": answer,
        "prompt_tokens": input_tokens,
        "completion_tokens": int(generated.shape[-1]),
        "generation_tokens_per_sec": round(float(generated.shape[-1]) / elapsed, 2),
        "inference_time_sec": round(elapsed, 3),
        "device": "cpu",
        "dtype": "bfloat16",
        "thinking": bool(thinking),
        "debug": {
            "rendered_chat_template": text,
            "input_ids": inputs["input_ids"][0].tolist(),
            "generated_token_ids": generated.tolist(),
            "eos_token_id": eos_id,
            "pad_token_id": pad_id,
            "eos_positions": eos_positions,
            "sequence_length": int(sequences.shape[-1]),
            "max_new_tokens": max_new_tokens,
            "scores_steps": len(output.scores) if output.scores is not None else 0,
            "finish_reason": "eos" if eos_positions else ("length" if generated.shape[-1] >= max_new_tokens else "unknown"),
        },
    }


def _load_tinyllama():
    global _TINYLLAMA, _TINYLLAMA_TOKENIZER
    with _TINYLLAMA_LOCK:
        if _TINYLLAMA is not None:
            return _TINYLLAMA_TOKENIZER, _TINYLLAMA
        from transformers import AutoModelForCausalLM, AutoTokenizer
        try:
            torch.set_num_threads(min(_parse_cpu_count() or 1, 16))
        except RuntimeError:
            # Gradio may already have started PyTorch work during app startup.
            pass
        tokenizer = AutoTokenizer.from_pretrained(_TINYLLAMA_REPO)
        model = AutoModelForCausalLM.from_pretrained(
            _TINYLLAMA_REPO,
            torch_dtype=torch.float32,
            low_cpu_mem_usage=True,
        )
        model.to("cpu")
        model.eval()
        _TINYLLAMA_TOKENIZER = tokenizer
        _TINYLLAMA = model
        return tokenizer, model


def _generate_tinyllama(prompt, max_new_tokens=64, temperature=0.7):
    tokenizer, model = _load_tinyllama()
    max_new_tokens = max(1, min(256, int(max_new_tokens)))
    temperature = max(0.0, min(2.0, float(temperature)))
    messages = [{"role": "user", "content": prompt}]
    if hasattr(tokenizer, "apply_chat_template"):
        templated = tokenizer.apply_chat_template(
            messages, add_generation_prompt=True, return_tensors="pt"
        )
        if isinstance(templated, dict) or hasattr(templated, "input_ids"):
            inputs = templated["input_ids"]
            attention_mask = templated.get("attention_mask", torch.ones_like(inputs))
        else:
            inputs = templated
            attention_mask = torch.ones_like(inputs)
    else:
        encoded = tokenizer(prompt, return_tensors="pt")
        inputs = encoded["input_ids"]
        attention_mask = encoded["attention_mask"]
    t0 = time.time()
    with torch.inference_mode():
        output = model.generate(
            input_ids=inputs,
            attention_mask=attention_mask,
            max_new_tokens=max_new_tokens,
            do_sample=temperature > 0,
            temperature=max(temperature, 0.01),
            pad_token_id=tokenizer.eos_token_id,
        )
    elapsed = max(time.time() - t0, 0.001)
    generated = output[0, inputs.shape[-1]:]
    text = tokenizer.decode(generated, skip_special_tokens=True)
    return {"text": text, "prompt_tokens": int(inputs.shape[-1]), "completion_tokens": int(generated.shape[-1]), "generation_tokens_per_sec": round(float(generated.shape[-1]) / elapsed, 2), "inference_time_sec": round(elapsed, 3), "device": "cpu"}


def load_model(name):
    if name not in MODEL_CACHE:
        MODEL_CACHE[name] = StableAudioModel.from_pretrained(name, device="cpu")
    return MODEL_CACHE[name]

def generate_audio(prompt, duration, steps, cfg_scale, seed, model_name):
    model = load_model(model_name)
    audio = model.generate(prompt=prompt, duration=duration, steps=steps, cfg_scale=cfg_scale, seed=seed, batch_size=1)
    audio = rearrange(audio, "b d n -> d (b n)")
    audio = audio.to(torch.float32).clamp(-1, 1).mul(32767).to(torch.int16).cpu()
    out = os.path.join(tempfile.gettempdir(), f"stable_audio_{seed}_{hash(prompt)&0xFFFFFFFF:08x}.wav")
    torchaudio.save(out, audio, 44100)
    return out


# ---------------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------------

with gr.Blocks(title="Respite API") as demo:
    gr.Markdown("# Respite API")
    with gr.Row():
        with gr.Column():
            model_name = gr.Dropdown(choices=["small-music","small-sfx"], value="small-music", label="Model")
            prompt = gr.Textbox(label="Prompt", placeholder="Describe the audio...", lines=2)
            duration = gr.Slider(minimum=1, maximum=120, value=30, step=1, label="Duration")
            steps = gr.Slider(minimum=1, maximum=50, value=8, step=1, label="Steps")
            cfg_scale = gr.Slider(minimum=0.0, maximum=10.0, value=1.0, step=0.1, label="CFG Scale")
            seed = gr.Number(value=-1, label="Seed (-1=random)")
            btn = gr.Button("Generate", variant="primary")
        with gr.Column():
            audio_output = gr.Audio(label="Generated Audio", type="filepath")
    btn.click(fn=generate_audio, inputs=[prompt, duration, steps, cfg_scale, seed, model_name], outputs=audio_output)


# ---------------------------------------------------------------------------
# Monkey-patch: intercept Gradio App creation to inject routes
# ---------------------------------------------------------------------------

from gradio.routes import App as GradioApp

_orig_create_app = GradioApp.create_app


def _patched_create_app(blocks, **kwargs):
    fa_app = _orig_create_app(blocks, **kwargs)

    @fa_app.get("/respite/resources")
    def _r(): return JSONResponse({"resources": API_RESOURCES})

    @fa_app.get("/respite/specs")
    def _s(): return JSONResponse({**API_SPECS, "runtime": _get_runtime_specs()})

    @fa_app.post("/respite/audio/generate")
    def _audio_api(payload: dict = Body(...)):
        prompt = str(payload.get("prompt", "")).strip()
        if not prompt:
            return JSONResponse({"error": "Missing 'prompt'"}, status_code=400)
        try:
            path = generate_audio(
                prompt,
                max(1, min(120, float(payload.get("duration", 30)))),
                max(1, min(50, int(payload.get("steps", 8)))),
                max(0.0, min(10.0, float(payload.get("cfg_scale", 1.0)))),
                int(payload.get("seed", -1)),
                str(payload.get("model", "small-music")),
            )
            from fastapi.responses import FileResponse
            return FileResponse(path, media_type="audio/wav", filename=os.path.basename(path))
        except Exception as e:
            return JSONResponse({"error": str(e)}, status_code=500)

    @fa_app.get("/respite/search")
    def _q(q: str="", backend: str="text", max_results: int=10, extract_top: int=0, region: str="wt-wt"):
        if not q: return JSONResponse({"error":"Missing 'q'"}, status_code=400)
        try: return JSONResponse(_search_web(q, backend=backend, max_results=max_results, extract_top=extract_top, region=region))
        except Exception as e: return JSONResponse({"error": str(e)}, status_code=502)

    @fa_app.post("/respite/text/qwen38")
    def _qwen38(payload: dict = Body(...)):
        prompt = str(payload.get("prompt", ""))
        if not prompt.strip():
            return JSONResponse({"error": "Missing 'prompt'"}, status_code=400)
        job = _start_benchmark_job(
            "qwen38",
            None,
            request={
                "prompt": prompt,
                "max_new_tokens": payload.get("max_new_tokens", 256),
                "temperature": payload.get("temperature", 0.7),
                "thinking": payload.get("thinking", True),
            },
        )
        return JSONResponse({"status": "accepted", "job_id": job, "poll": "/respite/text/qwen38?job_id=" + job}, status_code=202)

    @fa_app.get("/respite/text/qwen38")
    def _qwen38_status(job_id: str = ""):
        if not job_id:
            return JSONResponse({"error": "Missing 'job_id'"}, status_code=400)
        with _BENCH_JOBS_LOCK:
            return JSONResponse(_BENCH_JOBS.get(job_id, {"status": "not_found"}))

    @fa_app.post("/respite/text/generate")
    def _tinyllama(payload: dict = Body(...)):
        prompt = str(payload.get("prompt", ""))
        if not prompt.strip():
            return JSONResponse({"error": "Missing 'prompt'"}, status_code=400)
        try:
            return JSONResponse(_generate_tinyllama(
                prompt,
                payload.get("max_new_tokens", 64),
                payload.get("temperature", 0.7),
            ))
        except Exception as e:
            return JSONResponse({"error": str(e)}, status_code=500)

    @fa_app.get("/respite/benchmark/probe")
    def _probe(model: str="TinyLlama 1.1B (control)"):
        job = _start_benchmark_job("probe", model.strip())
        return JSONResponse({"status": "accepted", "job_id": job, "poll": "/respite/benchmark/diagnose?job_id=" + job}, status_code=202)

    @fa_app.get("/respite/benchmark/diagnose")
    def _diagnose(model: str="TinyLlama 1.1B (control)", job_id: str=""):
        if job_id:
            with _BENCH_JOBS_LOCK:
                return JSONResponse(_BENCH_JOBS.get(job_id, {"status": "not_found"}))
        job = _start_benchmark_job("diagnose", model.strip())
        return JSONResponse({"status": "accepted", "job_id": job, "poll": "/respite/benchmark/diagnose?job_id=" + job}, status_code=202)

    @fa_app.get("/respite/benchmark/smoke")
    def _smoke(model: str="Gemma 4 E4B", job_id: str=""):
        if job_id:
            with _BENCH_JOBS_LOCK:
                return JSONResponse(_BENCH_JOBS.get(job_id, {"status": "not_found"}))
        job = _start_benchmark_job("smoke", model.strip())
        return JSONResponse({"status": "accepted", "job_id": job, "poll": "/respite/benchmark/smoke?job_id=" + job}, status_code=202)

    @fa_app.get("/respite/benchmark")
    def _b(force: bool=False, model: str="", job_id: str=""):
        if job_id:
            with _BENCH_JOBS_LOCK:
                return JSONResponse(_BENCH_JOBS.get(job_id, {"status": "not_found"}))
        job = _start_benchmark_job("benchmark", model.strip() or None, force)
        return JSONResponse({"status": "accepted", "job_id": job, "poll": "/respite/benchmark?job_id=" + job}, status_code=202)

    # Node runtime capability probe + backend proxy
    try:
        node_probe.register_routes(fa_app)
        threading.Thread(target=node_runtime.start_node_backend, daemon=True).start()
    except Exception as _ne:
        _log(f"node probe registration failed: {_ne}")

    try:
        searx_bridge.register_routes(fa_app)
        threading.Thread(target=searx_runtime.start_searxng, daemon=True).start()
    except Exception as _se:
        _log(f"searxng bridge registration failed: {_se}")

    # ── Shared-secret gate for the unauthenticated compute routes ────────────
    # (no-op edit to force a rebuild: adding a Space secret does not restart a
    #  ZeroGPU space, and the restart endpoint is PRO-only, so the container
    #  keeps the env it booted with until something new is built.)
    # /respite/audio/generate, /respite/search, /respite/text/* and
    # /respite/benchmark* had no authentication of any kind: any process on the
    # internet could generate audio, run inference, and bill the Space owner's
    # audio key and ZeroGPU hours. CORS is irrelevant to them — a server-side
    # caller does not care what the browser enforces — so the boundary they
    # were missing is a shared secret, not a header policy.
    #
    # The secret is read from the Space's own `RESPIRE_SPACE_KEY` setting and is
    # never sent to a browser: the app holds it server-side and proxies through
    # `/api/audio`. If the setting is absent the routes fail CLOSED, because a
    # gate that opens when its key is missing is a gate that opens exactly when
    # it is not configured.
    try:
        import hmac as _hmac

        _SPACE_KEY = os.environ.get("RESpite_SPACE_KEY", "")
        # Which secret guards which prefix. The searx bridge gets its own key
        # (SEARX_API_KEY) so rotating search access never touches the audio
        # path; everything else shares RESPIRE_SPACE_KEY.
        _PROTECTED_PREFIXES = (
            "/respite/audio/",
            "/respite/search",
            "/respite/text/",
            "/respite/benchmark",
            # The searx bridge (searx_bridge.py) mounts under /respite/v2/ —
            # without a gate it is an open proxy: anyone on the internet can
            # run searches on the Space's IP for free.
            "/respite/v2/searx/",
        )

        def _expected_key(path: str) -> str:
            if path.startswith("/respite/v2/searx/"):
                return os.environ.get("SEARX_API_KEY", "")
            return os.environ.get("RESPIRE_SPACE_KEY", "")

        @fa_app.middleware("http")
        async def _respite_key_gate(request, call_next):
            if request.url.path.startswith(_PROTECTED_PREFIXES):
                _SPACE_KEY = _expected_key(request.url.path)
                if not _SPACE_KEY:
                    return JSONResponse(
                        {"error": "service not configured"}, status_code=503
                    )
                presented = request.headers.get("x-respite-key", "")
                if not _hmac.compare_digest(presented, _SPACE_KEY):
                    return JSONResponse({"error": "unauthorized"}, status_code=401)
            return await call_next(request)

        _log("compute routes key gate installed")
    except Exception as _ke:
        _log(f"compute routes key gate FAILED to install: {_ke}")

    # ── CORS gate for /respite/* ────────────────────────────────────────────
    # Gradio installs a global CORS middleware that reflects the caller's
    # Origin and sets `Access-Control-Allow-Credentials: true`. Applied to the
    # whole app, that turns every route we inject below into a cross-origin
    # readable endpoint — including the audio, search and inference routes,
    # which have no auth of their own. The per-route headers set in
    # node_probe.py are not enough on their own: this middleware runs outside
    # the route handlers and re-adds its own on the way out.
    #
    # So it is added here, after Gradio's (Starlette runs the most recently
    # added middleware outermost), and it rewrites the headers on the way out
    # of the inner stack. Scoped to /respite/* so the Gradio UI itself is left
    # alone.
    try:
        from node_probe import cors_headers as _respite_cors_headers

        @fa_app.middleware("http")
        async def _respite_cors_gate(request, call_next):
            response = await call_next(request)
            if request.url.path.startswith("/respite/"):
                for name in list(response.headers.keys()):
                    if name.lower().startswith("access-control-"):
                        del response.headers[name]
                for name, value in _respite_cors_headers(request).items():
                    response.headers[name] = value
            return response
    except Exception as _ce:
        _log(f"respite CORS gate not installed: {_ce}")

    _log("Routes injected via create_app monkey-patch")
    return fa_app


GradioApp.create_app = _patched_create_app

# Disable file watcher
import gradio.utils as _gr_utils
if hasattr(_gr_utils, "watchfn_spaces"):
    _gr_utils.watchfn_spaces = lambda *a, **kw: None

demo.queue(max_size=4, default_concurrency_limit=1)
demo.launch(ssr_mode=False)