app.py

"""
PARRY — compliant Gradio hosting layer (C8). 100% LOCAL inference by design.

Primary posture (per the verified research): `gradio.Server` (Gradio 6.x) on
`sdk: gradio` — a FastAPI subclass running Gradio's engine, with our custom
canvas served at GET / (custom routes take priority over the default UI; this
is the official `ysharma/text-behind-image` pattern, and the custom-UI award
text says "gr.Server is your friend").

NO external inference calls: the model runs in the visitor's browser via
WebGPU. (Organizer guidance, June 10: external API fallbacks are hard to
review — "the video is the fallback". The earlier Modal relay was removed.)

Defensive: if the pinned Gradio somehow lacks `Server`, we degrade to
FastAPI + gr.mount_gradio_app so the Space still boots while we consult the
hedge ladder (docker swap / gr.Blocks+gr.HTML — see ../space/app_blocks.py).

Endpoints
  GET  /              → static/index.html (the game)
  GET  /static/*      → built Vite bundle (immutable assets)
  api  "trace_digest" → validates + summarizes an exported BrainTrace (Gradio queue; gradio_client-callable)
  api  "about"        → build/model/grammar info via the queue
  POST /funnel        → enum-validated JSONL beacons (+ optional CommitScheduler → private HF Dataset)
  GET  /healthz       → {ok, version, grammar_id}

Secrets (Space settings): HF_TOKEN (funnel dataset, optional),
FUNNEL_DATASET (e.g. "user/parry-funnel", optional).
"""

from __future__ import annotations

import json
import os
import threading
import time
from pathlib import Path

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

HERE = Path(__file__).parent
STATIC = HERE / "static"
DATA = HERE / "data"
DATA.mkdir(exist_ok=True)
FUNNEL_PATH = DATA / "funnel.jsonl"

APP_VERSION = "parry-space-v1"
GRAMMAR_ID = os.environ.get("GRAMMAR_ID", "unset")  # injected by CI from grammar/hash

FUNNEL_EVENTS = {
    "page_load",
    "webgpu_detected",
    "webgpu_missing",
    "tier_selected",
    "model_download_started",
    "model_download_complete",
    "first_playable",
    "match_started",
    "match_completed",
    "fell_back_to_server",
    "bounced_during_download",
    "debug_override_used",
    "trace_exported",
    "boss_reflexes_on",
    "pose_mode_on",
}

_funnel_lock = threading.Lock()

# Optional: persist funnel JSONL to a private HF Dataset (ephemeral disk survival).
_scheduler = None
if os.environ.get("HF_TOKEN") and os.environ.get("FUNNEL_DATASET"):
    try:
        from huggingface_hub import CommitScheduler

        _scheduler = CommitScheduler(
            repo_id=os.environ["FUNNEL_DATASET"],
            repo_type="dataset",
            folder_path=str(DATA),
            every=5,  # minutes
            private=True,
        )
    except Exception as e:  # noqa: BLE001 — funnel persistence is best-effort
        print(f"[funnel] CommitScheduler unavailable: {e}")

def _append_funnel(row: dict) -> None:
    with _funnel_lock:
        with open(FUNNEL_PATH, "a", encoding="utf-8") as f:
            f.write(json.dumps(row, separators=(",", ":")) + "\n")
    print(f"[funnel] {row.get('event')}")


def _about() -> dict:
    return {
        "app": "parry",
        "version": APP_VERSION,
        "grammar_id": GRAMMAR_ID,
        "hero_model": "Qwen2.5-1.5B-Instruct (q4f16_1, in-browser WebGPU)",
        "fallback": "none — inference is 100% in-browser by design",
        "tuned_model": "Jainamshahhh/parry-tactician-1.5b-lora (published fine-tune)",
        "thesis": "a sub-100ms reaction loop no network round-trip can serve",
        "artifacts": {
            "fine_tune_merged": "https://huggingface.co/Jainamshahhh/parry-tactician-1.5b-merged",
            "fine_tune_lora": "https://huggingface.co/Jainamshahhh/parry-tactician-1.5b-lora",
            "gguf_llama_cpp": "https://huggingface.co/Jainamshahhh/parry-tactician-1.5b-gguf",
            "field_notes": "https://huggingface.co/datasets/Jainamshahhh/parry-field-notes",
            "agent_traces": "https://huggingface.co/datasets/Jainamshahhh/parry-traces",
            "badge_evidence": "see README.md / BADGES.md in this Space repo",
        },
    }


