Add Phase 1 MVP: D&D text-adventure server

FastAPI + Postgres backend with fastapi-users auth, games/characters
CRUD, and a WebSocket chat endpoint where an x.ai Grok DM narrates
play, rolls dice, and maintains character sheets via tool calls.
React + Tailwind frontend covers registration/login, game list and
creation, participation-code join flow, character viewer, and a
real-time chat session view. Docker Compose skeleton (Postgres +
backend) included; Caddy/frontend container wiring is deferred to
the deploy milestone.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/llm/client.py

@@ -0,0 +1,18 @@
+from functools import lru_cache
+from pathlib import Path
+
+from openai import AsyncOpenAI
+
+from app.config import settings
+
+PROMPTS_DIR = Path(__file__).parent / "prompts"
+
+
+@lru_cache
+def get_dm_system_prompt() -> str:
+    return (PROMPTS_DIR / "dm_system_prompt.txt").read_text(encoding="utf-8")
+
+
+@lru_cache
+def get_llm_client() -> AsyncOpenAI:
+    return AsyncOpenAI(api_key=settings.xai_api_key, base_url=settings.xai_base_url)

backend/app/llm/context.py

@@ -0,0 +1,59 @@
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.character import Character
+from app.models.message import Message
+from app.models.user import User
+
+# Rough char-based budget for Phase 1 (~4 chars/token, leaves headroom for system prompt + output).
+MAX_CONTEXT_CHARS = 40_000
+
+
+async def build_context(session: AsyncSession, game_id: uuid.UUID) -> list[dict]:
+    messages = (
+        await session.execute(
+            select(Message).where(Message.game_id == game_id).order_by(Message.id.asc())
+        )
+    ).scalars().all()
+    if not messages:
+        return []
+
+    user_ids = {m.user_id for m in messages if m.user_id is not None}
+    character_ids = {m.character_id for m in messages if m.character_id is not None}
+
+    names: dict[uuid.UUID, str] = {}
+    if user_ids:
+        rows = (await session.execute(select(User.id, User.name).where(User.id.in_(user_ids)))).all()
+        names = {row.id: row.name for row in rows}
+
+    char_names: dict[uuid.UUID, str] = {}
+    if character_ids:
+        rows = (
+            await session.execute(select(Character.id, Character.name).where(Character.id.in_(character_ids)))
+        ).all()
+        char_names = {row.id: row.name for row in rows}
+
+    entries: list[dict] = []
+    for m in messages:
+        if m.sender_type == "dm":
+            entries.append({"role": "assistant", "content": m.content})
+        elif m.sender_type == "player":
+            player_name = names.get(m.user_id, "Spieler") if m.user_id else "Spieler"
+            character_name = char_names.get(m.character_id) if m.character_id else None
+            label = f"{player_name} ({character_name})" if character_name else player_name
+            entries.append({"role": "user", "content": f"{label}: {m.content}"})
+        # sender_type == "system" messages are not sent to the model in Phase 1
+
+    # Keep the newest entries within the char budget, dropping oldest first.
+    total = 0
+    kept: list[dict] = []
+    for entry in reversed(entries):
+        total += len(entry["content"])
+        if total > MAX_CONTEXT_CHARS and kept:
+            break
+        kept.append(entry)
+    kept.reverse()
+
+    return kept

backend/app/llm/orchestrator.py

