app.py

"""Ravenmoor — gr.Server backend for the god-view town frontend.

The custom HTML5-canvas frontend (frontend/) renders a living top-down town:
NPCs walk their own schedules, the camera follows whoever you choose, and the
detective only you command. These endpoints are the authoritative game state.
"""

from __future__ import annotations

import os
import threading
import time
from pathlib import Path

from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from gradio import Server

ROOT = Path(__file__).resolve().parent
if not os.environ.get("CARTBRAWL_API_KEY"):
    _kf = ROOT / ".cartbrawl_api_key"
    if _kf.exists():
        os.environ["CARTBRAWL_API_KEY"] = _kf.read_text().strip()

from ravenmoor import npc as npcmod
from concurrent.futures import TimeoutError as FuturesTimeout

from ravenmoor import scores, weaver
from ravenmoor.casegen import LOCATIONS, MOTIVES, TOWN_FACTS
from ravenmoor.flavor import generate_flavor
from ravenmoor.game import (
    COST_ASK,
    COST_CONFRONT,
    COST_SEARCH,
    COST_SKILL,
    COST_STATEMENT,
    SKILLS,
    Game,
    new_game,
)

MAX_SESSIONS = 60
_sessions: dict[str, tuple[Game, float]] = {}
_lock = threading.Lock()


def _warm_backend():
    """Poke the Modal endpoint at startup so the first player skips the cold start."""
    try:
        import httpx

        base = os.environ.get("CARTBRAWL_LLM_URL", "https://wuzhenze--cartbrawl-llm-serve.modal.run/v1")
        httpx.get(
            base.rstrip("/") + "/models",
            headers={"Authorization": f"Bearer {os.environ.get('CARTBRAWL_API_KEY', '')}"},
            timeout=600,
        )
    except Exception:
        pass


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


def _get(sid: str) -> Game | None:
    with _lock:
        row = _sessions.get((sid or "anon")[:64])
        return row[0] if row else None


def _put(sid: str, g: Game):
    sid = (sid or "anon")[:64]
    with _lock:
        if len(_sessions) >= MAX_SESSIONS and sid not in _sessions:
            oldest = min(_sessions, key=lambda k: _sessions[k][1])
            del _sessions[oldest]
        _sessions[sid] = (g, time.time())


def render_state(g: Game) -> dict:
    lang = g.lang
    return {
        "lang": lang,
        "day": g.day,
        "clock": g.clock_min,
        "over": g.over,
        "won": g.won,
        "npcs": [
            {
                "id": n.id, "name": n.name, "avatar": n.avatar,
                "role": n.role[lang], "persona": n.persona[lang],
                "trait": n.trait[lang], "look": n.look[lang], "story": n.story[lang], "base": n.base,
                "exposed": g.exposed.get(n.id, []),
                "secret_out": n.id in g.secret_out,
                "statement_taken": n.id in g.statements,
                "attrs": n.attrs,
                "skills_used": {s: g.skill_used.get(f"{n.id}:{s}") == g.day for s in SKILLS},
                **g.behavior(n.id),
            }
            for n in g.case.npcs.values()
        ],
        "cards": [{"id": c["id"], "kind": c["kind"], "text": c["text"][lang]} for c in g.cards],
        "facts": {k: v[lang] for k, v in g.case.town_facts.items()},
        "started_at": g.started_at,
        "scene_loc": g.case.scene_loc,
        "town_name": g.case.town_name[lang],
        "layout": g.case.layout,
        "locations": {k: {"name": v[lang], "emoji": LOCATIONS[k]["emoji"]} for k, v in g.case.locations.items()},
        "motives": [
            {"key": k, "label": v[lang].split(":" if lang == "zh" else ":")[0]}
            for k, v in MOTIVES.items()
        ],
        "clues": [c["desc"][lang] for c in g.case.clues if c["id"] in g.clues_found],
        "searched": list(g.searched),
        "events": [e[lang] for e in g.events[-6:]],
        "victim_name": g.case.victim.get("name", ""),
        "intro": g.flavor.get("intro", ""),
    }


