feat: gradio app

README.md

@@ -1,3 +1,16 @@
+---
+title: God Simulator
+colorFrom: green
+colorTo: blue
+sdk: gradio
+sdk_version: 6.17.3
+python_version: 3.11
+app_file: app.py
+header: mini
+fullWidth: true
+startup_duration_timeout: 1h
+---
+
 # God Simulator
 
 Python-first foundation for an AI civilization sandbox.
@@ -49,6 +62,33 @@ Controls:
 - scroll to zoom;
 - use `WASD` or arrow keys to pan across the terrain.
 
+## Hugging Face Space
+
+This repo keeps the custom React/Three UI and wraps the same game runtime in
+`gradio.Server` via `app.py`.
+
+Build the custom UI before launching the Space app:
+
+```powershell
+npm install
+npm run gradio
+```
+
+The Gradio server serves the built UI from `dist/frontend` and exposes the same
+game endpoints:
+
+- `GET /health`
+- `GET /state`
+- `GET /claw3d/state`
+- `POST /tick`
+- `POST /god-command`
+
+It also registers Gradio API endpoints named `/tick` and `/god_command`, so the
+app still runs through Gradio's API engine while preserving the custom UI. For
+Hugging Face Spaces, `requirements.txt` installs the local project from
+`pyproject.toml`; set `GOD_SIMULATOR_CONFIG` if the Space should use a config
+other than `config/game.modal.local.json`.
+
 ## LLM Connector
 
 The default runtime config is `config/game.modal.local.json`, which uses the

app.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from god_simulator.api.gradio_app import create_gradio_app
+
+app = create_gradio_app(
+    config_path=Path(os.getenv("GOD_SIMULATOR_CONFIG", "config/game.modal.local.json")),
+    static_dir=Path(os.getenv("GOD_SIMULATOR_STATIC_DIR", "dist/frontend")),
+)
+
+
+def _server_port() -> int | None:
+    value = os.getenv("GRADIO_SERVER_PORT")
+    return int(value) if value else None
+
+
+if __name__ == "__main__":
+    app.launch(
+        server_name=os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"),
+        server_port=_server_port(),
+    )

frontend/src/api.ts

@@ -1,6 +1,13 @@
 import type { WorldSnapshot } from "./types";
 
