#!/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."** - Present **2-4 clear choices** at the end of each scene. - 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 { "choices": ["Choice 1", "Choice 2", "Choice 3"], "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: - **choices**: 2-4 brief action options presented to the player. - **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. 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 Events $log""") # 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() return SYSTEM_PROMPT.substitute(character=char, world=world, log=log) 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}") if not player_action and not last_narrative: parts.append( "## Instructions\n" "Establish the opening scene. Dillion is at the Splintered " "Tankard in the Keep. Describe the setting and present " "choices for what he might do. End with a JSON block." ) else: parts.append( "## Instructions\n" "Describe the outcome of the player's action using game " "mechanics where appropriate. Then present new choices. " "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}"}) # ── 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) # Extract JSON block — find the last ```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() # Remove the json block from the narrative narrative = text[: text.rfind("```json")] narrative = narrative.strip() try: data = json.loads(json_str) except json.JSONDecodeError: # Try to salvage partial JSON 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()