app = Server()


@app.api(name="new_game", concurrency_limit=8, stream_every=0.1)
def api_new_game(sid: str, lang: str) -> dict:
    """Streams weaving progress (a cold backend can take 1-2 min), retries once,
    and never falls back silently — the player is told when the classic cast is used."""
    lang = "zh" if str(lang).startswith("zh") or str(lang).startswith("中") else "en"
    g = new_game(lang)
    woven = None
    elapsed = 0
    for attempt in range(2):
        ex, fut = weaver.weave_async(g.case, lang)
        budget = 150 if attempt == 0 else 60
        waited = 0.0
        try:
            while waited < budget:
                try:
                    data = fut.result(timeout=3)
                    woven = data if weaver._valid(data, g.case) else None
                    break
                except FuturesTimeout:
                    waited += 3
                    elapsed += 3
                    yield {"kind": "progress", "seconds": elapsed}
                except Exception as e:
                    print(f"[weaver] attempt {attempt} error: {type(e).__name__}", flush=True)
                    break
        finally:
            ex.shutdown(wait=False, cancel_futures=True)
        if woven:
            print(f"[weaver] ok attempt {attempt} ~{elapsed}s", flush=True)
            break
    if woven:
        g.flavor = weaver.apply(g.case, woven, lang)
    else:
        print("[weaver] fell back to handwritten cast", flush=True)
        g.flavor = generate_flavor(g.case, lang)   # handwritten cast stays
    g.events.append({"day": 1, "zh": g.flavor["intro"], "en": g.flavor["intro"]})
    _put(sid, g)
    yield {"kind": "state", "woven": bool(woven), "state": render_state(g)}


@app.api(name="state", concurrency_limit=8)
def api_state(sid: str) -> dict:
    g = _get(sid)
    if not g:
        return {"kind": "error", "error": "no_game"}
    return {"kind": "state", "state": render_state(g)}


@app.api(name="ask", concurrency_limit=8, stream_every=0.1)
def api_ask(sid: str, npc_id: str, text: str) -> dict:
    """Streams the NPC's reply token by token, then the updated state.

    Annotated `-> dict` so gr.Server registers ONE streamed output; without it
    Gradio infers zero outputs and silently drops every yielded event.
    """
    g = _get(sid)
    text = (text or "").strip()[:300]
    if not g or g.over or npc_id not in g.case.npcs or not text:
        yield {"kind": "error", "error": "bad_request"}
        return
    hist = g.histories.setdefault(npc_id, [])
    hist.append({"role": "detective", "text": text})
    flags = {"secret_out": npc_id in g.secret_out, "exposed": g.exposed.get(npc_id, [])}
    answer = ""
    try:
        for delta in npcmod.reply(g.case, npc_id, hist[:-1], text, g.lang, flags):
            answer += delta
            yield {"kind": "delta", "text": delta}
    except Exception:
        fallback = "……(他沉默地看着你)" if g.lang == "zh" else "…(they stare at you in silence)"
        answer = answer or fallback
        yield {"kind": "delta", "text": fallback if not answer else ""}
    hist.append({"role": "npc", "text": answer})
    behavior = g.register_pressure(npc_id, text, answer)
    rolled = g.spend_time(COST_ASK)
    out = {"kind": "done", "behavior": behavior, "rollover": rolled, "state": render_state(g)}
    if g.over:
        out["reveal"] = g.reveal()
    yield out


