Spaces:

DragonForgeStudios
/

forge-model

Running on Zero

Upload app.py with huggingface_hub

9e667e3 verified about 1 month ago

18.8 kB

"""Forge model server v2 — free HF ZeroGPU Gradio Space.
Design constraints (all verified empirically, not assumed):
  * Free accounts: creating Docker or cpu-basic Spaces returns HTTP 402. ZeroGPU
    Gradio Spaces are the only free compute. ZeroGPU REQUIRES a function
    decorated with @spaces.GPU to exist at startup, or the Space is killed with
    "No @spaces.GPU function detected during startup" — v1 died exactly there.
  * Free ZeroGPU quota is 5 min/day on an RTX Pro 6000 Blackwell (48 GB), so the
    GPU is a fast path, not the workhorse. Unlimited generations run on a CPU
    engine: llamafile (prebuilt Cosmopolitan binary, no compiler needed) serving
    a Q4_K_M GGUF on localhost:7861, proxied through this app.
  * Clients choose the engine with the model name: anything ending in "-gpu"
    takes the ZeroGPU path; everything else takes the unlimited CPU path.
  * Gradio 6's default SSR mode runs a Node.js proxy that owns the port;
    ssr_mode=False serves straight from Python, which is what lets the
    OpenAI-compatible /v1/* routes below live on the same port as the UI.
  * `import spaces` must come before any CUDA-touching import: gradio imports
    spaces lazily in a watcher thread, and if torch has already initialised
    CUDA that import raises and kills the app.
"""
import spaces  # noqa: F401  — import order is the point; keep this first

import errno
import functools
import json
import os
import shutil
import subprocess
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path

# ---------------------------------------------------------------- config ----

MODEL_ID = os.environ.get("MODEL_ID", "mlabonne/Qwen3-4B-abliterated")
GGUF_REPO = os.environ.get("GGUF_REPO", "mradermacher/Qwen3-4B-abliterated-GGUF")
GGUF_FILE = os.environ.get("GGUF_FILE", "Qwen3-4B-abliterated.Q4_K_M.gguf")
LLAMAFILE_URL = os.environ.get(
    "LLAMAFILE_URL",
    "https://github.com/mozilla-ai/llamafile/releases/download/0.10.5/llamafile-0.10.5-thin",
)
MODEL_ALIAS = os.environ.get("MODEL_ALIAS", "qwen3-4b-abliterated")
GPU_ALIAS = MODEL_ALIAS + "-gpu"
ENGINE_DIR = Path(os.environ.get("FORGE_ENGINE_DIR", str(Path.home() / "forge-engine")))
LLAMA_PORT = int(os.environ.get("LLAMA_PORT", "7861"))
PORT = int(os.environ.get("PORT", "7860"))
ENGINE_BOOT_WAIT = int(os.environ.get("ENGINE_BOOT_WAIT", "600"))  # cold download is ~2.6 GB
GPU_DURATION = int(os.environ.get("GPU_DURATION", "120"))

ENGINE_BIN = ENGINE_DIR / "llamafile"
ENGINE_URL = "http://127.0.0.1:%d" % LLAMA_PORT

_engine = {"proc": None, "ready": False, "err": None, "since": 0.0}
_engine_lock = threading.Lock()
_gpu_lock = threading.Lock()
_gpu = {"model": None, "tok": None, "err": None}


# ---------------------------------------------------------------- gpu path ----

def _gpu_load():
    """Lazy-load the transformers model. On ZeroGPU this runs inside a
    @spaces.GPU call, so real CUDA is present. Elsewhere it falls back to CPU
    (which is how this file is tested on a machine with no GPU at all).
    Plain bf16, deliberately: the card has 48 GB so there is no reason to
    quantize a 4B model, and nf4 was measurably dumber — it copied the
    system-prompt example verbatim instead of designing the requested level
    (verified 2026-09-12). The Q4_K_M GGUF on the CPU path does not do this."""
    if _gpu["model"] is not None:
        return _gpu["model"], _gpu["tok"]
    import torch
    from transformers import AutoModelForCausalLM, AutoTokenizer

    tok = AutoTokenizer.from_pretrained(MODEL_ID)
    device = "cuda" if torch.cuda.is_available() else "cpu"
    dtype = torch.bfloat16 if device == "cuda" else torch.float32
    try:
        model = AutoModelForCausalLM.from_pretrained(
            MODEL_ID, dtype=dtype, device_map=device,
        )
    except Exception as exc:  # older transformers: torch_dtype spelling
        print("[gpu] dtype load failed (%r); trying torch_dtype" % exc, flush=True)
        model = AutoModelForCausalLM.from_pretrained(
            MODEL_ID, torch_dtype=dtype, device_map=device,
        )
    model.eval()
    _gpu["model"], _gpu["tok"] = model, tok
    return model, tok