@@ -0,0 +1,81 @@
+import json
+import logging
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger("app.llm.orchestrator")
+
+from app.config import settings
+from app.llm.client import get_dm_system_prompt, get_llm_client
+from app.llm.context import build_context
+from app.llm.tools import character_sheet, dice
+from app.models.message import Message
+
+MAX_TOOL_ROUNDS = 5
+MAX_TOKENS = 4096
+
+TOOLS = [
+    {"type": "function", "function": dice.TOOL_SCHEMA},
+    {"type": "function", "function": character_sheet.TOOL_SCHEMA},
+]
+
+
+async def _execute_tool_call(
+    session: AsyncSession, game_id: uuid.UUID, tool_name: str, tool_input: dict
+) -> dict:
+    """Runs one tool call. Errors are returned as a payload (not raised), so the DM sees them
+    as a tool result and can recover instead of the whole turn crashing."""
+    try:
+        if tool_name == "roll_dice":
+            return dice.roll(tool_input["notation"])
+        if tool_name == "upsert_character_sheet":
+            return await character_sheet.upsert(session, game_id, tool_input)
+        return {"error": f"Unknown tool {tool_name!r}"}
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("Tool call %s failed: %s", tool_name, exc)
+        await session.rollback()
+        return {"error": str(exc)}
+
+
+async def run_dm_turn(session: AsyncSession, game_id: uuid.UUID) -> Message:
+    client = get_llm_client()
+    messages = [{"role": "system", "content": get_dm_system_prompt()}]
+    messages.extend(await build_context(session, game_id))
+
+    final_text = ""
+    for _ in range(MAX_TOOL_ROUNDS):
+        response = await client.chat.completions.create(
+            model=settings.dm_model,
+            max_tokens=MAX_TOKENS,
+            messages=messages,
+            tools=TOOLS,
+        )
+        choice = response.choices[0]
+        message = choice.message
+        final_text = message.content or ""
+
+        if not message.tool_calls:
+            break
+
+        messages.append(message.model_dump(exclude_unset=True))
+
+        for tool_call in message.tool_calls:
+            tool_input = json.loads(tool_call.function.arguments)
+            result = await _execute_tool_call(session, game_id, tool_call.function.name, tool_input)
+            messages.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": tool_call.id,
+                    "content": json.dumps(result),
+                }
+            )
+    else:
+        if not final_text:
+            final_text = "(Der Dungeon Master braucht einen Moment länger als erwartet — bitte versuche es erneut.)"
+
+    dm_message = Message(game_id=game_id, sender_type="dm", content=final_text)
+    session.add(dm_message)
+    await session.commit()
+    await session.refresh(dm_message)
+    return dm_message

backend/app/llm/prompts/dm_system_prompt.txt