@app.api(name="statement", concurrency_limit=8, stream_every=0.1)
def api_statement(sid: str, npc_id: str) -> dict:
    """Formal statement: structured evidence cards + the NPC reciting their
    evening in character (streamed)."""
    g = _get(sid)
    if not g or g.over or npc_id not in g.case.npcs:
        yield {"kind": "error", "error": "bad_request"}
        return
    if npc_id in g.statements:
        yield {"kind": "error", "error": "taken"}
        return
    new = g.take_statement(npc_id)
    yield {
        "kind": "cards",
        "cards": [{"id": c["id"], "kind": c["kind"], "text": c["text"][g.lang]} for c in new],
    }
    time.sleep(0.25)
    flags = {"secret_out": npc_id in g.secret_out, "exposed": g.exposed.get(npc_id, [])}
    narration = ""
    try:
        for delta in npcmod.statement_narration(g.case, npc_id, g.lang, flags):
            narration += delta
            yield {"kind": "delta", "text": delta}
    except Exception:
        pass
    hist = g.histories.setdefault(npc_id, [])
    hist.append({"role": "detective", "text": "【索取口供】" if g.lang == "zh" else "[takes formal statement]"})
    hist.append({"role": "npc", "text": narration})
    rolled = g.spend_time(COST_STATEMENT)
    out = {"kind": "done", "rollover": rolled, "state": render_state(g)}
    if g.over:
        out["reveal"] = g.reveal()
    yield out


@app.api(name="confront", concurrency_limit=8, stream_every=0.1)
def api_confront(sid: str, npc_id: str, card_id: str) -> dict:
    """Present a card to an NPC's face (⚡1). The truth graph decides whether
    it's a real contradiction; the LLM only acts out the verdict."""
    g = _get(sid)
    if not g or g.over or npc_id not in g.case.npcs:
        yield {"kind": "error", "error": "bad_request"}
        return
    card = next((c for c in g.cards if c["id"] == card_id), None)
    if not card:
        yield {"kind": "error", "error": "no_card"}
        return
    result = g.confront_check(npc_id, card_id)
    n = g.case.npcs[npc_id]
    if result.get("break") == "alibi":
        g.events.append({
            "day": g.day,
            "zh": f"❗ {n.name}的不在场证明已被铁证全面戳穿!",
            "en": f"❗ {n.name}'s alibi has completely collapsed under hard evidence!",
        })
    elif result.get("break") == "secret":
        g.events.append({
            "day": g.day,
            "zh": f"💔 {n.name}崩溃了,坦白了一桩与命案无关的秘密。",
            "en": f"💔 {n.name} broke down and confessed a secret — unrelated to the murder.",
        })
    verdict_ev = {
        "kind": "verdict", "valid": result["valid"], "break": result.get("break"),
        "refuter": result.get("refuter"), "mode": result.get("mode"),
        "card": card["text"][g.lang],
    }
    yield verdict_ev
    time.sleep(0.25)
    card_text = card["text"][g.lang]
    if result.get("refuter"):
        card_text += (
            f"(可{result['refuter']}当时整段时间都在那里,却从没见过你)" if g.lang == "zh"
            else f" (yet {result['refuter']} was there the whole time and never saw you)"
        )
    hist = g.histories.setdefault(npc_id, [])
    hist.append({"role": "detective", "text": ("【出示证据】" if g.lang == "zh" else "[presents evidence] ") + card_text})
    flags = {"secret_out": npc_id in g.secret_out, "exposed": g.exposed.get(npc_id, [])}
    answer = ""
    try:
        for delta in npcmod.confront_reply(g.case, npc_id, card_text, result, g.lang, flags):
            answer += delta
            yield {"kind": "delta", "text": delta}
    except Exception:
        pass
    hist.append({"role": "npc", "text": answer})
    rolled = g.spend_time(COST_CONFRONT)
    out = {"kind": "done", "rollover": rolled, "verdict": verdict_ev, "state": render_state(g)}
    if g.over:
        out["reveal"] = g.reveal()
    yield out


