app.py

"""
Inkling — a chat UI served by gradio.Server.

Backend: gradio.Server (FastAPI + Gradio's queue/concurrency engine).
Model:   thinkingmachines/Inkling via Hugging Face Inference Providers,
         reached through the OpenAI-compatible router at
         https://router.huggingface.co/v1.
Frontend: a hand-written single-page chat client (index.html) that talks to
          the backend through the Gradio JS client, so every request goes
          through Gradio's queue — including the token stream.
"""

import os
import sys
import traceback
from pathlib import Path
from typing import Iterator

from fastapi.responses import HTMLResponse
from gradio import Server
from openai import OpenAI


# Python 3.13 + uvicorn + asyncio selectors raise a harmless
# "Invalid file descriptor" during interpreter shutdown because the
# BaseEventLoop's GC __del__ runs after uvicorn already closed its
# selector. The exception is only meaningful for live processes being
# reaped; on Space shutdown there's nothing to do. Silence it so the
# Space logs stay readable.
def _quiet_unraisable(unraisable_args) -> None:
    exc_type, exc_value = unraisable_args.exc_type, unraisable_args.exc_value
    if isinstance(exc_value, ValueError) and "Invalid file descriptor" in str(exc_value):
        return
    traceback.print_exception(
        unraisable_args.exc_type, unraisable_args.exc_value,
        unraisable_args.exc_traceback,
    )


sys.unraisablehook = _quiet_unraisable

# Hugging Face's OpenAI-compatible router. The ":<provider>" suffix on the
# model name tells the router how to route the request.
ROUTER_BASE_URL = "https://router.huggingface.co/v1"
DEFAULT_MODEL = "thinkingmachines/Inkling"

# Provider routing suffixes supported by the HF router. ":auto" uses the
# preferred provider from the user's HF settings; ":cheapest" and ":fastest"
# pick by cost or latency.
PROVIDERS = {
    "auto": ":auto",
    "cheapest": ":cheapest",
    "fastest": ":fastest",
}

app = Server()

# A single shared OpenAI client is fine — the underlying httpx client is
# thread-safe and connection-pooled. We build it lazily so the app boots even
# before HF_TOKEN is present (Hugging Face Spaces injects it at runtime).
_client: OpenAI | None = None


def get_client() -> OpenAI:
    """Return a cached OpenAI client pointed at the HF router."""
    global _client
    token = os.environ.get("HF_TOKEN")
    if not token:
        raise RuntimeError(
            "HF_TOKEN is not set. Add it as a Space secret "
            "(Settings → Repository secrets → New secret)."
        )
    if _client is None:
        _client = OpenAI(base_url=ROUTER_BASE_URL, api_key=token)
    return _client


def resolve_model(provider: str) -> str:
    """Turn a short provider key into a fully-suffixed model id."""
    suffix = PROVIDERS.get(provider, PROVIDERS["auto"])
    return f"{DEFAULT_MODEL}{suffix}"


@app.api(stream_every=0)
def chat(
    messages: list[dict],
    provider: str = "auto",
    system_prompt: str = "",
    temperature: float = 0.7,
    max_tokens: int = 2048,
) -> Iterator[str]:
    """Stream an assistant reply from Inkling, yielding accumulated text.

    ``messages`` is the conversation so far, each item ``{"role", "content"}``
    with role "user" or "assistant". Each yield is the full text accumulated
    so far, so the frontend can simply replace the bubble contents on every
    event.
    """
    # Compose the final message list, prepending a system prompt if given.
    full: list[dict] = []
    if system_prompt and system_prompt.strip():
        full.append({"role": "system", "content": system_prompt.strip()})
    full.extend(messages)

    try:
        client = get_client()
    except Exception as exc:  # pragma: no cover - surfaced to the user
        yield f"⚠️ {exc}"
        return

    try:
        stream = client.chat.completions.create(
            model=resolve_model(provider),
            messages=full,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=True,
        )
    except Exception as exc:  # pragma: no cover - surfaced to the user
        yield f"⚠️ Request failed: {exc}"
        return

    accumulated = ""
    for chunk in stream:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta.content
        if delta:
            accumulated += delta
            yield accumulated

    if not accumulated:
        yield "(The model returned no content.)"


@app.get("/config")
async def config():
    """Small endpoint the frontend can poll for model/provider metadata."""
    return {
        "model": DEFAULT_MODEL,
        "providers": list(PROVIDERS.keys()),
        "default_provider": "auto",
    }


@app.get("/", response_class=HTMLResponse)
async def homepage():
    """Serve the single-page chat client."""
    html_path = Path(__file__).resolve().parent / "index.html"
    return HTMLResponse(html_path.read_text(encoding="utf-8"))


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