"""
app.py — Mumbai Local Panic · HF ZeroGPU Space entrypoint (Gradio Server mode).

Mirrors geekwrestler/her-trace: ZeroGPU is Gradio-SDK-only, so this is a single root app.py that
builds `app = gr.Server()` (FastAPI + Gradio's engine), defines the routes, and calls
`app.launch()` as the LAST module-level statement (the import/launch IS the server; a separate
uvicorn or an __main__-guarded launch fails under HF's gradio runner).

  * Deterministic routes the React SPA calls with `fetch`:  POST /turn · POST /model · GET /sprites
  * The dispatcher ("lord nemo", NVIDIA Nemotron-3-Nano) runs via the transformers @spaces.GPU
    backend (frontend/server/gpu_llm.py); VoxCPM2 voice via GET /announcement_audio.
  * The built React+PixiJS SPA (frontend/dist) is served at /.
  * Sprites/audio come from the private HF dataset bucket (MLP_ASSETS_REPO), downloaded at startup.

NOTE (next step): a plain `fetch` that triggers @spaces.GPU does NOT forward the HF iframe auth
headers ZeroGPU needs — /turn (model on) and /announcement_audio should become Gradio API
endpoints (@app.api) called via @gradio/client. This file gets the Space RUNNING first.
"""
import json
import os
import sys
import threading

os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")

_ROOT = os.path.dirname(os.path.abspath(__file__))
for _p in (os.path.join(_ROOT, "backend"), os.path.join(_ROOT, "frontend", "server")):
    if _p not in sys.path:
        sys.path.insert(0, _p)

# Register a @spaces.GPU function at MODULE level. This wires up the ZeroGPU runtime, which is what
# keeps the Space process RESIDENT under HF's gradio reload (without it the process exits right after
# launch -> "Invalid file descriptor: -1" -> RUNTIME_ERROR). her-trace gets this for free via its
# module-level model load; we do it explicitly so staying-up doesn't depend on the (mamba-gated)
# Nemotron load. The function is never called — only the decoration matters.
try:
    import spaces
    import torch as _torch

    @spaces.GPU(duration=15)
    def _zerogpu_keepalive():
        return _torch.zeros(1, device="cuda")
except Exception as _e:
    print(f"[app] zerogpu keepalive hook unavailable: {_e}")

# Load the REAL dispatcher (NVIDIA Nemotron-3-Nano) onto the GPU at module level — registers its
# @spaces.GPU _generate and warms the model (nemotron_h needs mamba-ssm). Best-effort: if it fails,
# the trivial keepalive above still satisfies ZeroGPU's "@spaces.GPU at startup" rule and the first
# turn lazy-loads the model.
if os.environ.get("MLP_LLM_BACKEND") == "transformers":
    try:
        import gpu_llm  # noqa: F401
        print("[app] dispatcher warm — Nemotron on the GPU.")
    except Exception as _e:
        print(f"[app] dispatcher warmup failed (lazy-load on first turn): {_e}")

# Warm VoxCPM2 voice at module level too. It MUST load at startup (not via a lazy `import tts` inside
# a request handler): outside the import-time window ZeroGPU's CUDA emulation isn't active, so
# VoxCPM's cache init hits "Low-level CUDA init … not intercepted". Best-effort — a failure just
# disables voice; the dispatcher (and the rest of the game) is unaffected.
if os.environ.get("MLP_TTS", "off").lower() == "on":
    try:
        import tts  # noqa: F401
        print("[app] VoxCPM2 voice warm.")
    except Exception as _e:
        print(f"[app] VoxCPM2 warmup failed (voice disabled this run): {_e}")

import gradio as gr
import jsonschema
from fastapi import Request
from fastapi.responses import FileResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles

from session import GameSession, MODELS  # noqa: E402  (reuses the canon engine unchanged)
from api_routes import register_game_api  # noqa: E402  (the @app.api Gradio endpoints, ZeroGPU-auth)

_DIST = os.path.join(_ROOT, "frontend", "dist")
_SCHEMA_DIR = os.path.join(_ROOT, "claude-handoff", "schemas")
with open(os.path.join(_SCHEMA_DIR, "turn_request.schema.json"), encoding="utf-8") as _f:
    _turn_validator = jsonschema.Draft7Validator(json.load(_f))

# Sprites/audio are BUNDLED in the Space repo and served from disk. (A runtime snapshot_download of
# the bucket spun up an async transfer whose event-loop teardown closed a file descriptor the
# gradio/SSR server was using -> "Invalid file descriptor: -1" / "Stopping Node.js server" ->
# RUNTIME_ERROR. her-trace avoids this by mounting its bucket as a VOLUME, not downloading it.)
_SPRITES = os.environ.get("MLP_SPRITES_DIR") or os.path.join(_ROOT, "sprites")

_TTS_ON = os.environ.get("MLP_TTS", "off").lower() == "on"
_tts_cache: dict = {}
_tts_guard = threading.Lock()