@@ -0,0 +1,32 @@
+Rolle
+Du bist ein erfahrener D&D Dungeon Master und Regelexperte (Standard: D&D 5e, sofern nichts anderes vereinbart wird). Du erschaffst lebendige Abenteuer, leitest die Spieler souverän durch die Sitzung und sorgst für ein rundes Spielerlebnis – von der ersten Frage bis zum Abspann der Session.
+Phase 1: Session 0 (Vorbereitung)
+Bevor das Abenteuer beginnt, lernst du deine Spieler aktiv kennen. Warte nicht auf Informationen – stelle gezielt Fragen, eine nach der anderen, nicht alle auf einmal:
+
+1. Anzahl der Spieler und ihre Vornamen.
+2. Erfahrungsstand: Komplette Anfänger, Gelegenheitsspieler oder Regelkenner? Frage das pro Spieler ab, falls die Gruppe gemischt ist.
+3. Charaktere: Wollen sie eigene, fertige Charaktere mitbringen (Name, Klasse, Rasse, Stats) oder sollst du sie durch die Charaktererstellung führen?
+4. Regelkenntnis: Brauchen sie Erklärungen zu Grundmechaniken (Würfelwürfe, Angriffe, Rettungswürfe) oder reicht ein knapper Hinweis?
+5. Setting/Welt: Vorgefertigte Welt (z. B. Forgotten Realms), eine von dir erfundene Welt, oder wünschen sie sich ein bestimmtes Thema/Genre (High Fantasy, düster, humorvoll, Horror, Piraten, ...)?
+6. Umfang: Einzelnes One-Shot-Abenteuer (eine Sitzung) oder eine längere Kampagne über mehrere Sitzungen? Grobe Zeit, die sie investieren wollen?
+7. Inhaltliche Grenzen: Gibt es Themen, die sie im Spiel nicht wollen (Triggerwarnungen, Gewaltlevel, etc.)?
+
+Passe deinen Hilfe-Level dynamisch an: Anfängern erklärst du Mechaniken proaktiv und schlägst Optionen vor ("Du kannst jetzt A, B oder C tun"). Erfahrenen Spielern gibst du nur auf Nachfrage Regeldetails und lässt ihnen mehr erzählerische Freiheit.
+Phase 2: Das Spiel leiten
+
+* Sobald Charaktere feststehen, sprichst du die Spieler in Spielszenen mit ihrem Charakternamen an. Außerhalb der Spielszenen (Regelfragen, Meta-Absprachen) sprichst du sie mit ihrem echten Vornamen an – und zwar immer per "Du", niemals per "Sie".
+* Trenne klar zwischen IC (In Character – Spielwelt, Dialoge, Beschreibungen) und OOC (Out of Character – Regelklärungen, Meta-Fragen). Kennzeichne OOC-Kommentare deutlich, z. B. mit "(OOC: ...)".
+* Führe die Spieler aktiv durch die Handlung: beschreibe Szenen, NPCs und Umgebungen atmosphärisch, biete konkrete Handlungsoptionen an, wenn Spieler unsicher wirken, und treibe die Geschichte voran statt nur zu reagieren.
+* Für Würfelproben: Sage klar, welcher Wurf nötig ist (z. B. "Mach eine Geschicklichkeitsrettung, Schwierigkeit 13") und was Erfolg/Misserfolg bedeutet. Wenn Spieler keine echten Würfel haben, kannst du Würfe simulieren – frage vorher, was sie bevorzugen.
+* Behalte den Weltzustand im Blick: NPC-Namen, Orte, offene Handlungsstränge, Inventar und Fortschritt der Charaktere bleiben über die Sitzung(en) hinweg konsistent.
+* Baue die Sitzung dramaturgisch auf: Einstieg, ansteigende Spannung, Höhepunkt, Auflösung – bei Kampagnen jede Sitzung mit einem Cliffhanger oder klaren Abschluss beenden.
+* Im Spiel sprichst du im Ton und Slang passend zur Welt und den Charakteren (z. B. archaisch-episch in High Fantasy, derb-direkt bei Piraten). Außerhalb bleibst du locker, freundlich und klar in der Ansprache.
+
+Regeln & Wissensquellen
+
+* Du bist Regelexperte: Bei Unsicherheiten der Spieler erklärst du Mechaniken (Kämpfe, Zauber, Fertigkeiten, Zustände) verständlich und korrekt.
+* Wenn Handbücher/Regelwerke hochgeladen wurden, richtest du dich strikt nach deren Inhalt und zitierst/nennst bei Bedarf die relevante Regel. Wurde nichts hochgeladen, greifst du auf dein allgemeines D&D-5e-Regelwissen zurück und weist transparent darauf hin, falls du bei einer Detailregel unsicher bist, statt sie zu erfinden.
+* Bei Widersprüchen zwischen Spielspaß und exakter Regelauslegung fragst du kurz nach, welche Priorität die Gruppe bevorzugt (z. B. eher erzählerisch-locker oder regelgenau).
+
+Sprache
+Antworte immer in der Sprache, in der die Spieler mit dir kommunizieren. Wechselt die Gruppe die Sprache, wechselst du mit.

backend/app/llm/tools/character_sheet.py

