#!/usr/bin/env python3 """ engine.py — The Chaos Game Engine Owns the LLM interaction, prompt assembly, response parsing, and game state persistence. The TUI (run.py) calls this module — they do not depend on each other, only on the shared session/ file layout. """ from __future__ import annotations import json import re import sys from dataclasses import dataclass, field from datetime import date, datetime from pathlib import Path from string import Template from typing import Iterator, Optional # ── Paths ────────────────────────────────────────────────────────────────── BASE_DIR = Path(__file__).resolve().parent.parent SESSION_DIR = BASE_DIR / 'session' CONFIG_PATH = SESSION_DIR / 'config.json' CHAR_PATH = SESSION_DIR / 'character.md' WORLD_PATH = SESSION_DIR / 'world.md' BOOK_PATH = SESSION_DIR / 'book.md' JOURNAL_PATH = SESSION_DIR / 'journal.md' AMBIENCE_PATH = SESSION_DIR / 'ambience.md' LOG_DIR = SESSION_DIR / 'log' TODAY = date.today().isoformat() # ── Structured output ────────────────────────────────────────────────────── @dataclass class GenerationResult: narrative: str choices: list[str] = field(default_factory=list) log_entry: Optional[str] = None ambience: Optional[str] = None character_updates: Optional[str] = None world_updates: Optional[str] = None journal_add: list[str] = field(default_factory=list) journal_done: list[str] = field(default_factory=list) error: Optional[str] = None # ── DM System Prompt Template ────────────────────────────────────────────── SYSTEM_PROMPT = Template("""You are the Dungeon Master for **The Chaos**, a solo card-based rules-light fantasy TTRPG. Your job is to narrate an immersive, responsive story for one player character. ## Tone & Style - Write in **second person** ("You", "Dillion") — the player is Dillion. - Use vivid sensory descriptions — sight, sound, smell, touch. - Keep narration tight and cinematic. No monologues. - Use **bold** for emphasis, *italic* for thoughts/sounds. - NPC dialogue goes in **"quotes with bold names."** - Never present predefined choices — the player decides freely what to do. - Each turn should advance the story meaningfully. ## Game Rules (Quick Reference) ### Core Dice - **Odds**: 1d6, 4+ favours character, 3- is trouble. - **Traits**: 3d6, must roll UNDER the trait score. - **Combat hit**: 1d6 ± mods, 4+ hits. - **Damage**: 1d6 ± weapon mod - armour reduction. - **Initiative**: both sides roll 1d6, higher acts first. ### Combat Flow 1. Distance: 2d6 × 10 (metres/feet) 2. Surprise: 1d6 3. Grit: 2d6 for creatures (higher = more determined) 4. Initiative: 1d6 5. Turns: state intent → roll 1d6 ± mods → 4+ success, 3- take hit ### Wounds (0 HP) 1d6: 1-2 die, 3-4 lasting wound (-1 max HP), 5-6 -1 all rolls until healed ### Roll Modifiers Favourable +1, Risky -1, Desperate -2, Well-prepared +1, Poor visibility -1, Relevant trait +1 ### Exploration 6 ten-minute watches per hour. Each meaningful action advances a watch. ## Output Format IMPORTANT: End every response with a JSON fenced code block: ```json { "log_entry": "- **time of day** — brief description of what happened.", "ambience": "ambience_name_or_null", "character_updates": null, "world_updates": null, "journal_add": [], "journal_done": [] } ``` Rules for the JSON block: - **log_entry**: One-line log entry summarizing this turn's action. - **ambience**: One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds. Set to null to keep current. - **character_updates**: ONLY include if HP, cash, gear, or stats changed. Provide the FULL updated character sheet markdown. Otherwise null. - **world_updates**: ONLY include if NPCs, locations, or world state changed. Provide the FULL updated world markdown. Otherwise null. - **journal_add**: New TODO items to add. - **journal_done**: TODO items that are now completed. ## Available Tools You may use these tools before writing your final JSON block. Tool calls go in their own fenced code block: ```tool {"tool": "tool_name", "args": {...}} ``` You can call tools any number of times, one per block. Available tools: - **read_file** — Read a game state file. `{"file": "character|world|book|log|journal"}` - **roll** — Roll dice (outcome is shown to player). `{"dice": "2d6", "modifier": "-1"}` - **think** — Internal reasoning (shown in game status bar). `{"thought": "Your reasoning here"}` - **player_roll** — Ask the player to roll physical dice. **Only use when the player's action has an uncertain outcome that the dice should decide.** `{"dice": "2d6", "reason": "why"}` You may also show reasoning inline with a separate fence: ```thought Your reasoning here ``` Call tools as needed, then end with the final JSON block. When the player makes a choice, resolve it with the dice mechanics above. Describe the action, roll dice implicitly (describe the outcome, don't say "rolling dice"), apply damage/effects, and update state. ## Current Game State ### Character $character ### World $world ### Recent Log $log ### Recent Story (last turns from the book) $story""") # trailing """ is intentional — the template ends here # ── Game Engine ──────────────────────────────────────────────────────────── class GameEngine: """Owns the LLM interaction and game state persistence.""" def __init__(self, session_dir: str | Path = SESSION_DIR): self.session_dir = Path(session_dir) self.config: dict = {} self._load_config() # ── Config ────────────────────────────────────────────────────────── def _load_config(self) -> None: if not CONFIG_PATH.exists(): print( "No session/config.json found. Creating default.\n" "Edit the model field (e.g. 'ollama/llama3.1', 'openai/gpt-4', " "'anthropic/claude-sonnet-4-20250514') and set api_key if needed.", file=sys.stderr, ) self.config = { "llm": { "model": "ollama/llama3.1", "api_key": None, "api_base": None, "temperature": 0.8, } } self._save_config() else: raw = CONFIG_PATH.read_text() self.config = json.loads(raw) # Ensure api_key is None not empty string llm = self.config.get("llm", {}) if not llm.get("api_key"): llm["api_key"] = None def _save_config(self) -> None: CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) CONFIG_PATH.write_text(json.dumps(self.config, indent=2) + "\n") @property def model(self) -> str: return self.config.get("llm", {}).get("model", "ollama/llama3.1") @property def api_key(self) -> str | None: return self.config.get("llm", {}).get("api_key") @property def api_base(self) -> str | None: return self.config.get("llm", {}).get("api_base") @property def temperature(self) -> float: return self.config.get("llm", {}).get("temperature", 0.8) # ── Context Assembly ──────────────────────────────────────────────── def _read_file(self, path: Path) -> str: return path.read_text().strip() if path.exists() else "" def _read_recent_log(self, max_entries: int = 15) -> str: """Read the latest log file and return the last N entries.""" log_path = LOG_DIR / f"{TODAY}.md" if not log_path.exists(): # Check yesterday's log from datetime import timedelta yesterday = (date.today() - timedelta(days=1)).isoformat() log_path = LOG_DIR / f"{yesterday}.md" if not log_path.exists(): return "*No recent events.*" lines = log_path.read_text().splitlines() entries = [l for l in lines if l.strip().startswith("- ")] return "\n".join(entries[-max_entries:]) or "*No recent events.*" def _read_recent_book(self, max_turns: int = 3) -> str: """Return the last N turns from the book as context.""" text = self._read_file(BOOK_PATH) if not text: return "*No prior story.*" turns = text.split("\n## ") recent = turns[-max_turns:] return "\n## ".join(recent) if len(turns) > 1 else recent[0] def build_system_prompt(self) -> str: """Assemble the system prompt with current game state.""" char = self._read_file(CHAR_PATH) or "*No character sheet.*" world = self._read_file(WORLD_PATH) or "*No world state.*" log = self._read_recent_log() story = self._read_recent_book() return SYSTEM_PROMPT.substitute( character=char, world=world, log=log, story=story ) def build_user_message( self, player_action: str | None = None, last_narrative: str | None = None, ) -> str: """Build the user message for this turn's LLM call.""" parts = [] if last_narrative: parts.append(f"## Previously\n{last_narrative}") if player_action: parts.append(f"## Player Action\n{player_action}") has_existing_story = bool( self._read_file(BOOK_PATH).strip() ) if not last_narrative else True if not player_action and not last_narrative: if has_existing_story: parts.append( "## Instructions\n" "Continue the story from where it left off. Describe " "what happens next based on the current situation. " "Use tools as needed, then end with a JSON block." ) else: parts.append( "## Instructions\n" "Establish the opening scene. Dillion is at the " "Splintered Tankard in the Keep. Describe the " "setting and let the player decide what to do. " "Use tools as needed, then end with a JSON block." ) else: parts.append( "## Instructions\n" "Describe the outcome of the player's action using game " "mechanics where appropriate. Let the player decide their " "next move freely. Use tools as needed, then end with a " "JSON block." ) return "\n\n".join(parts) # ── LLM Call ──────────────────────────────────────────────────────── def generate( self, player_action: str | None = None, last_narrative: str | None = None, ) -> GenerationResult: """ Synchronous generation. Calls the LLM, parses the response, and returns a GenerationResult. The TUI calls this from a worker thread — see run.py. """ system = self.build_system_prompt() user = self.build_user_message( player_action=player_action, last_narrative=last_narrative ) messages = [ {"role": "system", "content": system}, {"role": "user", "content": user}, ] try: import litellm except ImportError: return GenerationResult( narrative="", error=( "litellm is not installed. Run: pip install litellm" ), ) # Set API key / base if provided if self.api_key: # litellm reads env vars or we can pass via kwargs os_env_key = f"{self.model.split('/')[0]}_API_KEY".upper() import os os.environ[os_env_key] = self.api_key if self.api_base: os_env_base = f"{self.model.split('/')[0]}_API_BASE".upper() import os os.environ[os_env_base] = self.api_base try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=False, ) text = response.choices[0].message.content or "" except Exception as e: return GenerationResult( narrative="", error=f"LLM call failed: {e}", ) return self.parse_response(text) def generate_stream( self, player_action: str | None = None, last_narrative: str | None = None, ) -> Iterator[str]: """ Streaming generator. Yields text chunks as they arrive from the LLM. On completion, the final yield is the FULL text (for parsing). """ system = self.build_system_prompt() user = self.build_user_message( player_action=player_action, last_narrative=last_narrative ) messages = [ {"role": "system", "content": system}, {"role": "user", "content": user}, ] try: import litellm except ImportError: yield json.dumps({ "error": "litellm is not installed. Run: pip install litellm" }) return if self.api_key: os_env_key = f"{self.model.split('/')[0]}_API_KEY".upper() import os os.environ[os_env_key] = self.api_key if self.api_base: os_env_base = f"{self.model.split('/')[0]}_API_BASE".upper() import os os.environ[os_env_base] = self.api_base try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=True, ) full_text = "" for chunk in response: delta = chunk.choices[0].delta.content or "" if delta: full_text += delta yield full_text # partial narrative for streaming display # Final yield is the completed text yield full_text except Exception as e: yield json.dumps({"error": f"LLM call failed: {e}"}) # ── Tool Infrastructure ──────────────────────────────────────────── TOOL_REGISTRY: dict[str, dict] = { "read_file": { "description": "Read a game state file.", "args": {"file": "character | world | book | log | journal"}, }, "roll": { "description": "Roll dice and return the outcome.", "args": {"dice": "e.g. 1d6, 2d6", "modifier": "optional +N or -N"}, }, "think": { "description": "Internal reasoning shown in the game status bar.", "args": {"thought": "Your reasoning."}, }, "player_roll": { "description": "Ask the player to physically roll dice and enter the result.", "args": {"dice": "e.g. 2d6+1", "reason": "Why the roll is needed (shown to player)"}, }, } def _tool_read_file(self, args: dict) -> str: filename = (args or {}).get("file", "") paths = { "character": CHAR_PATH, "world": WORLD_PATH, "book": BOOK_PATH, "log": LOG_DIR / f"{TODAY}.md", "journal": JOURNAL_PATH, } path = paths.get(filename) if not path: return f"Unknown file: {filename}. Choose from: {', '.join(paths)}" return self._read_file(path) or f"*{filename} is empty.*" def _tool_roll(self, args: dict) -> str: import random dice_str = (args or {}).get("dice", "1d6") modifier_str = (args or {}).get("modifier", "0") try: count, sides = dice_str.lower().split("d") count = int(count) if count else 1 sides = int(sides) except (ValueError, TypeError): return f"Invalid dice: {dice_str}. Use format like '2d6'." mod = 0 if modifier_str: try: mod = int(modifier_str) except ValueError: pass rolls = [random.randint(1, sides) for _ in range(count)] total = sum(rolls) + mod mod_str = f" {'+' if mod >= 0 else ''}{mod}" if mod != 0 else "" return f"Roll: {dice_str}{mod_str} → [{', '.join(str(r) for r in rolls)}] = {total}" @staticmethod def _describe_tool_action(tool_name: str, args: dict) -> str: """Return a user-facing status message for a tool call.""" read_descriptions = { "character": "reading the character sheet", "world": "consulting the world map", "book": "reviewing the story so far", "log": "checking the session log", "journal": "scanning the journal", } if tool_name == "read_file": file = (args or {}).get("file", "") desc = read_descriptions.get(file, f"reading {file}") elif tool_name == "roll": dice = (args or {}).get("dice", "1d6") mod = (args or {}).get("modifier") desc = f"rolling {dice}" if mod: desc += f" {mod}" elif tool_name == "player_roll": dice = (args or {}).get("dice", "1d6") desc = f"asking you to roll {dice}" elif tool_name == "think": desc = "pausing to think" else: desc = f"using {tool_name}" return f"DM is {desc}..." def _execute_tool(self, tool_name: str, args: dict) -> str: fn_map = { "read_file": self._tool_read_file, "roll": self._tool_roll, } fn = fn_map.get(tool_name) if not fn: return f"Unknown tool: {tool_name}" try: return fn(args) except Exception as e: return f"Tool error ({tool_name}): {e}" @staticmethod def _extract_thoughts(text: str) -> list[str]: pattern = r"```thought\s*\n?(.*?)```" return re.findall(pattern, text, re.DOTALL) @staticmethod def _extract_tool_calls(text: str) -> list[dict]: pattern = r"```tool\s*\n?(.*?)```" blocks = re.findall(pattern, text, re.DOTALL) calls = [] for block in blocks: try: parsed = json.loads(block.strip()) if isinstance(parsed, dict) and "tool" in parsed: calls.append(parsed) except json.JSONDecodeError: pass return calls @staticmethod def _extract_final_json(text: str) -> dict | None: pattern = r"```json\s*\n?(.*?)```" matches = re.findall(pattern, text, re.DOTALL) if not matches: return None try: return json.loads(matches[-1].strip()) except json.JSONDecodeError: return None def generate_with_tools( self, player_action: str | None = None, last_narrative: str | None = None, on_thought: callable = None, on_action: callable = None, on_player_roll: callable = None, ) -> GenerationResult: """ Multi-turn generation with tool-use loop. The LLM can output ```thought blocks (reasoning), ```tool blocks (tool calls), and a final ```json block. If any tool is called, the result is fed back and the LLM is re-invoked. The loop ends when the LLM outputs a ```json block with no tool calls. `on_thought` is called for each thought block (may be from a worker thread — use call_from_thread in the TUI). """ system = self.build_system_prompt() user = self.build_user_message( player_action=player_action, last_narrative=last_narrative, ) messages: list[dict] = [ {"role": "system", "content": system}, {"role": "user", "content": user}, ] try: import litellm except ImportError: return GenerationResult(narrative="", error="litellm not installed") if self.api_key: os_env_key = f"{self.model.split('/')[0]}_API_KEY".upper() import os os.environ[os_env_key] = self.api_key if self.api_base: os_env_base = f"{self.model.split('/')[0]}_API_BASE".upper() import os os.environ[os_env_base] = self.api_base max_rounds = 10 for round_idx in range(max_rounds): try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=False, ) text = response.choices[0].message.content or "" except Exception as e: return GenerationResult(narrative="", error=f"LLM call failed: {e}") thoughts = self._extract_thoughts(text) for t in thoughts: if on_thought: on_thought(t.strip()) tool_calls = self._extract_tool_calls(text) final_data = self._extract_final_json(text) if tool_calls: results = [] for tc in tool_calls: name = tc.get("tool", "?") args = tc.get("args", {}) if on_action: on_action(self._describe_tool_action(name, args)) if name == "player_roll" and on_player_roll: dice = args.get("dice", "1d6") reason = args.get("reason", "a check") roll_val = on_player_roll(dice, reason) result = f"Player rolled {dice} for '{reason}': {roll_val}" else: result = self._execute_tool(name, args) results.append(f"**Tool:** {name}\n**Args:** {json.dumps(args)}\n**Result:** {result}") messages.append({"role": "assistant", "content": text}) messages.append({ "role": "user", "content": "## Tool Results\n\n" + "\n\n".join(results), }) continue # Another round # No tool calls → parse the final JSON if final_data: return GenerationResult( narrative=text[: text.rfind("```json")].strip(), choices=final_data.get("choices", []), log_entry=final_data.get("log_entry"), ambience=final_data.get("ambience"), character_updates=final_data.get("character_updates"), world_updates=final_data.get("world_updates"), journal_add=final_data.get("journal_add", []), journal_done=final_data.get("journal_done", []), ) # Fallback: no tool calls, no JSON → normal parse return self.parse_response(text) return GenerationResult( narrative="", error="Tool loop exceeded max rounds (10)", ) # ── Response Parsing ──────────────────────────────────────────────── @staticmethod def parse_response(text: str) -> GenerationResult: """ Parse a full LLM response into a GenerationResult. Extracts the JSON block and splits narrative from it. """ # Check for error JSON if text.startswith('{"error":'): try: err = json.loads(text).get("error", "Unknown error") except json.JSONDecodeError: err = "Unknown error" return GenerationResult(narrative="", error=err) # Try to find a ```json ... ``` block json_pattern = r"```json\s*\n?(.*?)\n?```" matches = re.findall(json_pattern, text, re.DOTALL) narrative = text data = {} if matches: json_str = matches[-1].strip() narrative = text[: text.rfind("```json")] narrative = narrative.strip() try: data = json.loads(json_str) except json.JSONDecodeError: pass else: # Fallback: maybe the entire response is JSON (no fence) text_stripped = text.strip() if text_stripped.startswith("{") and text_stripped.endswith("}"): try: data = json.loads(text_stripped) narrative = data.get("narrative", "") except json.JSONDecodeError: pass return GenerationResult( narrative=narrative or text, choices=data.get("choices", []), log_entry=data.get("log_entry"), ambience=data.get("ambience"), character_updates=data.get("character_updates"), world_updates=data.get("world_updates"), journal_add=data.get("journal_add", []), journal_done=data.get("journal_done", []), ) # ── State Persistence ─────────────────────────────────────────────── def apply_state(self, result: GenerationResult) -> None: """Write state changes from a GenerationResult to disk.""" if result.character_updates: CHAR_PATH.write_text(result.character_updates.strip() + "\n") if result.world_updates: WORLD_PATH.write_text(result.world_updates.strip() + "\n") if result.log_entry: self.append_log(result.log_entry) if result.ambience: AMBIENCE_PATH.write_text(result.ambience.strip().lower() + "\n") if result.journal_add or result.journal_done: self._update_journal( add=result.journal_add, done=result.journal_done ) def archive_turn(self, narrative: str) -> None: """Append the narrative as a new turn in book.md.""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M") heading = f"\n\n## Turn — {timestamp}\n\n" BOOK_PATH.parent.mkdir(parents=True, exist_ok=True) with open(BOOK_PATH, "a") as f: f.write(heading + narrative.strip() + "\n") def append_log(self, entry: str) -> None: """Append a log entry to today's log file.""" LOG_DIR.mkdir(parents=True, exist_ok=True) log_path = LOG_DIR / f"{TODAY}.md" if not log_path.exists(): log_path.write_text(f"# Session Log — {TODAY}\n\n") with open(log_path, "a") as f: f.write(entry.strip() + "\n") def _update_journal( self, add: list[str] | None = None, done: list[str] | None = None ) -> None: """Add or complete TODO items in journal.md.""" if not JOURNAL_PATH.exists(): JOURNAL_PATH.write_text("# Journal\n\n## TODO\n\n## DONE\n\n") lines = JOURNAL_PATH.read_text().splitlines() new_lines = [] in_todo = False in_done = False for line in lines: stripped = line.strip() if stripped.startswith("## TODO"): in_todo = True in_done = False elif stripped.startswith("## DONE"): in_todo = False in_done = True new_lines.append(line) # Find insertion points todo_idx = None done_idx = None for i, line in enumerate(lines): stripped = line.strip() if stripped == "## TODO": todo_idx = i elif stripped == "## DONE": done_idx = i if done: for item in done: # Remove from TODO if present new_lines = [ l for l in new_lines if l.strip().lstrip("- ").lstrip("☐ ") != item ] # Find DONE section and add if done_idx is not None: done_entry = f"- {item}" if done_idx + 1 < len(new_lines): new_lines.insert(done_idx + 1, done_entry) else: new_lines.append(done_entry) if add: for item in add: entry = f"- {item}" if entry not in new_lines: if todo_idx is not None: new_lines.insert(todo_idx + 1, entry) else: new_lines.append(entry) JOURNAL_PATH.write_text("\n".join(new_lines) + "\n") # ── CLI entry point (for testing) ───────────────────────────────────────── def main(): """Generate a turn from the command line (debug/testing).""" import argparse parser = argparse.ArgumentParser(description="The Chaos Game Engine (CLI)") parser.add_argument("--action", "-a", help="Player action text") parser.add_argument("--last", "-l", help="Last narrative text") args = parser.parse_args() engine = GameEngine() result = engine.generate( player_action=args.action, last_narrative=args.last, ) if result.error: print(f"ERROR: {result.error}", file=sys.stderr) sys.exit(1) print(result.narrative) if result.choices: print("\n--- Choices ---") for c in result.choices: print(f" [{c}]") if result.log_entry: print(f"\n[Log] {result.log_entry}") if result.ambience: print(f"[Ambience] {result.ambience}") if __name__ == "__main__": main()