#!/usr/bin/env python3
"""
FastAPI web frontend for the text-based LLM interaction system.
Serves a simple web UI and exposes an API to interact with the LLM.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Optional, Tuple
from uuid import uuid4
from threading import Lock
from fastapi import FastAPI, Request, Response, Cookie, HTTPException
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from config import Config
from llm_client import LLMClient
from conversation import ConversationManager
from game_config import load_game_config
from game_state import GameState
# Application setup
app = FastAPI(title="Text Adventure - Web UI")
BASE_DIR = Path(__file__).resolve().parent
STATIC_DIR = BASE_DIR / "static"
STATIC_DIR.mkdir(parents=True, exist_ok=True)
SESSIONS_DIR = (BASE_DIR.parent / "data" / "sessions")
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
# Mount /static for assets (requires 'aiofiles' in env for async file serving)
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
# Models
class ChatRequest(BaseModel):
message: str
# Globals
CONFIG = Config()
CLIENT = LLMClient(CONFIG)
GAMECFG = load_game_config()
_SESSIONS: dict[str, ConversationManager] = {}
_GAME_STATES: dict[str, GameState] = {}
_SESSIONS_LOCK = Lock()
SESSION_COOKIE_NAME = "session_id"
def _get_or_create_session(session_id: Optional[str]) -> Tuple[str, ConversationManager, bool]:
"""
Return an existing conversation manager by session_id,
or create a new session if not present.
Returns (session_id, manager, created_flag).
"""
created = False
with _SESSIONS_LOCK:
if not session_id or session_id not in _SESSIONS:
session_id = uuid4().hex
cm = ConversationManager()
# Seed with system prompt if configured
if getattr(GAMECFG, "system_prompt", None):
cm.add_system_message(GAMECFG.system_prompt)
_SESSIONS[session_id] = cm
created = True
manager = _SESSIONS[session_id]
return session_id, manager, created
def _ensure_state(session_id: str) -> GameState:
"""
Ensure a GameState exists for a given session and return it.
Loads from disk if present, otherwise creates a new state and persists it.
"""
with _SESSIONS_LOCK:
if session_id not in _GAME_STATES:
path = SESSIONS_DIR / f"{session_id}.json"
gs = None
if path.exists():
try:
with path.open("r", encoding="utf-8") as f:
data = json.load(f) or {}
gs = GameState()
for field in [
"room_description",
"door_description",
"door_locked",
"door_open",
"door_id",
"lock_id",
"lock_key_id",
"key_description",
"key_hidden",
"key_revealed",
"key_taken",
"key_id",
"container_id",
"inventory",
"exits",
"completed",
]:
if field in data:
setattr(gs, field, data[field])
except Exception:
gs = None
if gs is None:
gs = GameState.from_files("state")
try:
with path.open("w", encoding="utf-8") as f:
json.dump(gs.__dict__, f)
except Exception:
pass
_GAME_STATES[session_id] = gs
return _GAME_STATES[session_id]
# Routes
@app.get("/", response_class=FileResponse)
def index() -> FileResponse:
"""
Serve the main web UI.
"""
index_path = STATIC_DIR / "index.html"
if not index_path.exists():
# Provide a minimal fallback page if index.html is missing
# This should not happen in normal usage.
fallback = """
Text Adventure
Text Adventure Web UI
index.html is missing.
"""
tmp_path = STATIC_DIR / "_fallback_index.html"
tmp_path.write_text(fallback, encoding="utf-8")
return FileResponse(str(tmp_path))
return FileResponse(str(index_path))
@app.get("/api/health")
def health() -> dict:
"""
Health check endpoint.
"""
return {"status": "ok"}
@app.get("/api/session")
def session_info(
response: Response,
session_id: Optional[str] = Cookie(default=None, alias=SESSION_COOKIE_NAME),
) -> JSONResponse:
"""
Ensure a session exists, set cookie if needed, and return scenario metadata and start message.
Also returns the current public game state snapshot.
"""
sid, conv, created = _get_or_create_session(session_id)
if created:
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=sid,
httponly=True,
samesite="lax",
max_age=7 * 24 * 3600,
path="/",
)
gs = _ensure_state(sid)
payload = {
"session_id": sid,
"created": created,
"start_message": getattr(GAMECFG, "start_message", "") or "",
"scenario": getattr(GAMECFG, "scenario", {}) or {},
"rules": getattr(GAMECFG, "rules", []) or [],
"state": gs.to_public_dict(),
}
return JSONResponse(payload)
@app.post("/api/chat")
def chat(
req: ChatRequest,
response: Response,
session_id: Optional[str] = Cookie(default=None, alias=SESSION_COOKIE_NAME),
) -> JSONResponse:
"""
Accept a user message, apply deterministic game logic to update state,
then ask the LLM to narrate the outcome. Maintains a server-side session.
"""
message = req.message.strip()
if not message:
raise HTTPException(status_code=400, detail="Message cannot be empty")
sid, conv, created = _get_or_create_session(session_id)
# Set session cookie if new
if created:
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=sid,
httponly=True,
samesite="lax",
max_age=7 * 24 * 3600, # 7 days
path="/",
)
# Determine outcome via the game engine, then request narration
try:
gs = _ensure_state(sid)
conv.add_user_message(message)
engine_outcome = gs.apply_action(message) # {"events": [...]}
# Build a transient system message with canonical facts for the narrator
narrator_directive = {
"ENGINE_OUTCOME": {
"events": engine_outcome.get("events", []),
"state": gs.to_public_dict(),
},
"NARRATION_RULES": [
"Narrate strictly according to ENGINE_OUTCOME. Do not invent state.",
"Do not add items, unlock objects, or change inventory; the engine already did that.",
"Use 2-5 sentences, present tense, second person. Be concise and vivid.",
"If the action was impossible, explain why using the facts provided.",
],
}
transient_system = {
"role": "system",
"content": "ENGINE CONTEXT (JSON): " + json.dumps(narrator_directive),
}
messages = list(conv.get_history()) + [transient_system]
reply = CLIENT.get_response(messages)
conv.add_assistant_message(reply)
# Persist updated state
try:
with (SESSIONS_DIR / f"{sid}.json").open("w", encoding="utf-8") as f:
json.dump(gs.__dict__, f)
except Exception:
pass
return JSONResponse({
"reply": reply,
"completed": gs.completed,
"events": engine_outcome.get("events", []),
"state": gs.to_public_dict(),
})
except Exception as e:
# Do not leak internal details in production; log as needed.
raise HTTPException(status_code=502, detail=f"LLM backend error: {e}") from e