-const API_BASE = import.meta.env.VITE_GOD_SIMULATOR_API ?? "http://127.0.0.1:8000";
+const API_BASE = import.meta.env.VITE_GOD_SIMULATOR_API ?? defaultApiBase();
+
+function defaultApiBase(): string {
+  if (window.location.hostname === "127.0.0.1" && window.location.port === "5173") {
+    return "http://127.0.0.1:8000";
+  }
+  return window.location.origin;
+}
 
 export class ApiResponseError extends Error {
   readonly status: number;

package.json

@@ -6,6 +6,7 @@
   "scripts": {
     "dev": "vite --host 127.0.0.1",
     "build": "tsc -b && vite build",
+    "gradio": "npm run build && uv run python app.py",
     "preview": "vite preview --host 127.0.0.1",
     "typecheck": "tsc -b",
     "test:e2e": "playwright test"

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.11"
 license = { text = "MIT" }
 authors = [{ name = "God Simulator Team" }]
 dependencies = [
+    "gradio>=6.17.3",
     "openai>=1.0",
 ]
 

requirements.txt

@@ -0,0 +1 @@
+-e .

src/god_simulator/api/gradio_app.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from http import HTTPStatus
+from pathlib import Path
+from typing import Any
+
+import gradio as gr
+from fastapi import HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+
+from god_simulator.api.runtime import GameRuntime, create_game_runtime
+from god_simulator.config import load_game_config
+from god_simulator.simulation.spawning import create_world
+
+
+class GodCommandRequest(BaseModel):
+    command: str
+
+
+def create_gradio_app(
+    *,
+    config_path: Path,
+    static_dir: Path = Path("dist/frontend"),
+) -> gr.Server:
+    config = load_game_config(config_path)
+    runtime = create_game_runtime(world=create_world(config), config=config)
+    app = gr.Server(title="God Simulator")
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=False,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    register_api_routes(app, runtime)
+    register_gradio_api(app, runtime)
+    register_static_routes(app, static_dir)
+    return app
+
+
+def register_api_routes(app: gr.Server, runtime: GameRuntime) -> None:
+    @app.get("/health")
+    def health() -> dict[str, Any]:
+        return runtime.health()
+
+    @app.get("/state")
+    def state() -> dict[str, Any]:
+        return runtime.state()
+
+    @app.get("/claw3d/state")
+    def claw3d_state(warmup: str | None = None) -> dict[str, Any]:
+        return runtime.claw3d_state(warmup=_truthy(warmup))
+
+    @app.post("/tick")
+    def tick() -> JSONResponse:
+        status, payload = runtime.tick()
+        return JSONResponse(status_code=int(status), content=payload)
+
+    @app.post("/god-command")
+    def god_command(request: GodCommandRequest) -> JSONResponse:
+        status, payload = runtime.god_command(request.command)
+        return JSONResponse(status_code=int(status), content=payload)
+
+
+def register_gradio_api(app: gr.Server, runtime: GameRuntime) -> None:
+    @app.api(name="tick", concurrency_limit=1)  # type: ignore[untyped-decorator]
+    def tick_api() -> dict[str, Any]:
+        status, payload = runtime.tick()
+        if status != HTTPStatus.OK:
+            raise HTTPException(status_code=int(status), detail=payload)
+        return payload
+
+    @app.api(name="god_command", concurrency_limit=1)  # type: ignore[untyped-decorator]
+    def god_command_api(command: str) -> dict[str, Any]:
+        status, payload = runtime.god_command(command)
+        if status != HTTPStatus.OK:
+            raise HTTPException(status_code=int(status), detail=payload)
+        return payload
+
+
+def register_static_routes(app: gr.Server, static_dir: Path) -> None:
+    static_dir = static_dir.resolve()
+    assets_dir = static_dir / "assets"
+    index_path = static_dir / "index.html"
+
+    if assets_dir.exists():
+        app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+    @app.get("/", response_class=HTMLResponse, response_model=None)
+    def root() -> FileResponse | HTMLResponse:
+        if index_path.exists():
+            return FileResponse(index_path)
+        return missing_frontend_response(static_dir)
+
+    @app.get("/{path:path}", response_class=HTMLResponse, response_model=None)
+    def spa_fallback(path: str) -> FileResponse | HTMLResponse:
+        requested_path = (static_dir / path).resolve()
+        if index_path.exists() and _is_within(static_dir, requested_path):
+            if requested_path.is_file():
+                return FileResponse(requested_path)
+            return FileResponse(index_path)
+        return missing_frontend_response(static_dir)
+
+
+def missing_frontend_response(static_dir: Path) -> HTMLResponse:
+    body = (
+        "<main style='font-family: system-ui; padding: 2rem'>"
+        "<h1>God Simulator frontend is not built</h1>"
+        f"<p>Expected built assets in <code>{static_dir}</code>.</p>"
+        "<p>Run <code>npm install</code> and <code>npm run build</code>, "
+        "then restart the Gradio app.</p>"
+        "</main>"
+    )
+    return HTMLResponse(body, status_code=503)
+
+
+def _truthy(value: str | None) -> bool:
+    return value is not None and value.lower() in ("1", "true", "yes")
+
+
+def _is_within(parent: Path, child: Path) -> bool:
+    try:
+        child.relative_to(parent)
+    except ValueError:
+        return False
+    return True

src/god_simulator/api/runtime.py

@@ -0,0 +1,424 @@
+from __future__ import annotations
+
+import time
+from copy import deepcopy
+from dataclasses import dataclass
+from http import HTTPStatus
+from threading import Lock, Thread
+from typing import Any, Protocol
+from urllib.error import HTTPError, URLError
+from urllib.request import urlopen
+
+from god_simulator.config import ConnectorConfig, GameConfig
+from god_simulator.domain import WorldState
+from god_simulator.rendering.claw3d_contract import to_claw3d_snapshot
+from god_simulator.simulation.connectors.base import WorldSimulator
+from god_simulator.simulation.connectors.factory import create_world_simulator
+from god_simulator.simulation.god_console import GodConsole, apply_god_command_plan
+from god_simulator.simulation.tick import apply_tick_plan, plan_world_tick
+
+
+class GameRuntime:
+    def __init__(
+        self,
+        *,
+        world: WorldState,
+        simulator: WorldSimulator,
+        god_console: GodConsole | None = None,
+        modal_health_warmer: HealthWarmer | None = None,
+    ) -> None:
+        self._world = world
+        self._simulator = simulator
+        self._god_console = god_console
+        self._modal_health_warmer = modal_health_warmer
+        self._lock = Lock()
+        self._tick_status = TickStatus()
+        self._god_status = GodCommandStatus()
+
+    @property
+    def simulator_name(self) -> str:
+        return self._simulator.name
+
+    def health(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "status": "ok",
+                "tick": self._world.tick,
+                "simulator": self._simulator.name,
+                "last_tick_source": self._world.last_tick_source,
+                **tick_status_payload(self._tick_status),
+            }
+
+    def state(self) -> dict[str, Any]:
+        with self._lock:
+            payload = self._world.to_dict()
+            payload["simulation"] = tick_status_payload(self._tick_status)
+            return payload
+
+    def claw3d_state(self, *, warmup: bool = False) -> dict[str, Any]:
+        if warmup and self._modal_health_warmer is not None:
+            self._modal_health_warmer.trigger(force=True)
+
+        with self._lock:
+            payload = to_claw3d_snapshot(
+                self._world,
+                tick_in_progress=self._tick_status.in_progress,
+                pending_tick=self._tick_status.pending_tick,
+            )
+        if self._modal_health_warmer is not None:
+            payload["simulation"]["models"] = self._modal_health_warmer.statuses()
+        return payload
+
+    def tick(self) -> tuple[HTTPStatus, dict[str, Any]]:
+        planning_world, next_tick, conflict = self._begin_tick()
+        if conflict is not None:
+            return HTTPStatus.CONFLICT, conflict
+        assert planning_world is not None
+        assert next_tick is not None
+
+        try:
+            planned_tick, plan = plan_world_tick(planning_world, self._simulator, next_tick)
+        except Exception as exc:
+            with self._lock:
+                self._tick_status.in_progress = False
+                self._tick_status.pending_tick = None
+            return HTTPStatus.INTERNAL_SERVER_ERROR, {
+                "error": "tick_failed",
+                "message": str(exc),
+            }
+
+        with self._lock:
+            apply_tick_plan(self._world, planned_tick, plan)
+            self._tick_status.in_progress = False
+            self._tick_status.pending_tick = None
+            payload = self._world.to_dict()
+            payload["simulation"] = tick_status_payload(self._tick_status)
+            return HTTPStatus.OK, payload
+
+    def god_command(self, command: str) -> tuple[HTTPStatus, dict[str, Any]]:
+        if self._god_console is None:
+            return HTTPStatus.BAD_REQUEST, {"error": "god_console_unavailable"}
+        if not command.strip():
+            return HTTPStatus.BAD_REQUEST, {
+                "error": "invalid_request",
+                "message": "Expected non-empty 'command'.",
+            }
+
+        planning_world, conflict = self._begin_god_command()
+        if conflict is not None:
+            return HTTPStatus.CONFLICT, conflict
+        assert planning_world is not None
+
+        try:
+            plan = self._god_console.propose(planning_world, command.strip())
+        except Exception as exc:
+            with self._lock:
+                self._god_status.in_progress = False
+            return HTTPStatus.INTERNAL_SERVER_ERROR, {
+                "error": "god_command_failed",
+                "message": str(exc),
+            }
+
+        with self._lock:
+            applied = apply_god_command_plan(self._world, plan)
+            self._god_status.in_progress = False
+            return HTTPStatus.OK, {
+                "source": plan.source,
+                "summary": plan.summary,
+                "applied": applied,
+                "snapshot": to_claw3d_snapshot(
+                    self._world,
+                    tick_in_progress=self._tick_status.in_progress,
+                    pending_tick=self._tick_status.pending_tick,
+                ),
+            }
+
+    def _begin_tick(self) -> tuple[WorldState | None, int | None, dict[str, Any] | None]:
+        with self._lock:
+            if self._tick_status.in_progress:
+                return None, None, {
+                    "error": "tick_in_progress",
+                    "tick": self._world.tick,
+                    **tick_status_payload(self._tick_status),
+                }
+            if self._god_status.in_progress:
+                return None, None, {
+                    "error": "god_command_in_progress",
+                    "message": "A god command is already modifying the world.",
+                    "tick": self._world.tick,
+                    **tick_status_payload(self._tick_status),
+                }
+
+            next_tick = self._world.tick + 1
+            planning_world = deepcopy(self._world)
+            self._tick_status.in_progress = True
+            self._tick_status.pending_tick = next_tick
+            return planning_world, next_tick, None
+
+    def _begin_god_command(self) -> tuple[WorldState | None, dict[str, Any] | None]:
+        with self._lock:
+            if self._tick_status.in_progress:
+                return None, {
+                    "error": "tick_in_progress",
+                    "message": "God commands are disabled while a tick is running.",
+                    **tick_status_payload(self._tick_status),
+                }
+            if self._god_status.in_progress:
+                return None, {
+                    "error": "god_command_in_progress",
+                    "message": "Another god command is already running.",
+                    **tick_status_payload(self._tick_status),
+                }
+
+            self._god_status.in_progress = True
+            return deepcopy(self._world), None
+
+
+def create_game_runtime(*, world: WorldState, config: GameConfig) -> GameRuntime:
+    simulator = create_world_simulator(config)
+    god_console_config = config.god_console or config.connector
+    god_console = (
+        GodConsole(god_console_config)
+        if god_console_config.type == "openai_compatible"
+        else None
+    )
+    return GameRuntime(
+        world=world,
+        simulator=simulator,
+        god_console=god_console,
+        modal_health_warmer=ModalHealthWarmer(modal_health_targets(config)),
+    )
+
+
+@dataclass(slots=True)
+class TickStatus:
+    in_progress: bool = False
+    pending_tick: int | None = None
+
+
+@dataclass(slots=True)
+class GodCommandStatus:
+    in_progress: bool = False
+
+
+def tick_status_payload(tick_status: TickStatus) -> dict[str, Any]:
+    return {
+        "tick_in_progress": tick_status.in_progress,
+        "pending_tick": tick_status.pending_tick,
+    }
+
+
+class ModalHealthWarmer:
+    def __init__(
+        self,
+        targets: list[ModalHealthTarget],
+        *,
+        cooldown_seconds: float = 300,
+        warmup_after_seconds: float = 2,
+        warmup_retry_seconds: float = 5,
+        opener: UrlOpener = urlopen,
+    ) -> None:
+        self._states = {
+            target.id: ModalHealthState(target=target)
+            for target in targets
+        }
+        self._cooldown_seconds = cooldown_seconds
+        self._warmup_after_seconds = warmup_after_seconds
+        self._warmup_retry_seconds = warmup_retry_seconds
+        self._opener = opener
+        self._lock = Lock()
+
+    def trigger(self, *, force: bool = False) -> None:
+        if not self._states:
+            return
+
+        now = time.monotonic()
+        launches: list[tuple[str, str, int]] = []
+
+        with self._lock:
+            for target_id, state in self._states.items():
+                if state.in_progress:
+                    continue
+                if not force and now - state.last_started_at < self._cooldown_seconds:
+                    continue
+                state.in_progress = True
+                state.status = "checking"
+                state.http_status = None
+                state.last_started_at = now
+                state.generation += 1
+                launches.append((target_id, state.target.health_url, state.generation))
+
+        for target_id, health_url, generation in launches:
+            Thread(
+                target=self._mark_warmup_if_pending,
+                args=(target_id, generation),
+                name="modal-health-warmup-status",
+                daemon=True,
+            ).start()
+            Thread(
+                target=self._run_one,
+                args=(target_id, health_url, generation),
+                name="modal-health-warmer",
+                daemon=True,
+            ).start()
+
+    def statuses(self) -> list[dict[str, Any]]:
+        with self._lock:
+            return [
+                {
+                    "id": state.target.id,
+                    "label": state.target.label,
+                    "model": state.target.model,
+                    "status": state.status,
+                    "http_status": state.http_status,
+                }
+                for state in self._states.values()
+            ]
+
+    def _mark_warmup_if_pending(self, target_id: str, generation: int) -> None:
+        time.sleep(self._warmup_after_seconds)
+        with self._lock:
+            state = self._states.get(target_id)
+            if (
+                state is not None
+                and state.generation == generation
+                and state.in_progress
+                and state.status == "checking"
+            ):
+                state.status = "warmup"
+
+    def _run_one(self, target_id: str, health_url: str, generation: int) -> None:
+        while True:
+            try:
+                response = self._opener(health_url)
+                try:
+                    http_status = int(getattr(response, "status", 0))
+                finally:
+                    response.close()
+            except HTTPError as exc:
+                http_status = int(exc.code)
+                if http_status != HTTPStatus.SERVICE_UNAVAILABLE:
+                    self._set_status(target_id, generation, "error", http_status)
+                    print(f"Modal health check failed for {health_url}: {exc}", flush=True)
+                    return
+            except (TimeoutError, URLError, OSError) as exc:
+                self._set_status(target_id, generation, "error", None)
+                print(f"Modal health check failed for {health_url}: {exc}", flush=True)
+                return
+
+            if http_status == HTTPStatus.OK:
+                self._set_status(target_id, generation, "ready", http_status)
+                print(f"Modal health check warmed {health_url}: {http_status}", flush=True)
+                return
+
+            if http_status == HTTPStatus.SERVICE_UNAVAILABLE:
+                if not self._set_status(
+                    target_id,
+                    generation,
+                    "warmup",
+                    http_status,
+                    in_progress=True,
+                ):
+                    return
+                print(f"Modal health check warming {health_url}: {http_status}", flush=True)
+                time.sleep(self._warmup_retry_seconds)
+                continue
+
+            self._set_status(target_id, generation, "error", http_status)
+            print(f"Modal health check failed for {health_url}: {http_status}", flush=True)
+            return
+
+    def _set_status(
+        self,
+        target_id: str,
+        generation: int,
+        status: str,
+        http_status: int | None,
+        *,
+        in_progress: bool = False,
+    ) -> bool:
+        with self._lock:
+            state = self._states.get(target_id)
+            if state is None or state.generation != generation:
+                return False
+            state.status = status
+            state.http_status = http_status
+            state.in_progress = in_progress
+            return True
+
+
+@dataclass(frozen=True, slots=True)
+class ModalHealthTarget:
+    id: str
+    label: str
+    model: str | None
+    health_url: str
+
+
+@dataclass(slots=True)
+class ModalHealthState:
+    target: ModalHealthTarget
+    status: str = "unknown"
+    http_status: int | None = None
+    last_started_at: float = 0.0
+    generation: int = 0
+    in_progress: bool = False
+
+
+class UrlResponse(Protocol):
+    status: int
+
+    def close(self) -> None: ...
+
+
+class UrlOpener(Protocol):
+    def __call__(self, url: str) -> UrlResponse: ...
+
+
+class HealthWarmer(Protocol):
+    def trigger(self, *, force: bool = False) -> None: ...
+
+    def statuses(self) -> list[dict[str, Any]]: ...
+
+
+def modal_health_urls(config: GameConfig) -> list[str]:
+    return [target.health_url for target in modal_health_targets(config)]
+
+
+def modal_health_targets(config: GameConfig) -> list[ModalHealthTarget]:
+    targets: list[ModalHealthTarget] = []
+    for target_id, label, connector in (
+        ("npc_model", "NPC model", config.connector),
+        ("god_console", "God Console", config.god_console),
+    ):
+        if connector is None:
+            continue
+        health_url = modal_health_url(connector)
+        if health_url is not None:
+            targets.append(
+                ModalHealthTarget(
+                    id=target_id,
+                    label=label,
+                    model=connector.model,
+                    health_url=health_url,
+                )
+            )
+    return targets
+
+
+def modal_health_url(connector: ConnectorConfig) -> str | None:
+    if connector.type != "openai_compatible":
+        return None
+
+    root = connector.base_url or connector.api_url
+    if not root:
+        return None
+
+    root = root.strip().rstrip("/")
+    for suffix in ("/v1/chat/completions", "/chat/completions", "/v1"):
+        if root.endswith(suffix):
+            root = root.removesuffix(suffix)
+
+    if ".modal.run" not in root:
+        return None
+    return f"{root}/health"

src/god_simulator/api/server.py

@@ -1,43 +1,57 @@
 from __future__ import annotations
 
 import json
-import time
-from copy import deepcopy
-from dataclasses import dataclass
 from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from threading import Lock, Thread
-from typing import Any, Protocol
-from urllib.error import HTTPError, URLError
+from typing import Any
 from urllib.parse import parse_qs, urlparse
-from urllib.request import urlopen
 
-from god_simulator.config import ConnectorConfig, GameConfig
+from god_simulator.api.runtime import (
+    GameRuntime,
+    create_game_runtime,
+)
+from god_simulator.api.runtime import (
+    HealthWarmer as _HealthWarmer,
+)
+from god_simulator.api.runtime import (
+    ModalHealthTarget as _ModalHealthTarget,
+)
+from god_simulator.api.runtime import (
+    ModalHealthWarmer as _ModalHealthWarmer,
+)
+from god_simulator.api.runtime import (
+    modal_health_targets as _modal_health_targets,
+)
+from god_simulator.api.runtime import (
+    modal_health_url as _modal_health_url,
+)
+from god_simulator.api.runtime import (
+    modal_health_urls as _modal_health_urls,
+)
+from god_simulator.config import GameConfig
 from god_simulator.domain import WorldState
-from god_simulator.rendering.claw3d_contract import to_claw3d_snapshot
 from god_simulator.simulation.connectors.base import WorldSimulator
-from god_simulator.simulation.connectors.factory import create_world_simulator
-from god_simulator.simulation.god_console import (
-    GodConsole,
-    apply_god_command_plan,
-)
-from god_simulator.simulation.tick import apply_tick_plan, plan_world_tick
+from god_simulator.simulation.god_console import GodConsole
+
+__all__ = [
+    "_ModalHealthTarget",
+    "_ModalHealthWarmer",
+    "_build_handler",
+    "_modal_health_targets",
+    "_modal_health_url",
+    "_modal_health_urls",
+    "build_handler",
+    "run_server",
+]
 
 
 def run_server(*, world: WorldState, config: GameConfig) -> None:
-    simulator = create_world_simulator(config)
-    god_console_config = config.god_console or config.connector
-    god_console = (
-        GodConsole(god_console_config)
-        if god_console_config.type == "openai_compatible"
-        else None
-    )
-    modal_health_warmer = _ModalHealthWarmer(_modal_health_targets(config))
-    handler = _build_handler(world, simulator, god_console, modal_health_warmer)
+    runtime = create_game_runtime(world=world, config=config)
+    handler = build_handler(runtime)
     address = (config.server.host, config.server.port)
     server = ThreadingHTTPServer(address, handler)
     print(f"God Simulator API listening on http://{config.server.host}:{config.server.port}")
-    print(f"World simulator connector: {simulator.name}")
+    print(f"World simulator connector: {runtime.simulator_name}")
     print(
         "Available endpoints: GET /health, GET /state, GET /claw3d/state, "
         "POST /tick, POST /god-command"
@@ -51,10 +65,17 @@ def _build_handler(
     god_console: GodConsole | None = None,
     modal_health_warmer: _HealthWarmer | None = None,
 ) -> type[BaseHTTPRequestHandler]:
-    lock = Lock()
-    tick_status = _TickStatus()
-    god_status = _GodCommandStatus()
+    return build_handler(
+        GameRuntime(
+            world=world,
+            simulator=simulator,
+            god_console=god_console,
+            modal_health_warmer=modal_health_warmer,
+        )
+    )
+
 
+def build_handler(runtime: GameRuntime) -> type[BaseHTTPRequestHandler]:
     class GodSimulatorRequestHandler(BaseHTTPRequestHandler):
         server_version = "GodSimulator/0.1"
 
@@ -68,38 +89,16 @@ def _build_handler(
             path = parsed_path.path
 
             if path == "/health":
-                with lock:
-                    payload = {
-                        "status": "ok",
-                        "tick": world.tick,
-                        "simulator": simulator.name,
-                        "last_tick_source": world.last_tick_source,
-                        **_tick_status_payload(tick_status),
-                    }
-                self._send_json(payload)
+                self._send_json(runtime.health())
                 return
-
             if path == "/state":
-                with lock:
-                    payload = world.to_dict()
-                    payload["simulation"] = _tick_status_payload(tick_status)
-                self._send_json(payload)
+                self._send_json(runtime.state())
                 return
-
             if path == "/claw3d/state":
-                if modal_health_warmer is not None:
-                    query = parse_qs(parsed_path.query)
-                    if _truthy_query_value(query, "warmup"):
-                        modal_health_warmer.trigger(force=True)
-                with lock:
-                    payload = to_claw3d_snapshot(
-                        world,
-                        tick_in_progress=tick_status.in_progress,
-                        pending_tick=tick_status.pending_tick,
-                    )
-                if modal_health_warmer is not None:
-                    payload["simulation"]["models"] = modal_health_warmer.statuses()
-                self._send_json(payload)
+                query = parse_qs(parsed_path.query)
+                self._send_json(
+                    runtime.claw3d_state(warmup=_truthy_query_value(query, "warmup"))
+                )
                 return
 
             self._send_json(
@@ -109,65 +108,22 @@ def _build_handler(
 
         def do_POST(self) -> None:
             if self.path == "/tick":
-                payload: dict[str, Any] | None
-                planning_world: WorldState | None = None
-                next_tick: int | None = None
-
-                with lock:
-                    if tick_status.in_progress:
-                        payload = {
-                            "error": "tick_in_progress",
-                            "tick": world.tick,
-                            **_tick_status_payload(tick_status),
-                        }
-                    elif god_status.in_progress:
-                        payload = {
-                            "error": "god_command_in_progress",
-                            "message": "A god command is already modifying the world.",
-                            "tick": world.tick,
-                            **_tick_status_payload(tick_status),
-                        }
-                    else:
-                        payload = None
-
-                        next_tick = world.tick + 1
-                        planning_world = deepcopy(world)
-                        tick_status.in_progress = True
-                        tick_status.pending_tick = next_tick
-
-                if payload is not None:
-                    self._send_json(payload, status=HTTPStatus.CONFLICT)
-                    return
+                status, payload = runtime.tick()
+                self._send_json(payload, status=status)
+                return
 
-                assert planning_world is not None
-                assert next_tick is not None
+            if self.path == "/god-command":
                 try:
-                    planned_tick, plan = plan_world_tick(planning_world, simulator, next_tick)
-                except Exception as exc:
-                    with lock:
-                        tick_status.in_progress = False
-                        tick_status.pending_tick = None
+                    command = self._read_command()
+                except ValueError as exc:
                     self._send_json(
-                        {
-                            "error": "tick_failed",
-                            "message": str(exc),
-                        },
-                        status=HTTPStatus.INTERNAL_SERVER_ERROR,
+                        {"error": "invalid_request", "message": str(exc)},
+                        status=HTTPStatus.BAD_REQUEST,
                     )
                     return
 
-                with lock:
-                    apply_tick_plan(world, planned_tick, plan)
-                    tick_status.in_progress = False
-                    tick_status.pending_tick = None
-                    payload = world.to_dict()
-                    payload["simulation"] = _tick_status_payload(tick_status)
-
-                self._send_json(payload)
-                return
-
-            if self.path == "/god-command":
-                self._handle_god_command()
+                status, payload = runtime.god_command(command)
+                self._send_json(payload, status=status)
                 return
 
             self._send_json(
@@ -195,78 +151,6 @@ def _build_handler(
             except (BrokenPipeError, ConnectionAbortedError, ConnectionResetError):
                 return
 
-        def _handle_god_command(self) -> None:
-            if god_console is None:
-                self._send_json(
-                    {"error": "god_console_unavailable"},
-                    status=HTTPStatus.BAD_REQUEST,
-                )
-                return
-
-            try:
-                command = self._read_command()
-            except ValueError as exc:
-                self._send_json(
-                    {"error": "invalid_request", "message": str(exc)},
-                    status=HTTPStatus.BAD_REQUEST,
-                )
-                return
-
-            with lock:
-                if tick_status.in_progress:
-                    payload = {
-                        "error": "tick_in_progress",
-                        "message": "God commands are disabled while a tick is running.",
-                        **_tick_status_payload(tick_status),
-                    }
-                    planning_world = None
-                elif god_status.in_progress:
-                    payload = {
-                        "error": "god_command_in_progress",
-                        "message": "Another god command is already running.",
-                        **_tick_status_payload(tick_status),
-                    }
-                    planning_world = None
-                else:
-                    payload = None
-                    planning_world = deepcopy(world)
-                    god_status.in_progress = True
-
-            if payload is not None:
-                self._send_json(payload, status=HTTPStatus.CONFLICT)
-                return
-
-            assert planning_world is not None
-            try:
-                plan = god_console.propose(planning_world, command)
-            except Exception as exc:
-                with lock:
-                    god_status.in_progress = False
-                self._send_json(
-                    {
-                        "error": "god_command_failed",
-                        "message": str(exc),
-                    },
-                    status=HTTPStatus.INTERNAL_SERVER_ERROR,
-                )
-                return
-
-            with lock:
-                applied = apply_god_command_plan(world, plan)
-                god_status.in_progress = False
-                response = {
-                    "source": plan.source,
-                    "summary": plan.summary,
-                    "applied": applied,
-                    "snapshot": to_claw3d_snapshot(
-                        world,
-                        tick_in_progress=tick_status.in_progress,
-                        pending_tick=tick_status.pending_tick,
-                    ),
-                }
-
-            self._send_json(response)
-
         def _send_cors_headers(self) -> None:
             self.send_header("Access-Control-Allow-Origin", "*")
             self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
@@ -298,253 +182,5 @@ def _build_handler(
     return GodSimulatorRequestHandler
 
 
-@dataclass(slots=True)
-class _TickStatus:
-    in_progress: bool = False
-    pending_tick: int | None = None
-
-
-@dataclass(slots=True)
-class _GodCommandStatus:
-    in_progress: bool = False
-
-
-def _tick_status_payload(tick_status: _TickStatus) -> dict[str, Any]:
-    return {
-        "tick_in_progress": tick_status.in_progress,
-        "pending_tick": tick_status.pending_tick,
-    }
-
-
-class _ModalHealthWarmer:
-    def __init__(
-        self,
-        targets: list[_ModalHealthTarget],
-        *,
-        cooldown_seconds: float = 300,
-        warmup_after_seconds: float = 2,
-        warmup_retry_seconds: float = 5,
-        opener: _UrlOpener = urlopen,
-    ) -> None:
-        self._states = {
-            target.id: _ModalHealthState(target=target)
-            for target in targets
-        }
-        self._cooldown_seconds = cooldown_seconds
-        self._warmup_after_seconds = warmup_after_seconds
-        self._warmup_retry_seconds = warmup_retry_seconds
-        self._opener = opener
-        self._lock = Lock()
-
-    def trigger(self, *, force: bool = False) -> None:
-        if not self._states:
-            return
-
-        now = time.monotonic()
-        launches: list[tuple[str, str, int]] = []
-
-        with self._lock:
-            for target_id, state in self._states.items():
-                if state.in_progress:
-                    continue
-                if (
-                    not force
-                    and (
-                        now - state.last_started_at < self._cooldown_seconds
-                    )
-                ):
-                    continue
-                state.in_progress = True
-                state.status = "checking"
-                state.http_status = None
-                state.last_started_at = now
-                state.generation += 1
-                launches.append((target_id, state.target.health_url, state.generation))
-
-        for target_id, health_url, generation in launches:
-            Thread(
-                target=self._mark_warmup_if_pending,
-                args=(target_id, generation),
-                name="modal-health-warmup-status",
-                daemon=True,
-            ).start()
-            Thread(
-                target=self._run_one,
-                args=(target_id, health_url, generation),
-                name="modal-health-warmer",
-                daemon=True,
-            ).start()
-
-    def statuses(self) -> list[dict[str, Any]]:
-        with self._lock:
-            return [
-                {
-                    "id": state.target.id,
-                    "label": state.target.label,
-                    "model": state.target.model,
-                    "status": state.status,
-                    "http_status": state.http_status,
-                }
-                for state in self._states.values()
-            ]
-
-    def _mark_warmup_if_pending(self, target_id: str, generation: int) -> None:
-        time.sleep(self._warmup_after_seconds)
-        with self._lock:
-            state = self._states.get(target_id)
-            if (
-                state is not None
-                and state.generation == generation
-                and state.in_progress
-                and state.status == "checking"
-            ):
-                state.status = "warmup"
-
-    def _run_one(self, target_id: str, health_url: str, generation: int) -> None:
-        while True:
-            try:
-                response = self._opener(health_url)
-                try:
-                    http_status = int(getattr(response, "status", 0))
-                finally:
-                    response.close()
-            except HTTPError as exc:
-                http_status = int(exc.code)
-                if http_status != HTTPStatus.SERVICE_UNAVAILABLE:
-                    self._set_status(target_id, generation, "error", http_status)
-                    print(
-                        f"Modal health check failed for {health_url}: {exc}",
-                        flush=True,
-                    )
-                    return
-            except (TimeoutError, URLError, OSError) as exc:
-                self._set_status(target_id, generation, "error", None)
-                print(f"Modal health check failed for {health_url}: {exc}", flush=True)
-                return
-
-            if http_status == HTTPStatus.OK:
-                self._set_status(target_id, generation, "ready", http_status)
-                print(
-                    f"Modal health check warmed {health_url}: {http_status}",
-                    flush=True,
-                )
-                return
-
-            if http_status == HTTPStatus.SERVICE_UNAVAILABLE:
-                if not self._set_status(
-                    target_id,
-                    generation,
-                    "warmup",
-                    http_status,
-                    in_progress=True,
-                ):
-                    return
-                print(
-                    f"Modal health check warming {health_url}: {http_status}",
-                    flush=True,
-                )
-                time.sleep(self._warmup_retry_seconds)
-                continue
-
-            self._set_status(target_id, generation, "error", http_status)
-            print(f"Modal health check failed for {health_url}: {http_status}", flush=True)
-            return
-
-    def _set_status(
-        self,
-        target_id: str,
-        generation: int,
-        status: str,
-        http_status: int | None,
-        *,
-        in_progress: bool = False,
-    ) -> bool:
-        with self._lock:
-            state = self._states.get(target_id)
-            if state is None or state.generation != generation:
-                return False
-            state.status = status
-            state.http_status = http_status
-            state.in_progress = in_progress
-            return True
-
-
-@dataclass(frozen=True, slots=True)
-class _ModalHealthTarget:
-    id: str
-    label: str
-    model: str | None
-    health_url: str
-
-
-@dataclass(slots=True)
-class _ModalHealthState:
-    target: _ModalHealthTarget
-    status: str = "unknown"
-    http_status: int | None = None
-    last_started_at: float = 0.0
-    generation: int = 0
-    in_progress: bool = False
-
-
-class _UrlResponse(Protocol):
-    status: int
-
-    def close(self) -> None: ...
-
-
-class _UrlOpener(Protocol):
-    def __call__(self, url: str) -> _UrlResponse: ...
-
-
-class _HealthWarmer(Protocol):
-    def trigger(self, *, force: bool = False) -> None: ...
-
-    def statuses(self) -> list[dict[str, Any]]: ...
-
-
-def _modal_health_urls(config: GameConfig) -> list[str]:
-    return [target.health_url for target in _modal_health_targets(config)]
-
-
-def _modal_health_targets(config: GameConfig) -> list[_ModalHealthTarget]:
-    targets: list[_ModalHealthTarget] = []
-    for target_id, label, connector in (
-        ("npc_model", "NPC model", config.connector),
-        ("god_console", "God Console", config.god_console),
-    ):
-        if connector is None:
-            continue
-        health_url = _modal_health_url(connector)
-        if health_url is not None:
-            targets.append(
-                _ModalHealthTarget(
-                    id=target_id,
-                    label=label,
-                    model=connector.model,
-                    health_url=health_url,
-                )
-            )
-    return targets
-
-
-def _modal_health_url(connector: ConnectorConfig) -> str | None:
-    if connector.type != "openai_compatible":
-        return None
-
-    root = connector.base_url or connector.api_url
-    if not root:
-        return None
-
-    root = root.strip().rstrip("/")
-    for suffix in ("/v1/chat/completions", "/chat/completions", "/v1"):
-        if root.endswith(suffix):
-            root = root.removesuffix(suffix)
-
-    if ".modal.run" not in root:
-        return None
-    return f"{root}/health"
-
-
 def _truthy_query_value(query: dict[str, list[str]], key: str) -> bool:
     return any(value.lower() in ("1", "true", "yes") for value in query.get(key, []))

tests/test_gradio_app.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from god_simulator.api.gradio_app import create_gradio_app
+
+
+def test_gradio_app_exposes_game_api(tmp_path: Path) -> None:
+    config_path = _write_config(tmp_path)
+    app = create_gradio_app(config_path=config_path, static_dir=tmp_path / "dist")
+    client = TestClient(app)
+
+    health = client.get("/health")
+    assert health.status_code == 200
+    assert health.json()["status"] == "ok"
+
+    snapshot = client.get("/claw3d/state?warmup=1")
+    assert snapshot.status_code == 200
+    assert snapshot.json()["terrain"]["kind"] == "plain_green"
+
+    tick = client.post("/tick")
+    assert tick.status_code == 200
+    assert tick.json()["tick"] == 1
+
+
+def test_gradio_app_serves_missing_frontend_page(tmp_path: Path) -> None:
+    config_path = _write_config(tmp_path)
+    app = create_gradio_app(config_path=config_path, static_dir=tmp_path / "missing-dist")
+    client = TestClient(app)
+
+    response = client.get("/")
+
+    assert response.status_code == 503
+    assert "frontend is not built" in response.text
+
+
+def _write_config(tmp_path: Path) -> Path:
+    config_path = tmp_path / "game.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "world": {
+                    "width": 80,
+                    "depth": 80,
+                    "terrain": "plain_green",
+                    "seed": 42,
+                },
+                "npcs": {"count": 2},
+                "simulation": {"tick_ms": 500},
+                "server": {"host": "127.0.0.1", "port": 8000},
+            }
+        ),
+        encoding="utf-8",
+    )
+    return config_path