class _Entry:
    __slots__ = ("session", "lock")

    def __init__(self, session):
        self.session = session
        self.lock = threading.Lock()


_sessions: dict = {}
_sessions_guard = threading.Lock()


def _entry(sid):
    with _sessions_guard:
        if sid not in _sessions:
            _sessions[sid] = _Entry(GameSession(seed=None))
        return _sessions[sid]


def _sid(request):
    return request.headers.get("X-Session-Id", "default")


app = gr.Server()


# --------------------------------------------------------------------------- game routes
@app.post("/turn")
async def turn(request: Request):
    try:
        body = await request.json()
    except Exception:
        return JSONResponse({"error": "body is not valid JSON"}, status_code=400)
    errors = sorted(_turn_validator.iter_errors(body), key=str)
    if errors:
        return JSONResponse({"error": f"invalid TurnRequest: {errors[0].message}"}, status_code=400)
    ent = _entry(_sid(request))
    with ent.lock:
        action = body["action"]
        if action == "new_game":
            old = ent.session
            ent.session = GameSession(seed=body.get("seed"), model_name=old.model_name,
                                      model_on=old.model_on)
            return JSONResponse(ent.session.last_state)
        if action == "play_card":
            ok, reason = ent.session.play_card(body["card"], body["location"])
            if not ok:
                return JSONResponse({"error": reason}, status_code=409)
            return JSONResponse(ent.session.last_state)
        return JSONResponse(ent.session.next_turn())


@app.post("/model")
async def model(request: Request):
    try:
        body = await request.json()
    except Exception:
        return JSONResponse({"error": "body is not valid JSON"}, status_code=400)
    if not isinstance(body, dict):
        return JSONResponse({"error": "body must be an object"}, status_code=400)
    if "on" in body and not isinstance(body["on"], bool):
        return JSONResponse({"error": "'on' must be a boolean"}, status_code=400)
    if "model" in body and body["model"] not in MODELS:
        return JSONResponse({"error": f"unknown model; available: {sorted(MODELS)}"}, status_code=400)
    ent = _entry(_sid(request))
    with ent.lock:
        if "model" in body:
            ent.session.set_model(body["model"])
        if "on" in body:
            ent.session.model_on = body["on"]
        return JSONResponse({"on": ent.session.model_on, "model": ent.session.model_name,
                             "models": sorted(MODELS), "tts": _TTS_ON})


@app.get("/announcement_audio")
def announcement_audio(request: Request):
    if not _TTS_ON:
        return Response(status_code=404)
    sid = request.query_params.get("sid") or _sid(request)
    ent = _sessions.get(sid)
    if ent is None:
        return Response(status_code=404)
    st = ent.session.last_state
    rnd = st.get("round")
    ann = ((st.get("ai") or {}).get("announcement") or "").strip()
    if not ann:
        return Response(status_code=204)
    key = (sid, rnd, ann)
    with _tts_guard:
        cached = _tts_cache.get(key)
    if cached is None:
        import tts
        voice = "male" if (rnd or 0) % 2 == 0 else "female"
        try:
            cached = tts.announce(ann, voice)
        except Exception as e:
            return JSONResponse({"error": f"tts failed: {e}"}, status_code=500)
        with _tts_guard:
            _tts_cache[key] = cached
            while len(_tts_cache) > 64:
                _tts_cache.pop(next(iter(_tts_cache)))
    return Response(content=cached, media_type="audio/wav")


# Gradio API endpoints (@app.api) — the frontend calls these via @gradio/client so the HF iframe
# auth forwards for ZeroGPU (the GPU dispatcher + VoxCPM run here). The plain routes above stay for
# direct/curl testing, but the SPA uses the Gradio API path.
register_game_api(app, _entry, _turn_validator, _TTS_ON)


# --------------------------------------------------------------------------- static (EXPLICIT, no
# catch-all: gr.Server registers /gradio_api/* + /config at launch; a wildcard would shadow them).
app.mount("/sprites", StaticFiles(directory=_SPRITES), name="sprites")
_ASSETS_DIR = os.path.join(_DIST, "assets")
if os.path.isdir(_ASSETS_DIR):
    app.mount("/assets", StaticFiles(directory=_ASSETS_DIR), name="assets")


@app.get("/")
def _index():
    idx = os.path.join(_DIST, "index.html")
    if os.path.isfile(idx):
        return FileResponse(idx)
    return JSONResponse({"error": "frontend not built — cd frontend && npm run build"}, status_code=503)


# --------------------------------------------------------------------------- launch (LAST statement)
# EXACTLY like her-trace: a plain blocking launch with SSR on (default). The "GRADIO_HOT_RELOAD …
# demo not found" log is benign (her-trace prints it too and runs fine). Do NOT pass ssr_mode=False
# — HF's runtime expects the SSR Node proxy on 7860; disabling it is what crashed the Space.
app.launch(
    server_name="0.0.0.0",
    server_port=int(os.environ.get("PORT", os.environ.get("GRADIO_SERVER_PORT", "7860"))),
    show_error=True,
)