#!/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' LLM_LOG_PATH = SESSION_DIR / 'llm.log' AMBIENCE_OPTIONS_PATH = SESSION_DIR / "ambience_options.md" AUDIO_DIR = SESSION_DIR / "audio" TODAY = date.today().isoformat() # ── Structured output ────────────────────────────────────────────────────── @dataclass class GenerationResult: """Legacy result — kept for backward compat with CLI main().""" 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 @dataclass class TurnResult: """Output of a complete turn via finalize_turn tool.""" book_log: str = "" user_prompt: str = "" ambience: Optional[str] = None log_entry: Optional[str] = None error: Optional[str] = None debug_info: str = "" # ── 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. - **Stick to the player's intent.** Don't invent your own actions for the player unless forced by environment or circumstance (e.g., they trigger a trap, an NPC reacts, etc.). - **Keep turns short** — each turn covers a single action or brief exchange, not a full scene. Advance the story one step at a time. ## 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. ## How Turns Work Each turn follows this sequence: 1. The player's action or response is given to you. 2. Think, read files, roll dice, or ask the player to roll — any number of steps. 3. **You MUST call `finalize_turn` to end the turn.** There is no other way to complete a turn. The loop will keep calling you until you do. The **finalize_turn** tool produces all data for this turn: - **book_log** `[Required]` — **The complete self-contained narrative of this turn.** Describe what the player did (based on their action input) and what happened as a result, with all sensory/dialogue/mechanical details. This is the permanent story record — it must stand alone without the player's input text. The player's action is implicit in the narrative, not quoted. - **user_prompt** `[Required]` — **Short prompt for the player only, NOT recorded in the book.** Ask what they do next. 1-3 sentences. Do NOT recap the action — that belongs in `book_log`. - **log_entry** `[Optional]` — One-sentence summary of what happened (action + outcome). Keep it tight. - **ambience** `[Optional]` — One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds. ### How the Loop Works Each round the system reads your ````tool` blocks, executes them, and feeds back the results. This repeats until you call `finalize_turn`. If you call tools but never call `finalize_turn`, the loop runs until it hits the round limit and the turn fails with an error. So: call `finalize_turn` when the player needs to see the outcome and make their next decision. **Important: Do not mix get tools with finalize_turn.** If you call `read_file`, `character_get`, `world_get`, or `journal_get` in a round, you are still gathering information — do NOT also call `finalize_turn` in that same round. Gather first, then finalize in a separate round. ### Journal & Quest Tracking The journal is the player's quest log and TODO list. Use dedicated tools to manage it: - **`journal_get`** — Read the full journal to review quests. - **`journal_update`** — Add new quests/goals via `"add"` and mark completed via `"done"`. - **Add quests** as they arise: `{"add": ["Investigate the Weeper beneath the mill"]}` - **Mark sub-tasks** as they emerge: `{"add": ["Find a way to open the iron grate", "Question Rina about the cult"]}` - **Mark completed** when resolved: `{"done": ["Investigate the Weeper beneath the mill"]}` - **Keep descriptions specific** — vague entries like "Explore the dungeon" are not helpful. - **Review the journal** regularly to maintain continuity. - Long-term goals stay in TODO until resolved; don't re-add the same quest every turn. ### Character & World State To read or update state files, use the dedicated tools: - **`character_get`** / **`character_update`** — Read or replace the full character sheet. ONLY update when HP/cash/gear/stats change. - **`world_get`** / **`world_update`** — Read or replace the full world state. ONLY update when NPCs/locations/threads change. **IMPORTANT: `finalize_turn` is mandatory.** Every turn ends with `finalize_turn`. If you don't call it, the loop will keep feeding you tool results until it hits the round limit and the turn fails. See "How the Loop Works" above. ## Available Tools Tool calls go in their own fenced code block (one call per block): ```tool {"tool": "read_file", "args": {"file": "character", "dm_status": "Checking Dillion's stats."}} ``` You may also show reasoning inline: ```thought Your reasoning here ``` Tools available: Every tool call **must** include a `"dm_status"` string in `args` — a short, public-facing description of what the DM is doing (e.g. `"consulting the archives"`, `"examining the wound"`, `"calculating the odds"`). The player sees this in the UI. Keep it vague — never reveal what the DM is actually reading or learning. Tool reference (`[R]` = required, `[O]` = optional): - **read_file** — Read a game state file. `[R] file`: "character" | "world" | "book" | "log" | "journal" `[R] dm_status`: "..." - **roll** — Auto-roll dice (outcome shown in status). `[O] dice`: "2d6" (default "1d6") `[O] modifier`: "-1" (default "0") `[R] dm_status`: "..." - **player_roll** — Ask the player to roll physical dice. Use when the outcome is uncertain. `[O] dice`: "2d6" (default "1d6") `[O] reason`: "why the roll matters" `[R] dm_status`: "..." - **character_get** — Read the full character sheet. `[R] dm_status`: "..." - **character_update** — Replace the full character sheet. `[R] content`: "full character sheet markdown" `[R] dm_status`: "..." - **world_get** — Read the full world state. `[R] dm_status`: "..." - **world_update** — Replace the full world state. `[R] content`: "full world state markdown" `[R] dm_status`: "..." - **journal_get** — Read the journal (TODO / DONE). `[R] dm_status`: "..." - **journal_update** — Add or complete journal entries. `[O] add`: ["new todo item", ...] `[O] done`: ["completed item", ...] `[R] dm_status`: "..." - **finalize_turn** — **REQUIRED to end the turn.** The loop will NOT stop without it. Call this ALONE — do not mix with get tools. `[R] book_log`: "self-contained narrative of what the player did this turn — permanent story record, must stand alone" `[R] user_prompt`: "short prompt for the player — NOT recorded, 1-3 sentences" `[O] log_entry`: "one-sentence summary (action + outcome)" `[O] ambience`: "soundscape name: silence|calm|combat|dungeon|forest|tavern|tension|town|wilds" 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) def _set_llm_env(self) -> None: """Set provider-specific env vars for litellm.""" prefix = self.model.split("/")[0].upper() import os key = self.api_key or "sk-placeholder" os.environ[f"{prefix}_API_KEY"] = key if self.api_base: os.environ[f"{prefix}_API_BASE"] = self.api_base # ── 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 = 5) -> 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 = 1) -> 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 _get_valid_ambiences(self) -> set[str]: """Parse ambience_options.md and return set of valid ambience names with associated audio files.""" valid = {"silence"} # silence always valid (stops music) if not AMBIENCE_OPTIONS_PATH.exists(): return valid in_table = False for line in AMBIENCE_OPTIONS_PATH.read_text().splitlines(): s = line.strip() if not s.startswith("|") or not s.endswith("|"): in_table = False continue if in_table and all(c in "-:| " for c in s): continue parts = [p.strip() for p in s.split("|") if p.strip()] if not parts: continue if not in_table: in_table = True continue name = parts[0].lower() files_str = parts[1] if len(parts) > 1 else "" files = [f.strip() for f in files_str.split(",")] # Only add if at least one file exists (or is listed) has_files = any((AUDIO_DIR / f).exists() or f for f in files) if has_files: valid.add(name) return valid 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_prompt: str | None = None, **kwargs: str | None, ) -> str: """Build the user message for this turn's LLM call.""" if kwargs: raise TypeError( f"build_user_message() got unexpected keyword arguments: " f"{set(kwargs)}. Did you mean 'last_prompt' instead of one of these?" ) parts = [] if last_prompt: parts.append(f"## Situation\n{last_prompt}") if player_action: parts.append(f"## Player Action\n{player_action}") has_existing_story = bool( self._read_file(BOOK_PATH).strip() ) if not last_prompt else True if not player_action and not last_prompt: if has_existing_story: parts.append( "## Instructions\n" "Continue the story from where it left off. Think, " "gather information, then call finalize_turn.\n" "Put each tool call in its own ```tool block." ) else: parts.append( "## Instructions\n" "Establish the opening scene. Dillion is at the " "Splintered Tankard in the Keep. Describe the " "setting, then call finalize_turn.\n" "Put each tool call in its own ```tool block." ) else: parts.append( "## Instructions\n" "Describe the outcome of the player's action using game " "mechanics where appropriate. Think, gather information, " "then call finalize_turn to complete the turn.\n" "Put each tool call in its own ```tool 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_prompt=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 self._set_llm_env() try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=False, timeout=60, ) 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_prompt=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 self._set_llm_env() try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=True, timeout=60, ) 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)"}, }, "character_get": { "description": "Read the full character sheet.", "args": {}, }, "character_update": { "description": "Replace the character sheet with a new full version (ONLY if HP/cash/gear/stats changed).", "args": {"content": "Full character sheet markdown"}, }, "world_get": { "description": "Read the full world state.", "args": {}, }, "world_update": { "description": "Replace the world state with a new full version (ONLY if NPCs/locations/threads changed).", "args": {"content": "Full world state markdown"}, }, "journal_get": { "description": "Read the journal (TODO / DONE).", "args": {}, }, "journal_update": { "description": "Add or complete journal entries.", "args": {"add": "Optional: list of new TODO items", "done": "Optional: list of completed items"}, }, "finalize_turn": { "description": "Complete the turn.", "args": { "book_log": "[Required] Full narrative — appended to story book (permanent record)", "user_prompt": "[Required] Short prompt for player — NOT recorded, 1-3 sentences", "log_entry": "[Optional] One-sentence summary", "ambience": "[Optional] Soundscape name", }, }, } def _tool_think(self, args: dict) -> str: """Think tool — content is displayed via dm_status in the status bar.""" return "" 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}" def _tool_character_get(self, args: dict) -> str: return self._read_file(CHAR_PATH) or "*Character sheet is empty.*" def _tool_character_update(self, args: dict) -> str: content = (args or {}).get("content", "") if not content: return "**Error:** `content` is required." if not self._validate_update_size("character", content, CHAR_PATH): return "**Error:** Update rejected — content is too short (likely a partial paste)." CHAR_PATH.write_text(content.strip() + "\n") return "Character sheet updated." def _tool_world_get(self, args: dict) -> str: return self._read_file(WORLD_PATH) or "*World state is empty.*" def _tool_world_update(self, args: dict) -> str: content = (args or {}).get("content", "") if not content: return "**Error:** `content` is required." if not self._validate_update_size("world", content, WORLD_PATH): return "**Error:** Update rejected — content is too short (likely a partial paste)." WORLD_PATH.write_text(content.strip() + "\n") return "World state updated." def _tool_journal_get(self, args: dict) -> str: return self._read_file(JOURNAL_PATH) or "*Journal is empty.*" def _tool_journal_update(self, args: dict) -> str: add = (args or {}).get("add", []) done = (args or {}).get("done", []) if isinstance(add, str): add = [add] if isinstance(done, str): done = [done] if not add and not done: return "**Error:** Provide at least one of `add` or `done`." self._update_journal(add=add, done=done) return "Journal updated." @staticmethod def _describe_tool_action(tool_name: str, args: dict) -> str: """Return a user-facing status message for a tool call. Prefer the LLM-provided dm_status — otherwise fall back to a generic description.""" dm_status = (args or {}).get("dm_status") if dm_status: return f"DM is {dm_status}..." 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 in ("character_get", "world_get", "journal_get"): file = tool_name.replace("_get", "") desc = read_descriptions.get(file, f"reading {file}") elif tool_name in ("character_update", "world_update"): desc = "updating the records" elif tool_name == "journal_update": desc = "updating the journal" 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}" 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, "think": self._tool_think, "character_get": self._tool_character_get, "character_update": self._tool_character_update, "world_get": self._tool_world_get, "world_update": self._tool_world_update, "journal_get": self._tool_journal_get, "journal_update": self._tool_journal_update, } 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, *, round_num: int = 0, on_debug: callable = None) -> list[dict]: """Extract tool calls from ```tool and ```json blocks. Uses json.JSONDecoder.raw_decode for strict parsing; falls back to heuristics if the LLM produces unescaped newlines in string values. """ calls = [] seen = set() def _try_parse(raw: str) -> dict | None: try: obj = json.loads(raw) if isinstance(obj, dict) and "tool" in obj: return obj except json.JSONDecodeError: pass return None for m in re.finditer(r"```(?:tool|json|finalize_turn)\s*\n?", text): fence_type = m.group(0).strip("``` \n\r") # 1) Strict: raw_decode from where the JSON should start obj = None try: decoder = json.JSONDecoder() obj, end = decoder.raw_decode(text, m.end()) except (json.JSONDecodeError, ValueError, StopIteration): pass if obj is None: # 2) Fallback: find closing backticks and repair unescaped newlines in strings close = text.find("```", m.end()) if close > 0: raw = text[m.end():close].strip() def _escape_in_strings(s: str) -> str: return re.sub(r'"(?:[^"\\]|\\.)*"', lambda x: x.group(0).replace("\n", "\\n"), s, flags=re.DOTALL) repaired = _escape_in_strings(raw) obj = _try_parse(repaired) if obj is not None and isinstance(obj, dict): # Normalize: fence type "finalize_turn" means the JSON is the args directly if fence_type == "finalize_turn": obj = {"tool": "finalize_turn", "args": obj} # If JSON has a "tool" key, keep as-is if "tool" not in obj: obj = None if obj is not None: key = (obj["tool"], json.dumps(obj.get("args", {}), sort_keys=True)) if key not in seen: seen.add(key) calls.append(obj) elif on_debug: preview = text[m.end():m.end() + 120].replace("\n", "\\n") on_debug("parse_error", {"round": round_num, "content": preview}) 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_prompt: str | None = None, on_thought: callable = None, on_action: callable = None, on_player_roll: callable = None, on_debug: callable = None, ) -> TurnResult: """ Multi-turn generation with tool-use loop. The LLM can output ```thought blocks, call ```tool blocks, and MUST call **finalize_turn** to complete the turn. Until then the loop continues feeding tool results back. `on_thought` / `on_action` / `on_debug` may be called 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_prompt=last_prompt, ) messages: list[dict] = [ {"role": "system", "content": system}, {"role": "user", "content": user}, ] self._set_llm_env() try: import litellm except ImportError: return TurnResult(error="litellm not installed") max_rounds = 30 debug_entries: list[str] = [] attempt = 0 round_used = 0 reminder_count = 0 from datetime import datetime self._append_llm_log(f"\n{'='*60}") self._append_llm_log(f"=== Turn — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ===") self._append_llm_log(f"{'='*60}") if player_action: self._append_llm_log(f"Player: {player_action}") elif last_prompt: self._append_llm_log(f"Resume from: {last_prompt[:120]}") while round_used < max_rounds: attempt += 1 round_log: list[str] = [f"── Attempt {attempt} (round {round_used + 1}/{max_rounds}) ──"] try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=False, timeout=60, max_tokens=512, ) text = response.choices[0].message.content or "" self._append_llm_log( f"\n--- Attempt {attempt} ---\n{text}" ) except Exception as e: self._append_llm_log(f"\n--- LLM ERROR (attempt {attempt}) ---\n{e}") if on_debug: on_debug("llm_error", {"error": str(e)}) return TurnResult(error=f"LLM call failed: {e}") if on_debug: on_debug("llm_response", {"round": attempt, "text": text}) # Thoughts thoughts = self._extract_thoughts(text) if thoughts: round_log.append(f" thoughts: {len(thoughts)}") for t in thoughts: if on_thought: on_thought(t.strip()) if on_debug: on_debug("thought", {"round": attempt, "text": t.strip()}) # Tool calls tool_calls = self._extract_tool_calls( text, round_num=attempt, on_debug=on_debug, ) finalize_call: dict | None = None other_calls: list[dict] = [] for tc in tool_calls: if tc.get("tool") == "finalize_turn": finalize_call = tc else: other_calls.append(tc) # Log tool call summary if tool_calls: names = [tc.get("tool", "?") for tc in tool_calls] round_log.append(f" tools: {', '.join(names)}") # Guard: mixed get tools + finalize_turn → execute get tools, reject finalize get_tools = {"read_file", "character_get", "world_get", "journal_get"} if finalize_call and any(tc.get("tool") in get_tools for tc in other_calls): # Execute only the get tools, drop finalize_turn results = [] for tc in other_calls: if tc.get("tool") not in get_tools: continue name = tc.get("tool", "?") args = tc.get("args", {}) if not args.get("dm_status"): err_msg = ( f"**Validation Error:** Tool `{name}` missing required `dm_status`. " f"Add `\"dm_status\": \"what the DM is doing\"` to the args." ) results.append(err_msg) round_log.append(f" {name}: MISSING dm_status") if on_debug: on_debug("validation_error", {"round": attempt, "type": "tool", "tool": name, "error": "missing dm_status"}) continue if on_action: on_action(self._describe_tool_action(name, args)) if on_debug: on_debug("tool_call", {"round": attempt, "tool": name, "args": args}) result = self._execute_tool(name, args) results.append(f"**Tool:** {name}\n**Args:** {json.dumps(args)}\n**Result:** {result}") round_log.append(f" {name}: OK") if on_debug: on_debug("tool_result", {"round": attempt, "tool": name, "result": result}) round_log.append(" finalize_turn ignored (mixed with get tools)") debug_entries.append("\n".join(round_log)) messages = messages[:2] messages.append({"role": "assistant", "content": text}) messages.append({ "role": "user", "content": "## Tool Results\n\n" + "\n\n".join(results) + "\n\n**Note:** `finalize_turn` was ignored because you called get tools in the same round. Call `finalize_turn` alone in the next round to complete the turn." }) if on_debug: on_debug("validation_error", {"round": attempt, "type": "mixed_get_finalize", "tools": [tc.get("tool") for tc in other_calls]}) round_used += 1 continue # finalize_turn present → validate and return if finalize_call: args = finalize_call.get("args", {}) errs = [] if not args.get("book_log"): errs.append("book_log [Required]") if not args.get("user_prompt"): errs.append("user_prompt [Required]") # Validate ambience ambience_name = args.get("ambience") if ambience_name and ambience_name != "silence": valid_ambiences = self._get_valid_ambiences() if not valid_ambiences or ambience_name not in valid_ambiences: errs.append(f"ambience '{ambience_name}' is invalid or has no associated audio files.") if errs: hint = ( f"Expected:\n" f'{{"tool": "finalize_turn", "args": {{' f'"book_log": "...", ' f'"user_prompt": "...", ' f'"log_entry": "...", ' f'"ambience": "..."' f"}}}}\n" f"Valid ambiences: {', '.join(valid_ambiences)}" ) round_log.append(f" finalize_turn validation errors: {', '.join(errs)}") debug_entries.append("\n".join(round_log)) messages = messages[:2] messages.append({"role": "assistant", "content": text}) messages.append({ "role": "user", "content": f"## Validation Error\nMissing required field(s): {', '.join(errs)}.\n\n{hint}Please provide all required fields and call finalize_turn again." }) if on_debug: on_debug("validation_error", {"round": attempt, "type": "finalize_turn", "errors": errs}) round_used += 1 continue if on_debug: on_debug("finalize", {"round": attempt, "args": args}) round_used += 1 self._append_llm_log( f"\n--- FINALIZE (attempt {attempt}) ---\n" f"book_log: {args.get('book_log','')[:200]}\n" f"user_prompt: {args.get('user_prompt','')[:200]}\n" f"log_entry: {args.get('log_entry','')}\n" f"ambience: {args.get('ambience','')}\n" ) return TurnResult( book_log=args.get("book_log", ""), user_prompt=args.get("user_prompt", ""), ambience=args.get("ambience"), log_entry=args.get("log_entry"), ) # Execute other tools if other_calls: results = [] for tc in other_calls: name = tc.get("tool", "?") args = tc.get("args", {}) # dm_status is required on every tool call if not args.get("dm_status"): err_msg = ( f"**Validation Error:** Tool `{name}` missing required `dm_status`. " f"Add `\"dm_status\": \"what the DM is doing\"` to the args.\n" f"Put each tool call in its own ```tool block." ) results.append(err_msg) round_log.append(f" {name}: MISSING dm_status") if on_debug: on_debug("validation_error", {"round": attempt, "type": "tool", "tool": name, "error": "missing dm_status"}) continue if on_action: on_action(self._describe_tool_action(name, args)) if on_debug: on_debug("tool_call", {"round": attempt, "tool": name, "args": 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}") round_log.append(f" {name}: OK") if on_debug: on_debug("tool_result", {"round": attempt, "tool": name, "result": result}) messages = messages[:2] messages.append({"role": "assistant", "content": text}) messages.append({ "role": "user", "content": "## Tool Results\n\n" + "\n\n".join(results), }) debug_entries.append("\n".join(round_log)) round_used += 1 continue # No tools, no finalize round_log.append(" no tool calls") if not text.strip(): # Empty response — model may be slow. Give it time and retry without adding context. if on_debug: on_debug("empty_response", {"round": attempt}) import time time.sleep(2) debug_entries.append("\n".join(round_log)) continue # Plain-text reasoning (no ```tool/```thought blocks) — log in debug but don't show to player round_used += 1 if on_debug: on_debug("thought", {"round": attempt, "text": text.strip()}) debug_entries.append("\n".join(round_log)) messages = messages[:2] messages.append({"role": "assistant", "content": text}) reminder_count += 1 if reminder_count % 3 == 0: reminder = ( "## Instructions\n" "Respond with tool calls or finalize_turn.\n\n" "Put each tool call in its own ```tool block:\n" "```tool\n{\"tool\": \"character_get\", \"args\": {\"dm_status\": \"...\"}}\n```\n\n" "When ready, call **finalize_turn** with `book_log` and `user_prompt`." ) else: reminder = "Use tools to gather information or call **finalize_turn** to end the turn." messages.append({"role": "user", "content": reminder}) if on_debug: on_debug("no_tool_calls", {"round": attempt}) debug_text = "\n\n".join(debug_entries) self._append_llm_log(f"\n--- LOOP EXCEEDED ({max_rounds} rounds) ---\n{debug_text}") return TurnResult( error=f"Turn loop exceeded max rounds ({max_rounds}). Below is a debug log of what the LLM did each round:\n\n{debug_text}", debug_info=debug_text, ) # ── 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 _validate_update_size(self, name: str, new_content: str, path: Path) -> bool: """Reject updates that are more than 30% shorter than the existing file — likely the LLM pasted a fragment instead of the full state.""" if not path.exists(): return True old = path.read_text().strip() if not old: return True ratio = len(new_content) / len(old) if ratio < 0.7: import sys print( f"WARNING: {name} update rejected ({ratio:.0%} of original size " f"= {len(new_content)} vs {len(old)} chars) — likely a partial paste.", file=sys.stderr, ) return False return True def apply_state(self, result: TurnResult) -> None: """Write state changes from a TurnResult to disk.""" if result.ambience: AMBIENCE_PATH.write_text(result.ambience.strip().lower() + "\n") 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 _append_llm_log(self, text: str) -> None: """Append raw LLM activity to llm.log for debugging.""" LLM_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with open(LLM_LOG_PATH, "a") as f: f.write(text + "\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() # Parse into sections: everything before TODO, TODO items, between, DONE items, after todo_items: list[str] = [] done_items: list[str] = [] before_todo: list[str] = [] between: list[str] = [] after_done: list[str] = [] section = "before_todo" for line in lines: stripped = line.strip() if stripped == "## TODO": section = "todo" before_todo.append(line) elif stripped == "## DONE": section = "done" between.append(line) elif section == "before_todo": before_todo.append(line) elif section == "todo": if stripped.startswith("- "): todo_items.append(stripped[2:]) else: between.append(line) elif section == "done": if stripped.startswith("- "): done_items.append(stripped[2:]) else: after_done.append(line) # Apply changes if done: done_set = set(done) todo_items = [i for i in todo_items if i not in done_set] new_done = [i for i in done if i not in done_items] done_items.extend(new_done) if add: todo_set = set(todo_items) new_todo = [i for i in add if i not in todo_set] # Insert new items at the top of TODO todo_items = new_todo + todo_items # Reconstruct out = list(before_todo) for item in todo_items: out.append(f"- {item}") out.extend(between) for item in done_items: out.append(f"- {item}") out.extend(after_done) # Clean up: collapse multiple blank lines cleaned = [] prev_blank = False for line in out: is_blank = line.strip() == "" if is_blank and prev_blank: continue cleaned.append(line) prev_blank = is_blank # Ensure trailing newline JOURNAL_PATH.write_text("\n".join(cleaned) + "\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()