def _gpu_generate_inner(messages, max_tokens, temperature):
    import torch

    model, tok = _gpu_load()
    # Qwen3 hybrid thinking: disable it — GameForge needs bare JSON, no <think>.
    try:
        text = tok.apply_chat_template(
            messages, tokenize=False, add_generation_prompt=True,
            enable_thinking=False,
        )
    except TypeError:  # template without the kwarg
        text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = tok(text, return_tensors="pt").to(model.device)
    with torch.inference_mode():
        out = model.generate(
            **inputs,
            max_new_tokens=int(max_tokens),
            do_sample=float(temperature) > 0,
            temperature=max(float(temperature), 1e-4),
            top_p=0.95,
            pad_token_id=tok.eos_token_id,
        )
    return tok.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)


@spaces.GPU(duration=GPU_DURATION)
def gpu_generate(messages, max_tokens=1200, temperature=0.7):
    """ZeroGPU fast path. Quota: ~5 min/day on a free account. If the cached
    model references a stale CUDA context (GPU detached between calls), drop it
    and reload once before giving up."""
    with _gpu_lock:
        try:
            return _gpu_generate_inner(messages, max_tokens, temperature)
        except Exception as first:
            print("[gpu] generate failed (%r); reloading once" % first, flush=True)
            _gpu["model"] = _gpu["err"] = None
            try:
                return _gpu_generate_inner(messages, max_tokens, temperature)
            except Exception as second:
                _gpu["err"] = repr(second)
                raise


# ---------------------------------------------------------------- cpu engine ----

def _log(tag, msg):
    print("[%s] %s" % (tag, msg), flush=True)


def _download(url, dest: Path):
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".part")
    _log("engine", "downloading %s -> %s" % (url, dest))
    with urllib.request.urlopen(url, timeout=120) as r, open(tmp, "wb") as fh:
        shutil.copyfileobj(r, fh, 1 << 20)
    tmp.rename(dest)


def _engine_boot():
    """Download llamafile + GGUF (if missing), start llama-server on localhost,
    wait for /health. Runs in a background thread so the Gradio app binds its
    port immediately — ZeroGPU validation only needs the decorated function."""
    tag = "engine"
    try:
        with _engine_lock:
            if _engine["proc"] is not None and _engine["proc"].poll() is None:
                return
            # An engine from a previous app instance may still be bound (it is
            # daemonised, so it survives the parent). Reuse it if healthy.
            try:
                with urllib.request.urlopen(ENGINE_URL + "/health", timeout=2) as r:
                    if r.status == 200:
                        _engine["ready"] = True
                        _log(tag, "reusing engine already running on %s" % ENGINE_URL)
                        return
            except Exception:
                pass
            ENGINE_DIR.mkdir(parents=True, exist_ok=True)
            if not ENGINE_BIN.exists():
                _download(LLAMAFILE_URL, ENGINE_BIN)
                ENGINE_BIN.chmod(0o755)
            from huggingface_hub import hf_hub_download

            gguf = Path(hf_hub_download(repo_id=GGUF_REPO, filename=GGUF_FILE))
            threads = max(2, min(8, os.cpu_count() or 2))
            cmd = [
                str(ENGINE_BIN), "--model", str(gguf),
                "--server", "--host", "127.0.0.1", "--port", str(LLAMA_PORT),
                "--ctx-size", "6144", "--threads", str(threads),
                "--alias", MODEL_ALIAS, "--jinja", "--no-webui",
            ]
            _log(tag, "starting: %s" % " ".join(cmd))
            logfh = open(ENGINE_DIR / "llamafile.log", "ab")

            def _popen(argv):
                # APE binaries are hybrid ELF+shell. Some kernels refuse the
                # ELF header with ENOEXEC; the documented fallback is /bin/sh,
                # which the shell-script half of the binary understands.
                try:
                    return subprocess.Popen(
                        argv, cwd=str(ENGINE_DIR), stdout=logfh,
                        stderr=subprocess.STDOUT, start_new_session=True,
                    )
                except OSError as exc:
                    if exc.errno != errno.ENOEXEC:
                        raise
                    return subprocess.Popen(
                        ["/bin/sh", *argv], cwd=str(ENGINE_DIR), stdout=logfh,
                        stderr=subprocess.STDOUT, start_new_session=True,
                    )

            proc = _popen(cmd)
            _engine["proc"] = proc
        deadline = time.time() + 300
        while time.time() < deadline:
            if proc.poll() is not None:
                raise RuntimeError(
                    "llamafile exited rc=%s — see llamafile.log" % proc.returncode
                )
            try:
                with urllib.request.urlopen(ENGINE_URL + "/health", timeout=2) as r:
                    if r.status == 200:
                        _engine["ready"] = True
                        _engine["err"] = None
                        _log(tag, "ready on %s" % ENGINE_URL)
                        return
            except Exception:
                time.sleep(1.5)
        raise RuntimeError("llamafile /health timeout after 300s")
    except Exception as exc:
        _engine["err"] = repr(exc)
        _log(tag, "BOOT FAILED: %r" % exc)