@app.api(name="skill", concurrency_limit=8, stream_every=0.1)
def api_skill(sid: str, npc_id: str, skill: str) -> dict:
    """Detective skill attempt: server rolls 2d6 vs the resisting attribute;
    success effects come straight from the truth graph."""
    g = _get(sid)
    if not g or g.over or npc_id not in g.case.npcs or skill not in SKILLS:
        yield {"kind": "error", "error": "bad_request"}
        return
    result = g.skill_check(npc_id, skill)
    if result.get("error"):
        yield {"kind": "error", "error": result["error"]}
        return
    # the dice line is formatted server-side and shipped as FLAT fields only:
    # the browser gradio client mangles nested arrays in streamed json diffs
    ATTR_LABELS = {"nerve": {"zh": "胆量", "en": "Nerve"},
                   "guard": {"zh": "戒心", "en": "Guard"},
                   "poise": {"zh": "沉稳", "en": "Poise"}}
    attr_label = ATTR_LABELS[result["attr"]][g.lang]
    verdict_txt = ("✅ 成功!" if result["success"] else "❌ 失败") if g.lang == "zh"         else ("✅ Success!" if result["success"] else "❌ Failed")
    dice_text = (f"🎲 {result['dice'][0]}+{result['dice'][1]}={result['roll']} "
                 f"vs {result['dc']}({attr_label}{result['attr_value']}+4) → {verdict_txt}")
    card = (result.get("payload") or {}).get("card")
    card_text = card["text"][g.lang] if card else ""
    dice_ev = {"kind": "dice", "text": dice_text, "success": bool(result["success"]),
               "effect": result.get("effect") or "", "card_text": card_text}
    yield dice_ev
    time.sleep(0.25)   # let the SSE consumer receive the roll as its own chunk
    hist = g.histories.setdefault(npc_id, [])
    label = {
        "intimidate": ("【严刑逼问】", "[intimidates]"),
        "befriend": ("【套近乎】", "[buys a drink]"),
        "bluff": ("【诈唬】「我们已经全都知道了」", '[bluffs] "We already know everything"'),
    }[skill]
    hist.append({"role": "detective", "text": label[0] if g.lang == "zh" else label[1]})
    flags = {"secret_out": npc_id in g.secret_out, "exposed": g.exposed.get(npc_id, [])}
    answer = ""
    try:
        for delta in npcmod.skill_reply(g.case, npc_id, skill, result, g.lang, flags):
            answer += delta
            yield {"kind": "delta", "text": delta}
    except Exception:
        pass
    if not answer and not result["success"]:
        # the LLM died before reacting and nothing happened: refund the daily use
        g.skill_used.pop(f"{npc_id}:{skill}", None)
        hist.pop()
        yield {"kind": "error", "error": "llm_down"}
        return
    hist.append({"role": "npc", "text": answer})
    rolled = g.spend_time(COST_SKILL)
    out = {"kind": "done", "rollover": rolled, "state": render_state(g),
           "dice_text": dice_text, "dice_success": dice_ev["success"],
           "dice_effect": dice_ev["effect"], "dice_card": card_text}
    if g.over:
        out["reveal"] = g.reveal()
    yield out


@app.api(name="pin", concurrency_limit=8)
def api_pin(sid: str, npc_id: str, text: str) -> dict:
    """Pin any line an NPC said in free chat as a quotable evidence card. Free."""
    g = _get(sid)
    text = (text or "").strip()[:200]
    if not g or g.over or npc_id not in g.case.npcs or not text:
        return {"kind": "error", "error": "bad_request"}
    n = g.case.npcs[npc_id]
    cid = f"quote:{npc_id}:{abs(hash(text)) % 10**8}"
    card = {
        "id": cid, "kind": "quote", "witness": npc_id, "subject": None,
        "text": {
            "zh": f"📌 {n.name}亲口说:「{text}」",
            "en": f'📌 {n.name} said to your face: "{text}"',
        },
    }
    g._add_card(card)
    return {"kind": "pinned", "state": render_state(g), "card_text": card["text"][g.lang]}


@app.api(name="leaderboard", concurrency_limit=8)
def api_leaderboard(sid: str) -> dict:
    return {"kind": "board", "rows": scores.top(15)}


@app.api(name="submit_score", concurrency_limit=8)
def api_submit_score(sid: str, name: str) -> dict:
    g = _get(sid)
    if not g or not g.won or g.solve_seconds is None:
        return {"kind": "error", "error": "no_win"}
    if g.score_submitted:
        return {"kind": "error", "error": "already"}
    try:
        rows = scores.submit(name, g.solve_seconds, min(g.day, 3), g.clock_used(), g.lang, g.case.seed)
        g.score_submitted = True
        return {"kind": "board", "rows": rows}
    except Exception as e:
        return {"kind": "error", "error": f"submit_failed: {e}"}