def _trace_digest(trace_json: str) -> dict:
    """Pure-local BrainTrace summarizer (no model, no external calls): validates
    an exported trace from the judge panel and returns its headline stats —
    companion utility for the shared-trace (Sharing-is-Caring) flow."""
    try:
        rows = json.loads(trace_json)
        assert isinstance(rows, list) and rows, "trace must be a non-empty JSON array"
    except Exception as e:  # noqa: BLE001
        return {"valid": False, "error": str(e)[:200]}
    intents: dict[str, int] = {}
    plans: list[str] = []
    for r in rows:
        if isinstance(r, dict):
            i = r.get("intent")
            if isinstance(i, str):
                intents[i] = intents.get(i, 0) + 1
            p = r.get("planString")
            if isinstance(p, str) and (not plans or plans[-1] != p):
                plans.append(p)
    return {
        "valid": True,
        "ticks": len(rows),
        "intent_histogram": intents,
        "distinct_plans": len(plans),
        "plan_timeline": plans[:12],
        "grammar_id": GRAMMAR_ID,
    }


HAS_SERVER = hasattr(gr, "Server")

if HAS_SERVER:
    app = gr.Server()
else:  # degrade gracefully; primary fix is pinning sdk_version per README
    print("[parry] WARNING: gradio.Server missing — booting FastAPI + mounted Blocks hedge")
    from fastapi import FastAPI

    app = FastAPI()

app.mount("/static", StaticFiles(directory=str(STATIC)), name="static")
# index.html is served at "/" with relative asset URLs → they resolve to /assets/*
app.mount("/assets", StaticFiles(directory=str(STATIC / "assets")), name="assets")
# pose-mode model + WASM are self-hosted under /pose (the worker fetches them
# by absolute origin URL — an unmounted path returns HTML 404s that die as
# "parse error" in TFJS, the same class of bug as the /assets mount above)
app.mount("/pose", StaticFiles(directory=str(STATIC / "pose")), name="pose")


@app.get("/")
async def homepage() -> FileResponse:
    # no-cache on the HTML only; hashed assets under /static are long-cached
    return FileResponse(STATIC / "index.html", headers={"Cache-Control": "no-cache"})


@app.post("/funnel")
async def funnel(req: Request) -> JSONResponse:
    try:
        row = await req.json()
    except Exception:  # noqa: BLE001 — sendBeacon may post opaque bodies
        return JSONResponse({"ok": False}, status_code=400)
    if row.get("event") not in FUNNEL_EVENTS:
        return JSONResponse({"ok": False, "error": "unknown event"}, status_code=400)
    _append_funnel({k: row.get(k) for k in ("v", "event", "session", "ts", "props")})
    return JSONResponse({"ok": True}, status_code=200)


@app.get("/healthz")
async def healthz() -> JSONResponse:
    return JSONResponse({"ok": True, "version": APP_VERSION, "grammar_id": GRAMMAR_ID, "inference": "in-browser"})


if HAS_SERVER:
    # Gradio-queue endpoints: Gradio's engine genuinely doing work (queue, SSE),
    # callable via gradio_client. Both are pure-local — no model, no external calls.
    @app.api(name="trace_digest")
    def trace_digest_api(trace_json: str) -> dict:
        return _trace_digest(trace_json)

    @app.api(name="about")
    def about_api() -> dict:
        return _about()

else:
    # Hedge boot: mount a minimal Blocks app so Gradio is still in the loop.
    with gr.Blocks() as demo:
        gr.Markdown("# Parry backend (hedge boot)")
        inp = gr.Textbox(label="exported BrainTrace JSON")
        out = gr.JSON(label="trace digest")
        gr.Button("digest").click(_trace_digest, inp, out)
    app = gr.mount_gradio_app(app, demo, path="/gradio")


if __name__ == "__main__":
    if HAS_SERVER:
        app.launch(show_error=True)
    else:
        import uvicorn

        uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 7860)))