def engine_wait(timeout=ENGINE_BOOT_WAIT):
    """Block until the CPU engine answers, so requests that arrive during a
    cold start wait instead of failing."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if _engine["ready"]:
            return True
        if _engine["err"] and _engine["proc"] is None:
            return False
        time.sleep(2)
    return False


def engine_chat(body):
    """POST an OpenAI-shaped body to the local llamafile. Qwen3 soft-switch:
    /no_think keeps the model from emitting <think> blocks into the JSON.
    Verified empirically: response_format json_schema is NOT honoured by this
    build (the grammar is not applied), so clients must not rely on it."""
    messages = body.get("messages") or []
    messages = list(messages)
    if messages and isinstance(messages[0], dict) and messages[0].get("role") == "system":
        if "/no_think" not in messages[0].get("content", ""):
            messages[0] = dict(messages[0])
            messages[0]["content"] = messages[0]["content"].rstrip() + "\n/no_think"
    else:
        # No system message: Qwen3's template defaults to thinking mode, which
        # lands in reasoning_content and leaves content empty. Suppress it.
        messages.insert(0, {"role": "system", "content": "/no_think"})
    # The soft switch is documented for the user turn, and some builds only
    # check the LAST user message — tag that too.
    for i in range(len(messages) - 1, -1, -1):
        if isinstance(messages[i], dict) and messages[i].get("role") == "user":
            if "/no_think" not in str(messages[i].get("content", "")):
                messages[i] = dict(messages[i])
                messages[i]["content"] = str(messages[i]["content"]).rstrip() + "\n/no_think"
            break
    body = dict(body, messages=messages, stream=False)
    # Harmless if this build ignores it; honoured by newer llama.cpp builds.
    body.setdefault("chat_template_kwargs", {"enable_thinking": False})
    if not body.get("model"):
        body["model"] = MODEL_ALIAS
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        ENGINE_URL + "/v1/chat/completions", data=data,
        headers={"Content-Type": "application/json"}, method="POST",
    )
    with urllib.request.urlopen(req, timeout=900) as r:
        return json.loads(r.read().decode("utf-8", "replace"))


# ---------------------------------------------------------------- ui ----

def status():
    gpu_state = "cached" if _gpu["model"] is not None else (
        "error: %s" % _gpu["err"] if _gpu["err"] else "cold"
    )
    if _engine["ready"]:
        eng = "cpu engine ready (%s)" % MODEL_ALIAS
    elif _engine["err"]:
        eng = "cpu engine ERROR: %s" % _engine["err"]
    else:
        eng = "cpu engine booting (cold start downloads ~2.6 GB)"
    return "%s | gpu(%s): %s | quota: ~5 min/day free" % (eng, GPU_ALIAS, gpu_state)


def ui_generate(system_prompt, user_prompt, max_tokens, use_gpu):
    try:
        if use_gpu:
            return gpu_generate(
                [{"role": "system", "content": system_prompt},
                 {"role": "user", "content": user_prompt}],
                int(max_tokens), 0.7,
            )
        if not engine_wait():
            return "engine not ready: %s" % (_engine["err"] or "timeout")
        out = engine_chat({
            "model": MODEL_ALIAS, "max_tokens": int(max_tokens), "temperature": 0.7,
            "messages": [{"role": "system", "content": system_prompt},
                         {"role": "user", "content": user_prompt}],
        })
        return out["choices"][0]["message"]["content"]
    except Exception as exc:
        return "error: %r" % exc


import gradio as gr  # noqa: E402

with gr.Blocks(title="Forge model server") as demo:
    gr.Markdown(
        "## Forge model server\n"
        "OpenAI-compatible backend for repo GameForge. Unlimited CPU engine "
        "(`%s`), ZeroGPU fast path (`%s`).\n\n"
        "Point an OpenAI client at `/v1` on this Space's URL." % (MODEL_ALIAS, GPU_ALIAS)
    )
    with gr.Row():
        sys_in = gr.Textbox(label="system", lines=5)
        usr_in = gr.Textbox(label="user", lines=5)
    with gr.Row():
        mt = gr.Slider(64, 2000, value=900, step=32, label="max tokens")
        gpu = gr.Checkbox(label="use ZeroGPU fast path (daily quota)")
    out = gr.Textbox(label="output", lines=16)
    gr.Button("generate").click(ui_generate, [sys_in, usr_in, mt, gpu], out, api_name="forge")
    st = gr.Textbox(label="status", value=status(), interactive=False)
    gr.Button("status").click(status, None, st, api_name="status")


# ---------------------------------------------------------------- routes ----

import asyncio  # noqa: E402
from fastapi import Request  # noqa: E402
from fastapi.responses import JSONResponse  # noqa: E402


def _openai_response(content, model):
    return {
        "id": "forge-%d" % int(time.time()),
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [{
            "index": 0,
            "message": {"role": "assistant", "content": content},
            "finish_reason": "stop",
        }],
        "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
    }


async def _chat_completions(request: Request):
    try:
        body = await request.json()
    except Exception:
        return JSONResponse({"error": {"message": "invalid JSON body"}}, status_code=400)
    if body.get("stream"):
        return JSONResponse(
            {"error": {"message": "streaming is not supported; send stream=false"}},
            status_code=400,
        )
    messages = body.get("messages")
    if not messages:
        return JSONResponse({"error": {"message": "messages required"}}, status_code=400)
    model = body.get("model") or MODEL_ALIAS
    max_tokens = int(body.get("max_tokens") or 1200)
    temperature = float(body.get("temperature") or 0.7)

    loop = asyncio.get_running_loop()

    if str(model).endswith("-gpu"):
        try:
            content = await loop.run_in_executor(
                None, functools.partial(gpu_generate, messages, max_tokens, temperature)
            )
            return JSONResponse(_openai_response(content, model))
        except Exception as exc:
            # Quota exhausted / GPU unavailable: 429 so clients fall back.
            return JSONResponse(
                {"error": {"message": "gpu path failed: %r" % exc}}, status_code=429
            )

    if not await loop.run_in_executor(None, engine_wait):
        return JSONResponse(
            {"error": {
                "message": "cpu engine not ready: %s" % (_engine["err"] or "cold start timeout")
            }},
            status_code=503,
        )
    try:
        out = await loop.run_in_executor(None, functools.partial(engine_chat, body))
        return JSONResponse(out)
    except urllib.error.HTTPError as exc:
        try:
            detail = json.loads(exc.read().decode("utf-8", "replace"))
        except Exception:
            detail = {"error": {"message": "engine HTTP %d" % exc.code}}
        return JSONResponse(detail, status_code=exc.code)
    except Exception as exc:
        return JSONResponse(
            {"error": {"message": "engine error: %r" % exc}}, status_code=500
        )


async def _models():
    loop = asyncio.get_running_loop()
    if not await loop.run_in_executor(None, functools.partial(engine_wait, 30)):
        return JSONResponse({"object": "list", "data": [
            {"id": MODEL_ALIAS, "object": "model"},
            {"id": GPU_ALIAS, "object": "model"},
        ]})
    try:
        with urllib.request.urlopen(ENGINE_URL + "/v1/models", timeout=20) as r:
            return JSONResponse(json.loads(r.read().decode("utf-8", "replace")))
    except Exception:
        return JSONResponse(
            {"object": "list", "data": [{"id": MODEL_ALIAS, "object": "model"}]}
        )


async def _healthz():
    return JSONResponse({"ok": True, "status": status()})


_launch_result = demo.launch(
    server_name="0.0.0.0", server_port=PORT,
    ssr_mode=False, prevent_thread_lock=True, quiet=True, show_error=True,
)
_fastapi = _launch_result[0] if isinstance(_launch_result, tuple) else None
if _fastapi is None:
    raise RuntimeError("gradio 6 launch() did not return the FastAPI app")

# Plain `def` endpoints: FastAPI runs them in its threadpool, so the blocking
# engine_wait()/generation calls never stall the event loop.
_fastapi.add_api_route("/v1/chat/completions", _chat_completions, methods=["POST"])
_fastapi.add_api_route("/v1/models", _models, methods=["GET"])
_fastapi.add_api_route("/healthz", _healthz, methods=["GET"])

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

if __name__ == "__main__":
    _log("app", "serving on :%d (engine on :%d)" % (PORT, LLAMA_PORT))
    demo.block_thread()

	"""Forge model server v2 — free HF ZeroGPU Gradio Space.

	Design constraints (all verified empirically, not assumed):
	* Free accounts: creating Docker or cpu-basic Spaces returns HTTP 402. ZeroGPU
	Gradio Spaces are the only free compute. ZeroGPU REQUIRES a function
	decorated with @spaces.GPU to exist at startup, or the Space is killed with
	"No @spaces.GPU function detected during startup" — v1 died exactly there.
	* Free ZeroGPU quota is 5 min/day on an RTX Pro 6000 Blackwell (48 GB), so the
	GPU is a fast path, not the workhorse. Unlimited generations run on a CPU
	engine: llamafile (prebuilt Cosmopolitan binary, no compiler needed) serving
	a Q4_K_M GGUF on localhost:7861, proxied through this app.
	* Clients choose the engine with the model name: anything ending in "-gpu"
	takes the ZeroGPU path; everything else takes the unlimited CPU path.
	* Gradio 6's default SSR mode runs a Node.js proxy that owns the port;
	ssr_mode=False serves straight from Python, which is what lets the
	OpenAI-compatible /v1/* routes below live on the same port as the UI.
	* `import spaces` must come before any CUDA-touching import: gradio imports
	spaces lazily in a watcher thread, and if torch has already initialised
	CUDA that import raises and kills the app.
	"""
	import spaces # noqa: F401 — import order is the point; keep this first

	import errno
	import functools
	import json
	import os
	import shutil
	import subprocess
	import threading
	import time
	import urllib.error
	import urllib.request
	from pathlib import Path

	# ---------------------------------------------------------------- config ----

	MODEL_ID = os.environ.get("MODEL_ID", "mlabonne/Qwen3-4B-abliterated")
	GGUF_REPO = os.environ.get("GGUF_REPO", "mradermacher/Qwen3-4B-abliterated-GGUF")
	GGUF_FILE = os.environ.get("GGUF_FILE", "Qwen3-4B-abliterated.Q4_K_M.gguf")
	LLAMAFILE_URL = os.environ.get(
	"LLAMAFILE_URL",
	"https://github.com/mozilla-ai/llamafile/releases/download/0.10.5/llamafile-0.10.5-thin",
	)
	MODEL_ALIAS = os.environ.get("MODEL_ALIAS", "qwen3-4b-abliterated")
	GPU_ALIAS = MODEL_ALIAS + "-gpu"
	ENGINE_DIR = Path(os.environ.get("FORGE_ENGINE_DIR", str(Path.home() / "forge-engine")))
	LLAMA_PORT = int(os.environ.get("LLAMA_PORT", "7861"))
	PORT = int(os.environ.get("PORT", "7860"))
	ENGINE_BOOT_WAIT = int(os.environ.get("ENGINE_BOOT_WAIT", "600")) # cold download is ~2.6 GB
	GPU_DURATION = int(os.environ.get("GPU_DURATION", "120"))

	ENGINE_BIN = ENGINE_DIR / "llamafile"
	ENGINE_URL = "http://127.0.0.1:%d" % LLAMA_PORT

	_engine = {"proc": None, "ready": False, "err": None, "since": 0.0}
	_engine_lock = threading.Lock()
	_gpu_lock = threading.Lock()
	_gpu = {"model": None, "tok": None, "err": None}


	# ---------------------------------------------------------------- gpu path ----

	def _gpu_load():
	"""Lazy-load the transformers model. On ZeroGPU this runs inside a
	@spaces.GPU call, so real CUDA is present. Elsewhere it falls back to CPU
	(which is how this file is tested on a machine with no GPU at all).

	Plain bf16, deliberately: the card has 48 GB so there is no reason to
	quantize a 4B model, and nf4 was measurably dumber — it copied the
	system-prompt example verbatim instead of designing the requested level
	(verified 2026-09-12). The Q4_K_M GGUF on the CPU path does not do this."""
	if _gpu["model"] is not None:
	return _gpu["model"], _gpu["tok"]
	import torch
	from transformers import AutoModelForCausalLM, AutoTokenizer

	tok = AutoTokenizer.from_pretrained(MODEL_ID)
	device = "cuda" if torch.cuda.is_available() else "cpu"
	dtype = torch.bfloat16 if device == "cuda" else torch.float32
	try:
	model = AutoModelForCausalLM.from_pretrained(
	MODEL_ID, dtype=dtype, device_map=device,
	)
	except Exception as exc: # older transformers: torch_dtype spelling
	print("[gpu] dtype load failed (%r); trying torch_dtype" % exc, flush=True)
	model = AutoModelForCausalLM.from_pretrained(
	MODEL_ID, torch_dtype=dtype, device_map=device,
	)
	model.eval()
	_gpu["model"], _gpu["tok"] = model, tok
	return model, tok


	def _gpu_generate_inner(messages, max_tokens, temperature):
	import torch

	model, tok = _gpu_load()
	# Qwen3 hybrid thinking: disable it — GameForge needs bare JSON, no <think>.
	try:
	text = tok.apply_chat_template(
	messages, tokenize=False, add_generation_prompt=True,
	enable_thinking=False,
	)
	except TypeError: # template without the kwarg
	text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
	inputs = tok(text, return_tensors="pt").to(model.device)
	with torch.inference_mode():
	out = model.generate(
	**inputs,
	max_new_tokens=int(max_tokens),
	do_sample=float(temperature) > 0,
	temperature=max(float(temperature), 1e-4),
	top_p=0.95,
	pad_token_id=tok.eos_token_id,
	)
	return tok.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)


	@spaces.GPU(duration=GPU_DURATION)
	def gpu_generate(messages, max_tokens=1200, temperature=0.7):
	"""ZeroGPU fast path. Quota: ~5 min/day on a free account. If the cached
	model references a stale CUDA context (GPU detached between calls), drop it
	and reload once before giving up."""
	with _gpu_lock:
	try:
	return _gpu_generate_inner(messages, max_tokens, temperature)
	except Exception as first:
	print("[gpu] generate failed (%r); reloading once" % first, flush=True)
	_gpu["model"] = _gpu["err"] = None
	try:
	return _gpu_generate_inner(messages, max_tokens, temperature)
	except Exception as second:
	_gpu["err"] = repr(second)
	raise


	# ---------------------------------------------------------------- cpu engine ----

	def _log(tag, msg):
	print("[%s] %s" % (tag, msg), flush=True)


	def _download(url, dest: Path):
	dest.parent.mkdir(parents=True, exist_ok=True)
	tmp = dest.with_suffix(dest.suffix + ".part")
	_log("engine", "downloading %s -> %s" % (url, dest))
	with urllib.request.urlopen(url, timeout=120) as r, open(tmp, "wb") as fh:
	shutil.copyfileobj(r, fh, 1 << 20)
	tmp.rename(dest)


	def _engine_boot():
	"""Download llamafile + GGUF (if missing), start llama-server on localhost,
	wait for /health. Runs in a background thread so the Gradio app binds its
	port immediately — ZeroGPU validation only needs the decorated function."""
	tag = "engine"
	try:
	with _engine_lock:
	if _engine["proc"] is not None and _engine["proc"].poll() is None:
	return
	# An engine from a previous app instance may still be bound (it is
	# daemonised, so it survives the parent). Reuse it if healthy.
	try:
	with urllib.request.urlopen(ENGINE_URL + "/health", timeout=2) as r:
	if r.status == 200:
	_engine["ready"] = True
	_log(tag, "reusing engine already running on %s" % ENGINE_URL)
	return
	except Exception:
	pass
	ENGINE_DIR.mkdir(parents=True, exist_ok=True)
	if not ENGINE_BIN.exists():
	_download(LLAMAFILE_URL, ENGINE_BIN)
	ENGINE_BIN.chmod(0o755)
	from huggingface_hub import hf_hub_download

	gguf = Path(hf_hub_download(repo_id=GGUF_REPO, filename=GGUF_FILE))
	threads = max(2, min(8, os.cpu_count() or 2))
	cmd = [
	str(ENGINE_BIN), "--model", str(gguf),
	"--server", "--host", "127.0.0.1", "--port", str(LLAMA_PORT),
	"--ctx-size", "6144", "--threads", str(threads),
	"--alias", MODEL_ALIAS, "--jinja", "--no-webui",
	]
	_log(tag, "starting: %s" % " ".join(cmd))
	logfh = open(ENGINE_DIR / "llamafile.log", "ab")

	def _popen(argv):
	# APE binaries are hybrid ELF+shell. Some kernels refuse the
	# ELF header with ENOEXEC; the documented fallback is /bin/sh,
	# which the shell-script half of the binary understands.
	try:
	return subprocess.Popen(
	argv, cwd=str(ENGINE_DIR), stdout=logfh,
	stderr=subprocess.STDOUT, start_new_session=True,
	)
	except OSError as exc:
	if exc.errno != errno.ENOEXEC:
	raise
	return subprocess.Popen(
	["/bin/sh", *argv], cwd=str(ENGINE_DIR), stdout=logfh,
	stderr=subprocess.STDOUT, start_new_session=True,
	)

	proc = _popen(cmd)
	_engine["proc"] = proc
	deadline = time.time() + 300
	while time.time() < deadline:
	if proc.poll() is not None:
	raise RuntimeError(
	"llamafile exited rc=%s — see llamafile.log" % proc.returncode
	)
	try:
	with urllib.request.urlopen(ENGINE_URL + "/health", timeout=2) as r:
	if r.status == 200:
	_engine["ready"] = True
	_engine["err"] = None
	_log(tag, "ready on %s" % ENGINE_URL)
	return
	except Exception:
	time.sleep(1.5)
	raise RuntimeError("llamafile /health timeout after 300s")
	except Exception as exc:
	_engine["err"] = repr(exc)
	_log(tag, "BOOT FAILED: %r" % exc)


	def engine_wait(timeout=ENGINE_BOOT_WAIT):
	"""Block until the CPU engine answers, so requests that arrive during a
	cold start wait instead of failing."""
	deadline = time.time() + timeout
	while time.time() < deadline:
	if _engine["ready"]:
	return True
	if _engine["err"] and _engine["proc"] is None:
	return False
	time.sleep(2)
	return False


	def engine_chat(body):
	"""POST an OpenAI-shaped body to the local llamafile. Qwen3 soft-switch:
	/no_think keeps the model from emitting <think> blocks into the JSON.
	Verified empirically: response_format json_schema is NOT honoured by this
	build (the grammar is not applied), so clients must not rely on it."""
	messages = body.get("messages") or []
	messages = list(messages)
	if messages and isinstance(messages[0], dict) and messages[0].get("role") == "system":
	if "/no_think" not in messages[0].get("content", ""):
	messages[0] = dict(messages[0])
	messages[0]["content"] = messages[0]["content"].rstrip() + "\n/no_think"
	else:
	# No system message: Qwen3's template defaults to thinking mode, which
	# lands in reasoning_content and leaves content empty. Suppress it.
	messages.insert(0, {"role": "system", "content": "/no_think"})
	# The soft switch is documented for the user turn, and some builds only
	# check the LAST user message — tag that too.
	for i in range(len(messages) - 1, -1, -1):
	if isinstance(messages[i], dict) and messages[i].get("role") == "user":
	if "/no_think" not in str(messages[i].get("content", "")):
	messages[i] = dict(messages[i])
	messages[i]["content"] = str(messages[i]["content"]).rstrip() + "\n/no_think"
	break
	body = dict(body, messages=messages, stream=False)
	# Harmless if this build ignores it; honoured by newer llama.cpp builds.
	body.setdefault("chat_template_kwargs", {"enable_thinking": False})
	if not body.get("model"):
	body["model"] = MODEL_ALIAS
	data = json.dumps(body).encode("utf-8")
	req = urllib.request.Request(
	ENGINE_URL + "/v1/chat/completions", data=data,
	headers={"Content-Type": "application/json"}, method="POST",
	)
	with urllib.request.urlopen(req, timeout=900) as r:
	return json.loads(r.read().decode("utf-8", "replace"))


	# ---------------------------------------------------------------- ui ----

	def status():
	gpu_state = "cached" if _gpu["model"] is not None else (
	"error: %s" % _gpu["err"] if _gpu["err"] else "cold"
	)
	if _engine["ready"]:
	eng = "cpu engine ready (%s)" % MODEL_ALIAS
	elif _engine["err"]:
	eng = "cpu engine ERROR: %s" % _engine["err"]
	else:
	eng = "cpu engine booting (cold start downloads ~2.6 GB)"
	return "%s \| gpu(%s): %s \| quota: ~5 min/day free" % (eng, GPU_ALIAS, gpu_state)


	def ui_generate(system_prompt, user_prompt, max_tokens, use_gpu):
	try:
	if use_gpu:
	return gpu_generate(
	[{"role": "system", "content": system_prompt},
	{"role": "user", "content": user_prompt}],
	int(max_tokens), 0.7,
	)
	if not engine_wait():
	return "engine not ready: %s" % (_engine["err"] or "timeout")
	out = engine_chat({
	"model": MODEL_ALIAS, "max_tokens": int(max_tokens), "temperature": 0.7,
	"messages": [{"role": "system", "content": system_prompt},
	{"role": "user", "content": user_prompt}],
	})
	return out["choices"][0]["message"]["content"]
	except Exception as exc:
	return "error: %r" % exc


	import gradio as gr # noqa: E402

	with gr.Blocks(title="Forge model server") as demo:
	gr.Markdown(
	"## Forge model server\n"
	"OpenAI-compatible backend for repo GameForge. Unlimited CPU engine "
	"(`%s`), ZeroGPU fast path (`%s`).\n\n"
	"Point an OpenAI client at `/v1` on this Space's URL." % (MODEL_ALIAS, GPU_ALIAS)
	)
	with gr.Row():
	sys_in = gr.Textbox(label="system", lines=5)
	usr_in = gr.Textbox(label="user", lines=5)
	with gr.Row():
	mt = gr.Slider(64, 2000, value=900, step=32, label="max tokens")
	gpu = gr.Checkbox(label="use ZeroGPU fast path (daily quota)")
	out = gr.Textbox(label="output", lines=16)
	gr.Button("generate").click(ui_generate, [sys_in, usr_in, mt, gpu], out, api_name="forge")
	st = gr.Textbox(label="status", value=status(), interactive=False)
	gr.Button("status").click(status, None, st, api_name="status")


	# ---------------------------------------------------------------- routes ----

	import asyncio # noqa: E402
	from fastapi import Request # noqa: E402
	from fastapi.responses import JSONResponse # noqa: E402


	def _openai_response(content, model):
	return {
	"id": "forge-%d" % int(time.time()),
	"object": "chat.completion",
	"created": int(time.time()),
	"model": model,
	"choices": [{
	"index": 0,
	"message": {"role": "assistant", "content": content},
	"finish_reason": "stop",
	}],
	"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
	}


	async def _chat_completions(request: Request):
	try:
	body = await request.json()
	except Exception:
	return JSONResponse({"error": {"message": "invalid JSON body"}}, status_code=400)
	if body.get("stream"):
	return JSONResponse(
	{"error": {"message": "streaming is not supported; send stream=false"}},
	status_code=400,
	)
	messages = body.get("messages")
	if not messages:
	return JSONResponse({"error": {"message": "messages required"}}, status_code=400)
	model = body.get("model") or MODEL_ALIAS
	max_tokens = int(body.get("max_tokens") or 1200)
	temperature = float(body.get("temperature") or 0.7)

	loop = asyncio.get_running_loop()

	if str(model).endswith("-gpu"):
	try:
	content = await loop.run_in_executor(
	None, functools.partial(gpu_generate, messages, max_tokens, temperature)
	)
	return JSONResponse(_openai_response(content, model))
	except Exception as exc:
	# Quota exhausted / GPU unavailable: 429 so clients fall back.
	return JSONResponse(
	{"error": {"message": "gpu path failed: %r" % exc}}, status_code=429
	)

	if not await loop.run_in_executor(None, engine_wait):
	return JSONResponse(
	{"error": {
	"message": "cpu engine not ready: %s" % (_engine["err"] or "cold start timeout")
	}},
	status_code=503,
	)
	try:
	out = await loop.run_in_executor(None, functools.partial(engine_chat, body))
	return JSONResponse(out)
	except urllib.error.HTTPError as exc:
	try:
	detail = json.loads(exc.read().decode("utf-8", "replace"))
	except Exception:
	detail = {"error": {"message": "engine HTTP %d" % exc.code}}
	return JSONResponse(detail, status_code=exc.code)
	except Exception as exc:
	return JSONResponse(
	{"error": {"message": "engine error: %r" % exc}}, status_code=500
	)


	async def _models():
	loop = asyncio.get_running_loop()
	if not await loop.run_in_executor(None, functools.partial(engine_wait, 30)):
	return JSONResponse({"object": "list", "data": [
	{"id": MODEL_ALIAS, "object": "model"},
	{"id": GPU_ALIAS, "object": "model"},
	]})
	try:
	with urllib.request.urlopen(ENGINE_URL + "/v1/models", timeout=20) as r:
	return JSONResponse(json.loads(r.read().decode("utf-8", "replace")))
	except Exception:
	return JSONResponse(
	{"object": "list", "data": [{"id": MODEL_ALIAS, "object": "model"}]}
	)


	async def _healthz():
	return JSONResponse({"ok": True, "status": status()})


	_launch_result = demo.launch(
	server_name="0.0.0.0", server_port=PORT,
	ssr_mode=False, prevent_thread_lock=True, quiet=True, show_error=True,
	)
	_fastapi = _launch_result[0] if isinstance(_launch_result, tuple) else None
	if _fastapi is None:
	raise RuntimeError("gradio 6 launch() did not return the FastAPI app")

	# Plain `def` endpoints: FastAPI runs them in its threadpool, so the blocking
	# engine_wait()/generation calls never stall the event loop.
	_fastapi.add_api_route("/v1/chat/completions", _chat_completions, methods=["POST"])
	_fastapi.add_api_route("/v1/models", _models, methods=["GET"])
	_fastapi.add_api_route("/healthz", _healthz, methods=["GET"])

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

	if __name__ == "__main__":
	_log("app", "serving on :%d (engine on :%d)" % (PORT, LLAMA_PORT))
	demo.block_thread()