@app.api(name="search", concurrency_limit=8)
def api_search(sid: str, loc: str) -> dict:
    g = _get(sid)
    if not g or g.over or loc not in LOCATIONS:
        return {"kind": "error", "error": "bad_request"}
    if loc in g.searched:
        return {"kind": "error", "error": "searched"}
    found = g.search(loc)
    texts = []
    if loc == g.case.scene_loc and g.flavor.get("scene"):
        texts.append("🏰 " + g.flavor["scene"])
        g.events.append({"day": g.day, "zh": "🏰 " + g.flavor["scene"], "en": "🏰 " + g.flavor["scene"]})
    for c in found:
        texts.append("🧩 " + c["desc"][g.lang])
        g.events.append({"day": g.day, "zh": "🧩 " + c["desc"]["zh"], "en": "🧩 " + c["desc"]["en"]})
    if not texts:
        msg = "搜查完毕,没有发现新东西。" if g.lang == "zh" else "You search thoroughly and find nothing new."
        texts.append(msg)
    rolled = g.spend_time(COST_SEARCH)
    out = {"kind": "found", "texts": texts, "rollover": rolled, "state": render_state(g)}
    if g.over:
        out["reveal"] = g.reveal()
    return out


@app.api(name="end_day", concurrency_limit=8)
def api_end_day(sid: str) -> dict:
    g = _get(sid)
    if not g or g.over:
        return {"kind": "error", "error": "bad_request"}
    g.end_day()
    out = {"kind": "day", "state": render_state(g)}
    if g.over:  # culprit fled
        out["reveal"] = g.reveal()
    return out


@app.api(name="accuse", concurrency_limit=8, stream_every=0.1)
def api_accuse(sid: str, npc_id: str, motive: str) -> dict:
    """Streams the verdict, then the culprit's final speech."""
    g = _get(sid)
    if not g or g.over or npc_id not in g.case.npcs or motive not in MOTIVES:
        yield {"kind": "error", "error": "bad_request"}
        return
    result = g.accuse(npc_id, motive)
    culprit = g.case.npcs[g.case.culprit]
    yield {
        "kind": "verdict",
        "correct_npc": result["correct_npc"],
        "correct_motive": result["correct_motive"],
        "reveal": result["reveal"],
        "solve_seconds": g.solve_seconds,
        "culprit": {"id": culprit.id, "name": culprit.name, "avatar": culprit.avatar},
        "state": render_state(g),
    }
    time.sleep(0.25)
    try:
        for delta in npcmod.final_speech(g.case, npc_id, g.lang):
            yield {"kind": "delta", "text": delta}
    except Exception:
        pass
    try:
        for delta in npcmod.epilogue_narration(g.case, result["reveal"], g.lang):
            yield {"kind": "epi_delta", "text": delta}
    except Exception:
        pass
    yield {"kind": "done", "state": render_state(g)}


_last_warm = {"t": 0.0}


@app.get("/", response_class=HTMLResponse)
async def homepage():
    # start waking the inference backend while the player reads the title screen
    if time.time() - _last_warm["t"] > 300:
        _last_warm["t"] = time.time()
        threading.Thread(target=_warm_backend, daemon=True).start()
    return (ROOT / "frontend" / "index.html").read_text(encoding="utf-8")


class FreshStatic(StaticFiles):
    """Small files, always revalidated — UI updates reach every browser."""

    def file_response(self, *args, **kwargs):
        resp = super().file_response(*args, **kwargs)
        resp.headers["Cache-Control"] = "no-cache, must-revalidate"
        return resp


app.mount("/static", FreshStatic(directory=ROOT / "frontend" / "static"), name="static")

if __name__ == "__main__":
    app.launch(show_error=True)