@@ -0,0 +1,89 @@
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.character import Character
+from app.models.game import GameParticipant
+from app.models.user import User
+
+TOOL_SCHEMA = {
+    "name": "upsert_character_sheet",
+    "description": (
+        "Create or update a player's character sheet as facts about them are confirmed or change during play "
+        "(name, race, class, ability stats, abilities, backstory). Call this proactively as soon as details are "
+        "settled — do not wait until the end of the session."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "player_name": {
+                "type": "string",
+                "description": "The real first name of the player this character belongs to, exactly as they introduced themselves.",
+            },
+            "character_id": {
+                "type": "string",
+                "description": "Omit when creating a new character. Provide the existing character's id when updating one.",
+            },
+            "name": {"type": "string"},
+            "race": {"type": "string"},
+            "char_class": {"type": "string"},
+            "stats": {
+                "type": "object",
+                "description": "Ability scores, e.g. {\"STR\": 15, \"DEX\": 12, \"CON\": 14, \"INT\": 10, \"WIS\": 8, \"CHA\": 13}",
+            },
+            "description": {
+                "type": "string",
+                "description": "Free-text description: special abilities, oaths, backstory, anything relevant.",
+            },
+        },
+        "required": ["player_name"],
+    },
+}
+
+
+async def upsert(session: AsyncSession, game_id: uuid.UUID, tool_input: dict) -> dict:
+    player_name = tool_input["player_name"]
+
+    participant_row = (
+        await session.execute(
+            select(GameParticipant, User)
+            .join(User, User.id == GameParticipant.user_id)
+            .where(GameParticipant.game_id == game_id, User.name == player_name)
+        )
+    ).first()
+    if participant_row is None:
+        raise ValueError(f"No participant named {player_name!r} found in this game")
+
+    participant, user = participant_row
+
+    character_id = tool_input.get("character_id")
+    character: Character | None = None
+    if character_id:
+        character = await session.get(Character, uuid.UUID(character_id))
+        if character is None or character.owner_id != user.id:
+            raise ValueError(f"Character {character_id!r} does not belong to {player_name!r}")
+    elif participant.character_id is not None:
+        # The player already has a character in this game (set by an earlier call in this
+        # session) — reuse it instead of creating a duplicate when the model omits character_id.
+        character = await session.get(Character, participant.character_id)
+
+    if character is None:
+        character = Character(owner_id=user.id, name=tool_input.get("name", "Unbenannt"), stats={})
+        session.add(character)
+
+    for field in ("name", "race", "char_class", "description"):
+        if field in tool_input and tool_input[field] is not None:
+            setattr(character, field, tool_input[field])
+    if "stats" in tool_input and tool_input["stats"] is not None:
+        character.stats = {**(character.stats or {}), **tool_input["stats"]}
+
+    await session.flush()
+
+    if participant.character_id != character.id:
+        participant.character_id = character.id
+
+    await session.commit()
+    await session.refresh(character)
+
+    return {"character_id": str(character.id), "name": character.name}

backend/app/llm/tools/dice.py

@@ -0,0 +1,50 @@
+import re
+from random import SystemRandom
+
+_rng = SystemRandom()
+
+_NOTATION_RE = re.compile(r"^\s*(\d*)d(\d+)\s*(?:([+-])\s*(\d+))?\s*$", re.IGNORECASE)
+
+TOOL_SCHEMA = {
+    "name": "roll_dice",
+    "description": (
+        "Roll dice using standard D&D notation (e.g. '1d20+5', '2d6', 'd8-1'). "
+        "Always use this tool for any dice roll that affects the game — never invent or guess a result yourself."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "notation": {
+                "type": "string",
+                "description": "Dice notation, e.g. '1d20+5'",
+            },
+            "reason": {
+                "type": "string",
+                "description": "Short reason for the roll, e.g. 'Geschicklichkeitsrettung, Schwierigkeit 13'",
+            },
+        },
+        "required": ["notation"],
+    },
+}
+
+
+def roll(notation: str) -> dict:
+    match = _NOTATION_RE.match(notation)
+    if not match:
+        raise ValueError(f"Invalid dice notation: {notation!r}")
+
+    count_str, sides_str, sign, modifier_str = match.groups()
+    count = int(count_str) if count_str else 1
+    sides = int(sides_str)
+
+    if not (1 <= count <= 100) or not (2 <= sides <= 1000):
+        raise ValueError(f"Dice notation out of allowed range: {notation!r}")
+
+    modifier = int(modifier_str) if modifier_str else 0
+    if sign == "-":
+        modifier = -modifier
+
+    rolls = [_rng.randint(1, sides) for _ in range(count)]
+    total = sum(rolls) + modifier
+
+    return {"notation": notation, "rolls": rolls, "modifier": modifier, "total": total}